API Reference

The supported public contract is now subpackage-first:

• nyc_mesh.models
• nyc_mesh.io
• nyc_mesh.analysis
• nyc_mesh.export
• nyc_mesh.pipeline
• nyc_mesh.cli

The root nyc_mesh namespace is intentionally minimal and only exposes __version__.

Root

nyc_mesh

Minimal root namespace for the nyc-mesh package.

Models

nyc_mesh.models

Public typed models for nyc-mesh.

Coordinate2D module-attribute

Coordinate2D = tuple[float, float]

Coordinate3D module-attribute

Coordinate3D = tuple[float, float, float]

ScalarValue module-attribute

ScalarValue = str | int | float | None

BoundingBox dataclass

Geographic clipping bounds in WGS84 order.

Source code in src/nyc_mesh/models/_core.py

@dataclass(frozen=True, slots=True)
class BoundingBox:
    """Geographic clipping bounds in WGS84 order."""

    min_lat: float
    min_lon: float
    max_lat: float
    max_lon: float

    def __post_init__(self) -> None:
        if self.min_lat >= self.max_lat:
            message = "BoundingBox.min_lat must be smaller than max_lat."
            raise ValueError(message)
        if self.min_lon >= self.max_lon:
            message = "BoundingBox.min_lon must be smaller than max_lon."
            raise ValueError(message)

    def contains(self, longitude: float, latitude: float) -> bool:
        """Return whether the WGS84 point falls inside the bounding box."""

        return (
            self.min_lon <= longitude <= self.max_lon
            and self.min_lat <= latitude <= self.max_lat
        )

min_lat instance-attribute

min_lat: float

min_lon instance-attribute

min_lon: float

max_lat instance-attribute

max_lat: float

max_lon instance-attribute

max_lon: float

contains

contains(longitude: float, latitude: float) -> bool

Return whether the WGS84 point falls inside the bounding box.

Source code in src/nyc_mesh/models/_core.py

def contains(self, longitude: float, latitude: float) -> bool:
    """Return whether the WGS84 point falls inside the bounding box."""

    return (
        self.min_lon <= longitude <= self.max_lon
        and self.min_lat <= latitude <= self.max_lat
    )

BuildingFeature dataclass

Height-aware building feature reprojected to WGS84 coordinates.

Source code in src/nyc_mesh/models/_core.py

@dataclass(frozen=True, slots=True)
class BuildingFeature:
    """Height-aware building feature reprojected to WGS84 coordinates."""

    building_id: str
    footprint_4326: tuple[Coordinate2D, ...]
    height: float
    properties: dict[str, ScalarValue] = field(default_factory=dict)

    @property
    def centroid(self) -> Coordinate2D:
        """Return the simple polygon-centroid average."""

        ring = self.footprint_4326[:-1] if self.footprint_4326 else ()
        if not ring:
            return (0.0, 0.0)
        return (
            sum(point[0] for point in ring) / len(ring),
            sum(point[1] for point in ring) / len(ring),
        )

    def with_properties(self, updates: dict[str, ScalarValue]) -> BuildingFeature:
        """Return a copy with merged properties."""

        merged = {**self.properties, **updates}
        return replace(self, properties=merged)

building_id instance-attribute

building_id: str

footprint_4326 instance-attribute

footprint_4326: tuple[Coordinate2D, ...]

height instance-attribute

height: float

properties class-attribute instance-attribute

properties: dict[str, ScalarValue] = field(
    default_factory=dict
)

centroid property

centroid: Coordinate2D

Return the simple polygon-centroid average.

with_properties

with_properties(
    updates: dict[str, ScalarValue],
) -> BuildingFeature

Return a copy with merged properties.

Source code in src/nyc_mesh/models/_core.py

def with_properties(self, updates: dict[str, ScalarValue]) -> BuildingFeature:
    """Return a copy with merged properties."""

    merged = {**self.properties, **updates}
    return replace(self, properties=merged)

CityGMLBuilding dataclass

Single CityGML building with an EPSG:2263 footprint and optional height.

Source code in src/nyc_mesh/models/_core.py

@dataclass(frozen=True, slots=True)
class CityGMLBuilding:
    """Single CityGML building with an EPSG:2263 footprint and optional height."""

    building_id: str
    footprint_2263: tuple[Coordinate2D, ...]
    measured_height: float | None

building_id instance-attribute

building_id: str

footprint_2263 instance-attribute

footprint_2263: tuple[Coordinate2D, ...]

measured_height instance-attribute

measured_height: float | None

CityGMLDataset dataclass

Loaded local CityGML source and parsed building records.

Source code in src/nyc_mesh/models/_core.py

@dataclass(frozen=True, slots=True)
class CityGMLDataset:
    """Loaded local CityGML source and parsed building records."""

    source: Path
    buildings: tuple[CityGMLBuilding, ...]

source instance-attribute

source: Path

buildings instance-attribute

buildings: tuple[CityGMLBuilding, ...]

DataSourceMetadata dataclass

Metadata describing one cached or user-staged public dataset.

Source code in src/nyc_mesh/models/_core.py

@dataclass(frozen=True, slots=True)
class DataSourceMetadata:
    """Metadata describing one cached or user-staged public dataset."""

    name: str
    source_url: str
    cache_path: Path
    refreshed_at: datetime
    record_count: int
    notes: str = ""

name instance-attribute

name: str

source_url instance-attribute

source_url: str

cache_path instance-attribute

cache_path: Path

refreshed_at instance-attribute

refreshed_at: datetime

record_count instance-attribute

record_count: int

notes class-attribute instance-attribute

notes: str = ''

DEMDataset dataclass

Loaded digital elevation model.

Source code in src/nyc_mesh/models/_core.py

@dataclass(frozen=True, slots=True)
class DEMDataset:
    """Loaded digital elevation model."""

    source: Path
    values: tuple[tuple[float | None, ...], ...]
    origin_x: float
    origin_y: float
    cell_size: float
    nodata: float | None
    crs: str

    @property
    def rows(self) -> int:
        return len(self.values)

    @property
    def cols(self) -> int:
        return len(self.values[0]) if self.values else 0

source instance-attribute

source: Path

values instance-attribute

values: tuple[tuple[float | None, ...], ...]

