Data I/O & Processing¶

Shared I/O and spatial processing utilities for LAS/LAZ files, voxelization, LiDAR source discovery, and GeoPackage operations. Consumed by both the weak supervision pipeline and inference.

src.las_io ¶

LAS/LAZ file I/O via PDAL.

Shared read/write helpers used by inference, preprocessing, and evaluation.

read_las ¶

read_las(filepath: Union[str, Path]) -> Tuple[ndarray, Dict[str, Any]]

Read a LAS/LAZ file via PDAL.

Parameters:

Name	Type	Description	Default
`filepath`	`Union[str, Path]`	Path to input LAS/LAZ file.	required

Returns:

Type	Description
`Tuple[ndarray, Dict[str, Any]]`	Tuple of (arrays, metadata): arrays: Structured numpy array with all LAS fields (X, Y, Z, Intensity, Classification, etc.). metadata: PDAL pipeline metadata dict.

write_las ¶

write_las(output_path: Union[str, Path], arrays: ndarray, srs: str = 'EPSG:3857') -> None

Write a LAS/LAZ file via PDAL with standard options.

Preserves all extra dims and forward headers from the input arrays. Creates parent directories if they don't exist.

Parameters:

Name	Type	Description	Default
`output_path`	`Union[str, Path]`	Path to write the output file.	required
`arrays`	`ndarray`	Structured numpy array (e.g. from read_las, with Classification updated).	required
`srs`	`str`	Spatial reference system. Default: EPSG:3857.	`'EPSG:3857'`

normalize_intensity ¶

normalize_intensity(intensity: ndarray) -> ndarray

Normalize intensity values to 0-1 range.

Parameters:

Name	Type	Description	Default
`intensity`	`ndarray`	numpy array of intensity values.	required

Returns:

Type	Description
`ndarray`	Normalized array (0-1). Returns unchanged if max is 0.

src.voxelization ¶

Shared voxelization utilities for training and inference.

Converts raw point clouds into discrete voxel grids with aggregated features. Used by both the training data loader and inference pipeline.

VoxelResult `dataclass` ¶

Result of voxelizing a point cloud.

Attributes:

Name	Type	Description
`unique_coords`	`ndarray`	(M, 3) int32 unique voxel grid coordinates.
`voxel_features`	`ndarray`	(M, 1) float32 mean intensity per voxel.
`inverse_map`	`ndarray`	(N,) int array mapping each input point to its voxel index.
`voxel_labels`	`Optional[ndarray]`	(M,) int64 majority-vote label per voxel (training only, None at inference).

voxelize ¶

voxelize(xyz: ndarray, voxel_size: float, intensity: ndarray, labels: Optional[ndarray] = None) -> VoxelResult

Voxelize a point cloud with mean-intensity aggregation.

Quantizes xyz into a discrete voxel grid, deduplicates voxels, and computes mean intensity per voxel. When labels are provided (training), also computes majority-vote labels per voxel.

Parameters:

Name	Type	Description	Default
`xyz`	`ndarray`	(N, 3) point coordinates, should be shifted so min ~= 0.	required
`voxel_size`	`float`	Voxel edge length in meters (e.g. 0.1).	required
`intensity`	`ndarray`	(N,) or (N, 1) intensity values.	required
`labels`	`Optional[ndarray]`	Optional (N,) integer class labels for majority-vote aggregation.	`None`

Returns:

Type	Description
`VoxelResult`	VoxelResult with unique voxel coordinates, aggregated features,
`VoxelResult`	point-to-voxel inverse map, and optionally majority-vote labels.

src.lidar_utils ¶

Shared utilities for lidar source discovery, bridge file naming, and sampling.

Depends on geopandas. NOT imported by src/constants.py (which stays lightweight).

safe_source_name ¶

safe_source_name(source_name: str) -> str

Sanitize lidar source name for use in filenames.

Parameters:

Name	Type	Description	Default
`source_name`	`str`	Raw lidar source name (may contain slashes, colons, spaces).	required

Returns:

Type	Description
`str`	Filename-safe string with special characters replaced by underscores.

bridge_stem ¶

bridge_stem(osm_id: str, source_name: str) -> str

Construct bridge filename stem: bridge_{osm_id}_{safe_source}.

Parameters:

Name	Type	Description	Default
`osm_id`	`str`	OpenStreetMap bridge ID.	required
`source_name`	`str`	Raw lidar source name (will be sanitized).	required

Returns:

Type	Description
`str`	Canonical bridge filename stem.

parse_bridge_stem ¶

parse_bridge_stem(stem: str) -> Optional[Tuple[str, str]]

Parse bridge_{osm_id}_{source} into (osm_id, source) or None.

Parameters:

Name	Type	Description	Default
`stem`	`str`	Filename stem (without extension).	required

Returns:

Type	Description
`Optional[Tuple[str, str]]`	Tuple of (osm_id, source_name), or None if stem doesn't match the expected pattern.

load_lidar_index ¶

load_lidar_index(path: str, epsg: int = EPSG) -> GeoDataFrame

Load lidar_resources.geojson and reproject to the target CRS.

Parameters:

Name	Type	Description	Default
`path`	`str`	Path to the GeoJSON file containing USGS 3DEP EPT source geometries.	required
`epsg`	`int`	Target EPSG code for reprojection. Default: 3857 (Web Mercator).	`EPSG`

Returns:

Type	Description
`GeoDataFrame`	GeoDataFrame of lidar sources reprojected to the target CRS.

find_intersecting_sources ¶

find_intersecting_sources(lidar_gdf: GeoDataFrame, bridge_geometry: Any, buffer_meters: float = DEFAULT_BUFFER) -> List[Dict[str, str]]

Find lidar sources whose footprints intersect the buffered bridge geometry.

Parameters:

Name	Type	Description	Default
`lidar_gdf`	`GeoDataFrame`	GeoDataFrame of lidar sources (from load_lidar_index).	required
`bridge_geometry`	`Any`	Bridge geometry in EPSG:3857 (same CRS as lidar_gdf).	required
`buffer_meters`	`float`	Buffer distance around the bridge geometry. Default: 10.0.	`DEFAULT_BUFFER`

Returns:

Type	Description
`List[Dict[str, str]]`	List of dicts with 'url' and 'name' keys for each intersecting source.

stratified_sample ¶

stratified_sample(entries: List[Dict[str, Any]], sample_size: int, max_per_group: int, seed: int, group_key: str = 'huc_id', id_key: str = 'osm_id') -> List[Dict[str, Any]]

Round-robin sample across groups, deduplicating by id_key within each group.

Parameters:

Name	Type	Description	Default
`entries`	`List[Dict[str, Any]]`	List of dicts, each with at least group_key and id_key fields.	required
`sample_size`	`int`	Target number of entries to return.	required
`max_per_group`	`int`	Maximum entries from any single group.	required
`seed`	`int`	Random seed for reproducibility.	required
`group_key`	`str`	Dict key to group by. Default: 'huc_id'.	`'huc_id'`
`id_key`	`str`	Dict key for deduplication within groups. Default: 'osm_id'.	`'osm_id'`

Returns:

Type	Description
`List[Dict[str, Any]]`	Sampled list of entries, sorted by (group_key, id_key).

src.gpkg_utils ¶

GeoPackage I/O utilities for bridge data.

Shared read, write, split, and iteration helpers used across the weak supervision pipeline, inference preparation, and analysis tools.

read_bridge_gpkg ¶

read_bridge_gpkg(path: Union[str, Path], required_cols: Sequence[str] = ('osmid',), target_epsg: int = 3857) -> GeoDataFrame

Read a bridge GeoPackage, validate columns, reproject, and cast osmid to str.