origin_x instance-attribute

origin_x: float

origin_y instance-attribute

origin_y: float

cell_size instance-attribute

cell_size: float

nodata instance-attribute

nodata: float | None

crs instance-attribute

crs: str

rows property

rows: int

cols property

cols: int

ExportTarget dataclass

Destination metadata for export commands.

Source code in src/nyc_mesh/models/_core.py

@dataclass(frozen=True, slots=True)
class ExportTarget:
    """Destination metadata for export commands."""

    format: str
    output_path: Path

format instance-attribute

format: str

output_path instance-attribute

output_path: Path

FootprintDataset dataclass

Loaded footprint features used for PLUTO-style joins.

Source code in src/nyc_mesh/models/_core.py

@dataclass(frozen=True, slots=True)
class FootprintDataset:
    """Loaded footprint features used for PLUTO-style joins."""

    source: Path
    features: tuple[FootprintFeature, ...]

source instance-attribute

source: Path

features instance-attribute

features: tuple[FootprintFeature, ...]

FootprintFeature dataclass

Generic polygon feature loaded from GeoJSON or shapefile footprints.

Source code in src/nyc_mesh/models/_core.py

@dataclass(frozen=True, slots=True)
class FootprintFeature:
    """Generic polygon feature loaded from GeoJSON or shapefile footprints."""

    feature_id: str
    footprint_4326: tuple[Coordinate2D, ...]
    properties: dict[str, ScalarValue] = field(default_factory=dict)

    @property
    def centroid(self) -> Coordinate2D:
        ring = self.footprint_4326[:-1] if self.footprint_4326 else ()
        if not ring:
            return (0.0, 0.0)
        return (
            sum(point[0] for point in ring) / len(ring),
            sum(point[1] for point in ring) / len(ring),
        )

feature_id instance-attribute

feature_id: str

footprint_4326 instance-attribute

footprint_4326: tuple[Coordinate2D, ...]

properties class-attribute instance-attribute

properties: dict[str, ScalarValue] = field(
    default_factory=dict
)

centroid property

centroid: Coordinate2D

LidarDataset dataclass

Loaded LiDAR point cloud.

Source code in src/nyc_mesh/models/_core.py

@dataclass(frozen=True, slots=True)
class LidarDataset:
    """Loaded LiDAR point cloud."""

    source: Path
    points: tuple[LidarPoint, ...]

source instance-attribute

source: Path

points instance-attribute

points: tuple[LidarPoint, ...]

LidarPoint dataclass

Single LiDAR point.

Source code in src/nyc_mesh/models/_core.py

@dataclass(frozen=True, slots=True)
class LidarPoint:
    """Single LiDAR point."""

    x: float
    y: float
    z: float
    intensity: int | None = None

x instance-attribute

x: float

y instance-attribute

y: float

z instance-attribute

z: float

intensity class-attribute instance-attribute

intensity: int | None = None

NeighborhoodRequest dataclass

A named extraction request for a neighborhood-scale clip.

Source code in src/nyc_mesh/models/_core.py

@dataclass(frozen=True, slots=True)
class NeighborhoodRequest:
    """A named extraction request for a neighborhood-scale clip."""

    name: str
    bbox: BoundingBox

name instance-attribute

name: str

bbox instance-attribute

bbox: BoundingBox

StudyAreaAssetManifest dataclass

Manifest of the official assets used for one mesh study area.

Source code in src/nyc_mesh/models/_core.py

@dataclass(frozen=True, slots=True)
class StudyAreaAssetManifest:
    """Manifest of the official assets used for one mesh study area."""

    study_area_name: str
    bbox: BoundingBox
    cache_dir: Path
    citygml_source: Path
    metadata: tuple[DataSourceMetadata, ...]
    footprints_source: Path | None = None
    pluto_source: Path | None = None
    dem_source: Path | None = None
    lidar_source: Path | None = None

study_area_name instance-attribute

study_area_name: str

bbox instance-attribute

bbox: BoundingBox

cache_dir instance-attribute

cache_dir: Path

citygml_source instance-attribute

citygml_source: Path

metadata instance-attribute

metadata: tuple[DataSourceMetadata, ...]

footprints_source class-attribute instance-attribute

footprints_source: Path | None = None

pluto_source class-attribute instance-attribute

pluto_source: Path | None = None

dem_source class-attribute instance-attribute

dem_source: Path | None = None

lidar_source class-attribute instance-attribute

lidar_source: Path | None = None

TerrainMeshDataset dataclass

Triangulated terrain mesh.

Source code in src/nyc_mesh/models/_core.py

@dataclass(frozen=True, slots=True)
class TerrainMeshDataset:
    """Triangulated terrain mesh."""

    source: str
    vertices: tuple[Coordinate3D, ...]
    triangles: tuple[tuple[int, int, int], ...]

source instance-attribute

source: str

vertices instance-attribute

vertices: tuple[Coordinate3D, ...]

triangles instance-attribute

triangles: tuple[tuple[int, int, int], ...]

IO

nyc_mesh.io

Public loaders for nyc-mesh.

NYC_BUILDING_FOOTPRINTS_API_URL module-attribute

NYC_BUILDING_FOOTPRINTS_API_URL = (
    "https://data.cityofnewyork.us/resource/5zhs-2jue.json"
)

NYC_CITYGML_ARCHIVE_URL module-attribute

NYC_CITYGML_ARCHIVE_URL = (
    "https://maps.nyc.gov/download/3dmodel/DA_WISE_GML.zip"
)

NYC_DEM_DATASET_URL module-attribute

NYC_DEM_DATASET_URL = "https://data.cityofnewyork.us/City-Government/1-foot-Digital-Elevation-Model-DEM-/dpc8-z3jc/data?no_mobile=true"

NYC_LIDAR_DATASET_URL module-attribute

NYC_LIDAR_DATASET_URL = "https://noaa-nos-coastal-lidar-pds.s3.amazonaws.com/laz/geoid18/9306/index.html"

NYC_PLUTO_API_URL module-attribute

NYC_PLUTO_API_URL = (
    "https://data.cityofnewyork.us/resource/64uk-42ks.json"
)

load_citygml

load_citygml(source: str | Path) -> CityGMLDataset

Load building footprints and measured heights from a local CityGML file.

Source code in src/nyc_mesh/io/_core.py

def load_citygml(source: str | Path) -> CityGMLDataset:
    """Load building footprints and measured heights from a local CityGML file."""

    if isinstance(source, str) and source.startswith(("http://", "https://")):
        message = "load_citygml() supports local file paths only."
        raise ValueError(message)

    source_path = Path(source).expanduser()
    if not source_path.exists():
        message = f"CityGML source does not exist: {source_path}"
        raise FileNotFoundError(message)
    if not source_path.is_file():
        message = f"CityGML source must be a file path: {source_path}"
        raise ValueError(message)
    parser = etree.XMLParser(
        resolve_entities=False,
        no_network=True,
        remove_comments=True,
    )
    if source_path.suffix.lower() == ".zip":
        _, member_bytes = _zip_members(source_path, suffixes=(".gml", ".xml"))
        root = etree.fromstring(member_bytes, parser=parser)
    else:
        root = etree.parse(str(source_path), parser=parser).getroot()

    buildings: list[CityGMLBuilding] = []
    for building_node in cast(
        "list[etree._Element]",
        root.xpath(".//bldg:Building", namespaces=_NS),
    ):
        footprint_2263: tuple[Coordinate2D, ...] = ()
        for polygon in cast(
            "list[etree._Element]",
            building_node.xpath(".//gml:Polygon", namespaces=_NS),
        ):
            footprint_2263 = _extract_exterior_ring(polygon)
            if footprint_2263:
                break
        if not footprint_2263:
            continue
        buildings.append(
            CityGMLBuilding(
                building_id=building_node.get(_GML_ID)
                or f"building-{len(buildings) + 1}",
                footprint_2263=footprint_2263,
                measured_height=_extract_measured_height(building_node),
            )
        )
    return CityGMLDataset(source=source_path.resolve(), buildings=tuple(buildings))

load_dem

load_dem(source: str | Path) -> DEMDataset

Load a DEM from ESRI ASCII grid or JSON grid data.

Source code in src/nyc_mesh/io/_core.py

def load_dem(source: str | Path) -> DEMDataset:
    """Load a DEM from ESRI ASCII grid or JSON grid data."""

    source_path = Path(source).expanduser()
    if not source_path.exists():
        message = f"DEM source does not exist: {source_path}"
        raise FileNotFoundError(message)

    suffix = source_path.suffix.lower()
    if suffix == ".zip":
        member_name, member_bytes = _zip_members(
            source_path, suffixes=(".tif", ".tiff", ".asc")
        )
        member_suffix = Path(member_name).suffix.lower()
        if member_suffix in {".tif", ".tiff"}:
            if rasterio is None or MemoryFile is None:
                message = (
                    "GeoTIFF DEM loading requires the optional `rasterio` dependency."
                )
                raise RuntimeError(message)
            with MemoryFile(member_bytes) as memory_file, memory_file.open() as dataset:
                return _dem_from_rasterio(dataset, source_path)
        return _dem_from_ascii_text(member_bytes.decode("utf-8"), source_path)

    if suffix in {".tif", ".tiff"}:
        if rasterio is None:
            message = "GeoTIFF DEM loading requires the optional `rasterio` dependency."
            raise RuntimeError(message)
        with rasterio.open(source_path) as dataset:
            return _dem_from_rasterio(dataset, source_path)
    if suffix == ".json":
        payload = json.loads(source_path.read_text(encoding="utf-8"))
        values = tuple(
            tuple(None if value is None else float(value) for value in row)
            for row in payload["values"]
        )
        return DEMDataset(
            source=source_path.resolve(),
            values=values,
            origin_x=float(payload["origin_x"]),
            origin_y=float(payload["origin_y"]),
            cell_size=float(payload["cell_size"]),
            nodata=None if payload.get("nodata") is None else float(payload["nodata"]),
            crs=str(payload.get("crs", "EPSG:2263")),
        )

    if suffix != ".asc":
        message = f"Unsupported DEM format: {source_path.suffix}"
        raise ValueError(message)
    return _dem_from_ascii_text(source_path.read_text(encoding="utf-8"), source_path)

load_footprints

load_footprints(source: str | Path) -> FootprintDataset

Load footprint polygons from GeoJSON or ESRI shapefile sources.

Source code in src/nyc_mesh/io/_core.py

def load_footprints(source: str | Path) -> FootprintDataset:
    """Load footprint polygons from GeoJSON or ESRI shapefile sources."""

    source_path = Path(source).expanduser()
    if not source_path.exists():
        message = f"Footprint source does not exist: {source_path}"
        raise FileNotFoundError(message)

    suffix = source_path.suffix.lower()
    if suffix in {".geojson", ".json"}:
        payload = json.loads(source_path.read_text(encoding="utf-8"))
        features = payload.get("features")
        if not isinstance(features, list):
            message = f"{source_path} must contain a GeoJSON FeatureCollection."
            raise TypeError(message)
        loaded = []
        for index, feature in enumerate(features, start=1):
            if not isinstance(feature, dict):
                continue
            geometry = feature.get("geometry")
            properties = feature.get("properties", {})
            if not isinstance(geometry, dict):
                continue
            ring = _coerce_geojson_footprint(geometry)
            if not ring:
                continue
            feature_id = str(
                properties.get("building_id")
                or properties.get("bbl")
                or feature.get("id")
                or f"feature-{index}"
            )
            loaded.append(
                FootprintFeature(
                    feature_id=feature_id,
                    footprint_4326=ring,
                    properties={
                        str(key): value
                        for key, value in properties.items()
                        if isinstance(value, (str, int, float)) or value is None
                    },
                )
            )
        return FootprintDataset(source=source_path.resolve(), features=tuple(loaded))

    if suffix == ".shp":
        if shapefile is None:
            message = "Shapefile loading requires the optional `pyshp` dependency."
            raise RuntimeError(message)
        reader = shapefile.Reader(str(source_path))
        field_names = [
            field[0] for field in reader.fields if field[0] != "DeletionFlag"
        ]
        loaded = []
        for index, shape_record in enumerate(reader.iterShapeRecords(), start=1):
            if shape_record.shape.shapeTypeName not in {"POLYGON", "POLYGONZ"}:
                continue
            properties = {
                key: value
                for key, value in zip(field_names, shape_record.record, strict=False)
                if isinstance(value, (str, int, float)) or value is None
            }
            ring = normalise_ring(
                tuple(
                    (float(point[0]), float(point[1]))
                    for point in shape_record.shape.points
                )
            )
            if not ring:
                continue
            feature_id = str(
                properties.get("building_id")
                or properties.get("bbl")
                or f"feature-{index}"
            )
            loaded.append(
                FootprintFeature(
                    feature_id=feature_id,
                    footprint_4326=ring,
                    properties=properties,
                )
            )
        return FootprintDataset(source=source_path.resolve(), features=tuple(loaded))

    message = f"Unsupported footprint format: {source_path.suffix}"
    raise ValueError(message)