Parameters:

Name	Type	Description	Default
`path`	`Union[str, Path]`	Path to the .gpkg file.	required
`required_cols`	`Sequence[str]`	Columns that must be present. Raises ValueError if any are missing.	`('osmid',)`
`target_epsg`	`int`	Target CRS EPSG code. Default: 3857 (Web Mercator).	`3857`

Returns:

Type	Description
`GeoDataFrame`	GeoDataFrame reprojected to target_epsg with osmid as str (if present).

write_gpkg ¶

write_gpkg(gdf: GeoDataFrame, path: Union[str, Path]) -> None

Write a GeoDataFrame to a GeoPackage file, creating parent directories.

split_gpkg_by_column ¶

split_gpkg_by_column(gdf: GeoDataFrame, column: str, output_dir: Union[str, Path], filename_template: str = DEFAULT_GPKG_TEMPLATE, verbose: bool = False) -> Dict[str, Path]

Split a GeoDataFrame into per-value GeoPackage files.

Groups by column, writes one GPKG per group into {output_dir}/{value}/{filename_template} where {huc_id} in the template is replaced by the group value.

Parameters:

Name	Type	Description	Default
`gdf`	`GeoDataFrame`	Input GeoDataFrame.	required
`column`	`str`	Column to group by.	required
`output_dir`	`Union[str, Path]`	Root output directory.	required
`filename_template`	`str`	Filename with `{huc_id}` placeholder.	`DEFAULT_GPKG_TEMPLATE`
`verbose`	`bool`	Print progress every 200 groups.	`False`

Returns:

Type	Description
`Dict[str, Path]`	Dict mapping group values to written file paths.

iter_huc_gpkgs ¶

iter_huc_gpkgs(hucs_dir: Union[str, Path], filename_template: str = DEFAULT_GPKG_TEMPLATE, huc_ids: Optional[Sequence[str]] = None) -> Iterator[Tuple[str, Path]]

Yield (huc_id, gpkg_path) for each HUC directory containing a matching GPKG.

Iterates subdirectories of hucs_dir in sorted order. Each subdirectory name is treated as a HUC ID. Directories without a matching GPKG file are silently skipped.

Parameters:

Name	Type	Description	Default
`hucs_dir`	`Union[str, Path]`	Root directory containing per-HUC subdirectories.	required
`filename_template`	`str`	GPKG filename with `{huc_id}` placeholder.	`DEFAULT_GPKG_TEMPLATE`
`huc_ids`	`Optional[Sequence[str]]`	Optional allowlist of HUC IDs to include.	`None`

Yields:

Type	Description
`Tuple[str, Path]`	(huc_id, gpkg_path) tuples.

filter_by_ids ¶

filter_by_ids(gdf: GeoDataFrame, column: str, ids: Sequence) -> GeoDataFrame

Filter a GeoDataFrame to rows where column value is in ids.

Handles mixed types by casting both the column and the filter values to strings before comparison.

Parameters:

Name	Type	Description	Default
`gdf`	`GeoDataFrame`	Input GeoDataFrame.	required
`column`	`str`	Column to filter on.	required
`ids`	`Sequence`	Values to keep (will be cast to str).	required

Returns:

Type	Description
`GeoDataFrame`	Filtered GeoDataFrame.

Data I/O & Processing¶

src.las_io ¶

read_las ¶

write_las ¶

normalize_intensity ¶

src.voxelization ¶

VoxelResult dataclass ¶

voxelize ¶

src.lidar_utils ¶

safe_source_name ¶

bridge_stem ¶

parse_bridge_stem ¶

load_lidar_index ¶

find_intersecting_sources ¶

stratified_sample ¶

src.gpkg_utils ¶

read_bridge_gpkg ¶

write_gpkg ¶

split_gpkg_by_column ¶

iter_huc_gpkgs ¶

filter_by_ids ¶

VoxelResult `dataclass` ¶