load_lidar

load_lidar(source: str | Path) -> LidarDataset

Load LiDAR point data from LAS/LAZ, CSV, XYZ, or JSON inputs.

Source code in src/nyc_mesh/io/_core.py

def load_lidar(source: str | Path) -> LidarDataset:
    """Load LiDAR point data from LAS/LAZ, CSV, XYZ, or JSON inputs."""

    source_path = Path(source).expanduser()
    if not source_path.exists():
        message = f"LiDAR source does not exist: {source_path}"
        raise FileNotFoundError(message)

    suffix = source_path.suffix.lower()
    if suffix == ".zip":
        if laspy is None:
            message = (
                "ZIP-wrapped LAS/LAZ loading requires the optional `laspy` dependency."
            )
            raise RuntimeError(message)
        _, member_bytes = _zip_members(source_path, suffixes=(".las", ".laz"))
        las = laspy.read(BytesIO(member_bytes))
        points = tuple(
            LidarPoint(
                x=float(x_coord),
                y=float(y_coord),
                z=float(z_coord),
                intensity=None if las.intensity is None else int(las.intensity[index]),
            )
            for index, (x_coord, y_coord, z_coord) in enumerate(
                zip(las.x, las.y, las.z, strict=True)
            )
        )
        return LidarDataset(source=source_path.resolve(), points=points)
    if suffix in {".las", ".laz"}:
        if laspy is None:
            message = "LAS/LAZ loading requires the optional `laspy` dependency."
            raise RuntimeError(message)
        las = laspy.read(source_path)
        points = tuple(
            LidarPoint(
                x=float(x_coord),
                y=float(y_coord),
                z=float(z_coord),
                intensity=None if las.intensity is None else int(las.intensity[index]),
            )
            for index, (x_coord, y_coord, z_coord) in enumerate(
                zip(las.x, las.y, las.z, strict=True)
            )
        )
        return LidarDataset(source=source_path.resolve(), points=points)

    if suffix in {".csv", ".xyz", ".txt"}:
        return LidarDataset(
            source=source_path.resolve(),
            points=_load_xyz_text(source_path),
        )

    if suffix == ".json":
        payload = json.loads(source_path.read_text(encoding="utf-8"))
        raw_points = payload.get("points") if isinstance(payload, dict) else payload
        if not isinstance(raw_points, list):
            message = f"{source_path} must contain a list of points."
            raise TypeError(message)
        points = tuple(
            LidarPoint(
                x=float(point["x"]),
                y=float(point["y"]),
                z=float(point["z"]),
                intensity=None
                if point.get("intensity") is None
                else int(point["intensity"]),
            )
            for point in raw_points
            if isinstance(point, dict)
        )
        return LidarDataset(source=source_path.resolve(), points=points)

    message = f"Unsupported LiDAR format: {source_path.suffix}"
    raise ValueError(message)

build_asset_manifest

build_asset_manifest(
    *,
    study_area_name: str,
    bbox: BoundingBox,
    cache_dir: Path,
    citygml_source: Path,
    metadata: tuple[DataSourceMetadata, ...],
    footprints_source: Path | None = None,
    pluto_source: Path | None = None,
    dem_source: Path | None = None,
    lidar_source: Path | None = None,
) -> StudyAreaAssetManifest

Build a typed manifest for one official mesh study area.

Source code in src/nyc_mesh/io/_official.py

def build_asset_manifest(
    *,
    study_area_name: str,
    bbox: BoundingBox,
    cache_dir: Path,
    citygml_source: Path,
    metadata: tuple[DataSourceMetadata, ...],
    footprints_source: Path | None = None,
    pluto_source: Path | None = None,
    dem_source: Path | None = None,
    lidar_source: Path | None = None,
) -> StudyAreaAssetManifest:
    """Build a typed manifest for one official mesh study area."""

    return StudyAreaAssetManifest(
        study_area_name=study_area_name,
        bbox=bbox,
        cache_dir=cache_dir,
        citygml_source=citygml_source,
        metadata=metadata,
        footprints_source=footprints_source,
        pluto_source=pluto_source,
        dem_source=dem_source,
        lidar_source=lidar_source,
    )

build_enriched_footprint_geojson

build_enriched_footprint_geojson(
    footprint_rows: list[dict[str, Any]],
    pluto_rows: list[dict[str, Any]],
) -> dict[str, Any]

Merge PLUTO attributes into official building-footprint features.

Source code in src/nyc_mesh/io/_official.py

def build_enriched_footprint_geojson(
    footprint_rows: list[dict[str, Any]],
    pluto_rows: list[dict[str, Any]],
) -> dict[str, Any]:
    """Merge PLUTO attributes into official building-footprint features."""

    pluto_by_bbl = {
        _normalize_bbl(row.get("bbl")): row
        for row in pluto_rows
        if row.get("bbl") is not None
    }
    features = []
    for row in footprint_rows:
        geometry = row.get("the_geom")
        if not isinstance(geometry, dict):
            continue
        bbl = _normalize_bbl(row.get("mappluto_bbl") or row.get("base_bbl") or "")
        pluto = pluto_by_bbl.get(bbl, {})
        properties = {
            "building_id": str(row.get("bin") or row.get("doitt_id") or bbl),
            "bin": str(row.get("bin") or ""),
            "bbl": bbl,
            "base_bbl": str(row.get("base_bbl") or ""),
            "mappluto_bbl": str(row.get("mappluto_bbl") or ""),
            "height_roof": row.get("height_roof"),
            "ground_elevation": row.get("ground_elevation"),
            "construction_year": row.get("construction_year"),
            "address": str(pluto.get("address") or ""),
            "borough": str(pluto.get("borough") or ""),
            "landuse": str(pluto.get("landuse") or ""),
            "ownername": str(pluto.get("ownername") or ""),
            "yearbuilt": str(pluto.get("yearbuilt") or ""),
            "numfloors": str(pluto.get("numfloors") or ""),
            "zipcode": str(pluto.get("zipcode") or ""),
        }
        features.append(
            {
                "type": "Feature",
                "id": properties["building_id"],
                "properties": properties,
                "geometry": geometry,
            }
        )
    return {"type": "FeatureCollection", "features": features}

ensure_citygml_archive

ensure_citygml_archive(
    target_path: Path,
    *,
    refresh: bool = False,
    source_url: str = NYC_CITYGML_ARCHIVE_URL,
) -> Path

Download the official CityGML archive when explicitly requested.

Source code in src/nyc_mesh/io/_official.py

def ensure_citygml_archive(
    target_path: Path,
    *,
    refresh: bool = False,
    source_url: str = NYC_CITYGML_ARCHIVE_URL,
) -> Path:
    """Download the official CityGML archive when explicitly requested."""

    if target_path.exists() and not refresh:
        return target_path
    target_path.parent.mkdir(parents=True, exist_ok=True)
    with urlopen(source_url, timeout=120.0) as response:
        target_path.write_bytes(response.read())
    return target_path

fetch_building_footprints_for_bbls

fetch_building_footprints_for_bbls(
    bbls: tuple[str, ...], *, page_size: int = 5000
) -> list[dict[str, Any]]

Fetch official building-footprint rows for the supplied BBL set.

Source code in src/nyc_mesh/io/_official.py

def fetch_building_footprints_for_bbls(
    bbls: tuple[str, ...],
    *,
    page_size: int = 5000,
) -> list[dict[str, Any]]:
    """Fetch official building-footprint rows for the supplied BBL set."""

    normalized_bbls = tuple(sorted({_normalize_bbl(value) for value in bbls if value}))
    if not normalized_bbls:
        return []
    joined = ", ".join(f"'{value}'" for value in normalized_bbls)
    where = f"mappluto_bbl in ({joined}) or base_bbl in ({joined})"
    return _fetch_paginated(
        NYC_BUILDING_FOOTPRINTS_API_URL,
        where=where,
        page_size=page_size,
    )

fetch_pluto_records_for_bbox

fetch_pluto_records_for_bbox(
    bbox: BoundingBox, *, page_size: int = 5000
) -> list[dict[str, Any]]

Fetch PLUTO rows whose centroid falls inside the requested bbox.

Source code in src/nyc_mesh/io/_official.py

def fetch_pluto_records_for_bbox(
    bbox: BoundingBox,
    *,
    page_size: int = 5000,
) -> list[dict[str, Any]]:
    """Fetch PLUTO rows whose centroid falls inside the requested bbox."""

    where = (
        f"latitude >= {bbox.min_lat} and latitude <= {bbox.max_lat} and "
        f"longitude >= {bbox.min_lon} and longitude <= {bbox.max_lon}"
    )
    return _fetch_paginated(NYC_PLUTO_API_URL, where=where, page_size=page_size)

metadata_row

metadata_row(
    *,
    name: str,
    source_url: str,
    cache_path: Path,
    record_count: int,
    notes: str = "",
) -> DataSourceMetadata

Create a metadata row timestamped at the current UTC time.

Source code in src/nyc_mesh/io/_official.py

def metadata_row(
    *,
    name: str,
    source_url: str,
    cache_path: Path,
    record_count: int,
    notes: str = "",
) -> DataSourceMetadata:
    """Create a metadata row timestamped at the current UTC time."""

    return DataSourceMetadata(
        name=name,
        source_url=source_url,
        cache_path=cache_path,
        refreshed_at=datetime.now(tz=UTC),
        record_count=record_count,
        notes=notes,
    )

require_local_asset

require_local_asset(
    provided_path: str | Path | None,
    *,
    label: str,
    source_url: str,
) -> Path

Resolve a user-staged official asset or raise with download guidance.

Source code in src/nyc_mesh/io/_official.py

def require_local_asset(
    provided_path: str | Path | None,
    *,
    label: str,
    source_url: str,
) -> Path:
    """Resolve a user-staged official asset or raise with download guidance."""

    if provided_path is None:
        message = (
            f"{label} must be staged locally for this workflow. Download it from "
            f"{source_url} and pass the local path."
        )
        raise FileNotFoundError(message)
    path = Path(provided_path).expanduser().resolve()
    if not path.exists():
        message = f"{label} does not exist: {path}"
        raise FileNotFoundError(message)
    return path

Analysis

nyc_mesh.analysis

Public analysis helpers for nyc-mesh.

clip_to_bbox

clip_to_bbox(
    data: tuple[BuildingFeature, ...], bbox: BoundingBox
) -> tuple[BuildingFeature, ...]

Clip building features to a WGS84 bounding box.

Source code in src/nyc_mesh/analysis/_core.py

def clip_to_bbox(
    data: tuple[BuildingFeature, ...],
    bbox: BoundingBox,
) -> tuple[BuildingFeature, ...]:
    """Clip building features to a WGS84 bounding box."""

    return tuple(
        feature
        for feature in data
        if bbox_intersects(ring_bounds(feature.footprint_4326), bbox)
    )

extract_buildings

extract_buildings(
    citygml_data: CityGMLDataset,
) -> tuple[BuildingFeature, ...]

Extract height-aware WGS84 building footprints from loaded CityGML data.

Source code in src/nyc_mesh/analysis/_core.py

def extract_buildings(citygml_data: CityGMLDataset) -> tuple[BuildingFeature, ...]:
    """Extract height-aware WGS84 building footprints from loaded CityGML data."""

    extracted: list[BuildingFeature] = []
    for building in citygml_data.buildings:
        if building.measured_height is None:
            continue
        footprint_4326 = project_ring_to_wgs84(building.footprint_2263)
        if not footprint_4326:
            continue
        extracted.append(
            BuildingFeature(
                building_id=building.building_id,
                footprint_4326=footprint_4326,
                height=building.measured_height,
            )
        )
    return tuple(extracted)

generate_terrain_mesh

generate_terrain_mesh(
    lidar_or_dem_data: LidarDataset | DEMDataset,
) -> TerrainMeshDataset

Generate a lightweight terrain mesh from DEM or LiDAR inputs.

Source code in src/nyc_mesh/analysis/_core.py

def generate_terrain_mesh(
    lidar_or_dem_data: LidarDataset | DEMDataset,
) -> TerrainMeshDataset:
    """Generate a lightweight terrain mesh from DEM or LiDAR inputs."""

    if isinstance(lidar_or_dem_data, DEMDataset):
        return _terrain_mesh_from_dem(lidar_or_dem_data)
    return _terrain_mesh_from_lidar(lidar_or_dem_data)

join_pluto

join_pluto(
    buildings: tuple[BuildingFeature, ...],
    pluto_data: FootprintDataset,
) -> tuple[BuildingFeature, ...]

Join footprint properties onto buildings by id or centroid overlap.

Source code in src/nyc_mesh/analysis/_core.py

def join_pluto(
    buildings: tuple[BuildingFeature, ...],
    pluto_data: FootprintDataset,
) -> tuple[BuildingFeature, ...]:
    """Join footprint properties onto buildings by id or centroid overlap."""

    features_by_id = {feature.feature_id: feature for feature in pluto_data.features}
    joined = []
    for building in buildings:
        match = features_by_id.get(building.building_id)
        if match is None:
            centroid = building.centroid
            for feature in pluto_data.features:
                bounds = ring_bounds(feature.footprint_4326)
                if bounds.contains(*centroid) and point_in_polygon(
                    centroid,
                    feature.footprint_4326,
                ):
                    match = feature
                    break
        if match is None:
            joined.append(building)
            continue
        joined.append(building.with_properties(match.properties))
    return tuple(joined)

Export

nyc_mesh.export

Public exporters for nyc-mesh.

export_3d_tiles

export_3d_tiles(
    data: tuple[BuildingFeature, ...], target: ExportTarget
) -> Path

Export a minimal Cesium 3D Tiles package backed by a glTF payload.

Source code in src/nyc_mesh/export/_core.py

def export_3d_tiles(data: tuple[BuildingFeature, ...], target: ExportTarget) -> Path:
    """Export a minimal Cesium 3D Tiles package backed by a glTF payload."""

    output_path = _validate_target_format(target, expected_formats=("3dtiles", "json"))
    if not data:
        message = "export_3d_tiles() requires at least one building feature."
        raise ValueError(message)

    gltf_path = output_path.parent / "content.gltf"
    export_gltf(data, ExportTarget(format="gltf", output_path=gltf_path))

    bounds = _feature_bounds(data)
    max_height = max(feature.height for feature in data)
    payload = {
        "asset": {"version": "1.1"},
        "extensionsUsed": ["3DTILES_content_gltf"],
        "extensionsRequired": ["3DTILES_content_gltf"],
        "geometricError": 0.0,
        "root": {
            "boundingVolume": {
                "region": region_from_bounds(
                    bounds,
                    min_height=0.0,
                    max_height=max_height,
                )
            },
            "geometricError": 0.0,
            "refine": "ADD",
            "content": {"uri": gltf_path.name},
        },
    }
    output_path.write_text(f"{json.dumps(payload, indent=2)}\n", encoding="utf-8")
    return output_path

export_geojson

export_geojson(
    data: tuple[BuildingFeature, ...], target: ExportTarget
) -> Path

Export WGS84 building features to a GeoJSON FeatureCollection.

Source code in src/nyc_mesh/export/_core.py

def export_geojson(data: tuple[BuildingFeature, ...], target: ExportTarget) -> Path:
    """Export WGS84 building features to a GeoJSON ``FeatureCollection``."""

    output_path = _validate_target_format(target, expected_formats=("geojson",))
    payload = {
        "type": "FeatureCollection",
        "features": [
            {
                "type": "Feature",
                "id": feature.building_id,
                "properties": {
                    "building_id": feature.building_id,
                    "height": feature.height,
                    **feature.properties,
                },
                "geometry": {
                    "type": "Polygon",
                    "coordinates": [
                        [
                            [longitude, latitude]
                            for longitude, latitude in feature.footprint_4326
                        ]
                    ],
                },
            }
            for feature in data
        ],
    }
    output_path.write_text(f"{json.dumps(payload, indent=2)}\n", encoding="utf-8")
    return output_path

export_geoparquet

export_geoparquet(
    data: tuple[BuildingFeature, ...], target: ExportTarget
) -> Path

Export building features to GeoParquet using a WKB geometry column.

Source code in src/nyc_mesh/export/_core.py

def export_geoparquet(data: tuple[BuildingFeature, ...], target: ExportTarget) -> Path:
    """Export building features to GeoParquet using a WKB geometry column."""

    if pa is None or pq is None:
        message = "GeoParquet export requires the optional `pyarrow` dependency."
        raise RuntimeError(message)

    output_path = _validate_target_format(
        target, expected_formats=("parquet", "geoparquet")
    )
    building_ids = [feature.building_id for feature in data]
    heights = [feature.height for feature in data]
    properties_json = [
        json.dumps(feature.properties, sort_keys=True) for feature in data
    ]
    geometries = [_polygon_to_wkb(feature.footprint_4326) for feature in data]
    table = pa.table(
        {
            "building_id": building_ids,
            "height": heights,
            "properties_json": properties_json,
            "geometry": geometries,
        }
    )
    geo_metadata = {
        "version": "1.1.0",
        "primary_column": "geometry",
        "columns": {
            "geometry": {
                "encoding": "WKB",
                "geometry_types": ["Polygon"],
                "crs": "OGC:CRS84",
            }
        },
    }
    table = table.replace_schema_metadata(
        {
            **(table.schema.metadata or {}),
            b"geo": json.dumps(geo_metadata).encode("utf-8"),
        }
    )
    pq.write_table(table, output_path)
    return output_path

export_gltf

export_gltf(
    data: tuple[BuildingFeature, ...] | TerrainMeshDataset,
    target: ExportTarget,
) -> Path

Export a lightweight glTF scene for 3D viewers.

Source code in src/nyc_mesh/export/_core.py

def export_gltf(
    data: tuple[BuildingFeature, ...] | TerrainMeshDataset,
    target: ExportTarget,
) -> Path:
    """Export a lightweight glTF scene for 3D viewers."""

    output_path = _validate_target_format(target, expected_formats=("gltf",))
    if isinstance(data, TerrainMeshDataset):
        vertices, indices = _terrain_mesh(data)
    else:
        vertices, indices = _building_mesh(data)
    gltf_payload = _build_gltf_json(vertices, indices)
    output_path.write_text(f"{json.dumps(gltf_payload, indent=2)}\n", encoding="utf-8")
    return output_path

Pipeline

nyc_mesh.pipeline

High-level pipeline helpers for nyc-mesh.

DEFAULT_STUDY_AREAS module-attribute

DEFAULT_STUDY_AREAS = {
    "lower-manhattan-skyline": BoundingBox(
        min_lat=40.7035,
        min_lon=-74.0185,
        max_lat=40.7178,
        max_lon=-73.9975,
    ),
    "empire-state-building": BoundingBox(
        min_lat=40.7445,
        min_lon=-73.9899,
        max_lat=40.7502,
        max_lon=-73.9839,
    ),
}

extract_citygml_buildings

extract_citygml_buildings(
    source: str | Path, *, bbox: BoundingBox | None = None
) -> tuple[BuildingFeature, ...]

Load local or archived CityGML, extract buildings, and optionally clip.

Source code in src/nyc_mesh/pipeline.py

def extract_citygml_buildings(
    source: str | Path,
    *,
    bbox: BoundingBox | None = None,
) -> tuple[BuildingFeature, ...]:
    """Load local or archived CityGML, extract buildings, and optionally clip."""

    dataset = load_citygml(source)
    buildings = extract_buildings(dataset)
    return clip_to_bbox(buildings, bbox) if bbox is not None else buildings

build_study_area_manifest

build_study_area_manifest(
    *,
    study_area_name: str,
    bbox: BoundingBox,
    cache_dir: str | Path,
    citygml_path: str | Path | None = None,
    allow_citygml_download: bool = False,
    dem_path: str | Path | None = None,
    lidar_path: str | Path | None = None,
) -> StudyAreaAssetManifest

Prepare official cache assets for one mesh study area.

Source code in src/nyc_mesh/pipeline.py

def build_study_area_manifest(
    *,
    study_area_name: str,
    bbox: BoundingBox,
    cache_dir: str | Path,
    citygml_path: str | Path | None = None,
    allow_citygml_download: bool = False,
    dem_path: str | Path | None = None,
    lidar_path: str | Path | None = None,
) -> StudyAreaAssetManifest:
    """Prepare official cache assets for one mesh study area."""

    cache_root = Path(cache_dir).expanduser().resolve()
    cache_root.mkdir(parents=True, exist_ok=True)

    if citygml_path is not None:
        citygml_source = require_local_asset(
            citygml_path,
            label="CityGML source",
            source_url=NYC_CITYGML_ARCHIVE_URL,
        )
        citygml_metadata = metadata_row(
            name="citygml_source",
            source_url=NYC_CITYGML_ARCHIVE_URL,
            cache_path=citygml_source,
            record_count=1,
            notes="User-staged official CityGML source.",
        )
    elif allow_citygml_download:
        citygml_source = ensure_citygml_archive(cache_root / "DA_WISE_GML.zip")
        citygml_metadata = metadata_row(
            name="citygml_source",
            source_url=NYC_CITYGML_ARCHIVE_URL,
            cache_path=citygml_source,
            record_count=1,
            notes="Downloaded official CityGML archive.",
        )
    else:
        citygml_source = require_local_asset(
            None,
            label="CityGML source",
            source_url=NYC_CITYGML_ARCHIVE_URL,
        )
        citygml_metadata = metadata_row(
            name="citygml_source",
            source_url=NYC_CITYGML_ARCHIVE_URL,
            cache_path=citygml_source,
            record_count=1,
        )

    pluto_rows = fetch_pluto_records_for_bbox(bbox)
    pluto_path = cache_root / "pluto.json"
    pluto_path.write_text(f"{json.dumps(pluto_rows, indent=2)}\n", encoding="utf-8")
    pluto_metadata = metadata_row(
        name="pluto_rows",
        source_url=NYC_PLUTO_API_URL,
        cache_path=pluto_path,
        record_count=len(pluto_rows),
        notes="Filtered to the requested study-area bbox.",
    )

    bbls = tuple(
        str(row.get("bbl", "")).split(".", maxsplit=1)[0]
        for row in pluto_rows
        if row.get("bbl")
    )
    footprint_rows = fetch_building_footprints_for_bbls(bbls)
    footprints_payload = build_enriched_footprint_geojson(footprint_rows, pluto_rows)
    footprints_path = cache_root / "footprints.geojson"
    footprints_path.write_text(
        f"{json.dumps(footprints_payload, indent=2)}\n",
        encoding="utf-8",
    )
    footprints_metadata = metadata_row(
        name="building_footprints",
        source_url=NYC_BUILDING_FOOTPRINTS_API_URL,
        cache_path=footprints_path,
        record_count=len(footprints_payload["features"]),
        notes="Fetched by BBL after PLUTO bbox selection.",
    )

    dem_source = (
        None
        if dem_path is None
        else require_local_asset(
            dem_path,
            label="DEM source",
            source_url=NYC_DEM_DATASET_URL,
        )
    )
    lidar_source = (
        None
        if lidar_path is None
        else require_local_asset(
            lidar_path,
            label="LiDAR source",
            source_url=NYC_LIDAR_DATASET_URL,
        )
    )

    metadata = [citygml_metadata, pluto_metadata, footprints_metadata]
    if dem_source is not None:
        metadata.append(
            metadata_row(
                name="dem_source",
                source_url=NYC_DEM_DATASET_URL,
                cache_path=dem_source,
                record_count=1,
                notes="User-staged official DEM source.",
            )
        )
    if lidar_source is not None:
        metadata.append(
            metadata_row(
                name="lidar_source",
                source_url=NYC_LIDAR_DATASET_URL,
                cache_path=lidar_source,
                record_count=1,
                notes="User-staged official LiDAR source.",
            )
        )

    manifest = build_asset_manifest(
        study_area_name=study_area_name,
        bbox=bbox,
        cache_dir=cache_root,
        citygml_source=citygml_source,
        metadata=tuple(metadata),
        footprints_source=footprints_path,
        pluto_source=pluto_path,
        dem_source=dem_source,
        lidar_source=lidar_source,
    )
    manifest_path = cache_root / "asset-manifest.json"
    manifest_payload = {
        "study_area_name": manifest.study_area_name,
        "bbox": {
            "min_lat": manifest.bbox.min_lat,
            "min_lon": manifest.bbox.min_lon,
            "max_lat": manifest.bbox.max_lat,
            "max_lon": manifest.bbox.max_lon,
        },
        "citygml_source": str(manifest.citygml_source),
        "footprints_source": None
        if manifest.footprints_source is None
        else str(manifest.footprints_source),
        "pluto_source": None
        if manifest.pluto_source is None
        else str(manifest.pluto_source),
        "dem_source": None if manifest.dem_source is None else str(manifest.dem_source),
        "lidar_source": None
        if manifest.lidar_source is None
        else str(manifest.lidar_source),
        "sources": [
            {
                "name": item.name,
                "source_url": item.source_url,
                "cache_path": str(item.cache_path),
                "refreshed_at": item.refreshed_at.isoformat(),
                "record_count": item.record_count,
                "notes": item.notes,
            }
            for item in manifest.metadata
        ],
    }
    manifest_path.write_text(
        f"{json.dumps(manifest_payload, indent=2)}\n",
        encoding="utf-8",
    )
    return manifest

extract_manifest_buildings

extract_manifest_buildings(
    manifest: StudyAreaAssetManifest,
) -> tuple[BuildingFeature, ...]

Load, clip, and enrich buildings for a prepared study-area manifest.

Source code in src/nyc_mesh/pipeline.py

def extract_manifest_buildings(
    manifest: StudyAreaAssetManifest,
) -> tuple[BuildingFeature, ...]:
    """Load, clip, and enrich buildings for a prepared study-area manifest."""

    buildings = extract_citygml_buildings(manifest.citygml_source, bbox=manifest.bbox)
    if manifest.footprints_source is None:
        return buildings
    return join_pluto(buildings, load_footprints(manifest.footprints_source))

export_citygml_geojson

export_citygml_geojson(
    source: str | Path,
    output_path: str | Path,
    *,
    bbox: BoundingBox | None = None,
) -> Path

Run the CityGML happy path and write GeoJSON.

Source code in src/nyc_mesh/pipeline.py

def export_citygml_geojson(
    source: str | Path,
    output_path: str | Path,
    *,
    bbox: BoundingBox | None = None,
) -> Path:
    """Run the CityGML happy path and write GeoJSON."""

    buildings = extract_citygml_buildings(source, bbox=bbox)
    target = ExportTarget(format="geojson", output_path=Path(output_path).expanduser())
    return export_geojson(buildings, target)

export_citygml_geoparquet

export_citygml_geoparquet(
    source: str | Path,
    output_path: str | Path,
    *,
    bbox: BoundingBox | None = None,
) -> Path

Run the CityGML happy path and write GeoParquet.

Source code in src/nyc_mesh/pipeline.py

def export_citygml_geoparquet(
    source: str | Path,
    output_path: str | Path,
    *,
    bbox: BoundingBox | None = None,
) -> Path:
    """Run the CityGML happy path and write GeoParquet."""

    buildings = extract_citygml_buildings(source, bbox=bbox)
    target = ExportTarget(
        format="geoparquet", output_path=Path(output_path).expanduser()
    )
    return export_geoparquet(buildings, target)

CLI

nyc_mesh.cli

Public CLI entry points for nyc-mesh.

build_parser

build_parser() -> argparse.ArgumentParser

Build the CLI parser for the current extraction and export workflows.

Source code in src/nyc_mesh/cli/_main.py

def build_parser() -> argparse.ArgumentParser:
    """Build the CLI parser for the current extraction and export workflows."""

    parser = argparse.ArgumentParser(
        prog="nyc-mesh",
        description=(
            "Extract height-aware outputs from local CityGML files using the "
            "implemented nyc-mesh workflow."
        ),
        epilog=(
            "Current assumption: CityGML coordinates are in NYC EPSG:2263 and "
            "are reprojected to WGS84 (EPSG:4326) before optional clipping."
        ),
    )
    parser.add_argument(
        "--version",
        action="version",
        version=f"%(prog)s {_VERSION}",
    )

    subparsers = parser.add_subparsers(dest="command", metavar="command")
    subparsers.required = True

    for command_name, help_text, output_flag in (
        (
            "export-geojson",
            "Export the CityGML happy path to GeoJSON.",
            _run_export_geojson,
        ),
        (
            "export-geoparquet",
            "Export the CityGML happy path to GeoParquet.",
            _run_export_geoparquet,
        ),
    ):
        subparser = subparsers.add_parser(command_name, help=help_text)
        subparser.add_argument(
            "--input",
            type=Path,
            required=True,
            help="Path to a local CityGML file.",
        )
        subparser.add_argument(
            "--output",
            type=Path,
            required=True,
            help="Output path for the exported file.",
        )
        _attach_bbox_arguments(subparser)
        subparser.set_defaults(handler=output_flag, parser=subparser)

    return parser

main

main(argv: Sequence[str] | None = None) -> int

Run the installed nyc-mesh console script.

Source code in src/nyc_mesh/cli/_main.py

def main(argv: Sequence[str] | None = None) -> int:
    """Run the installed ``nyc-mesh`` console script."""

    parser = build_parser()
    try:
        args = parser.parse_args(list(argv) if argv is not None else None)
        handler = cast("Callable[[argparse.Namespace], int]", args.handler)
        return handler(args)
    except SystemExit as exc:
        return exc.code if isinstance(exc.code, int) else 1
    except (
        etree.XMLSyntaxError,
        FileNotFoundError,
        OSError,
        RuntimeError,
        ValueError,
    ) as exc:
        _write_line(sys.stderr, f"nyc-mesh: error: {exc}")
        return 1