API Reference

def main(
    version: str | None = None,
    input_path: Path | None = None,
    snapshot_date: str | None = None,
) -> int:
    """Build the Minnesota precinct layer.

    Args:
        version: Snapshot version used for output directory naming.
        input_path: Optional explicit source GeoJSON path.
        snapshot_date: Source snapshot date in YYYY-MM-DD format.

    Returns:
        0 when the build succeeds; otherwise 1.
    """
    try:
        build_cfg = _load_build_cfg()

        if not version:
            raise BuildError("version is required")

        if input_path is not None:
            source_path = input_path
        else:
            source_path = (
                get_data_in_dir() / "states" / "minnesota" / f"precincts_{version}.json"
            )

        if not source_path.exists():
            raise BuildError(f"Input not found: {source_path}")

        out_dir = _out_dir(version)

        gdf: gpd.GeoDataFrame = gpd.read_file(source_path)

        gdf = _normalize_columns(
            gdf,
            to_lower=bool(build_cfg.get("fields_lowercase", True)),
            trim=bool(build_cfg.get("fields_trim", True)),
        )

        gdf = _rename_columns(
            gdf,
            mapping=build_cfg.get("fields_rename") or {},
        )

        gdf = _keep_columns(
            gdf,
            keep=build_cfg.get("fields_keep") or [],
        )

        add_fields: dict[str, Any] = {
            "snapshot_version": version,
        }

        if snapshot_date is not None:
            add_fields["snapshot_date"] = snapshot_date

        gdf = _add_constant_fields(
            gdf,
            add_fields=add_fields,
        )

        if bool(build_cfg.get("repair_geometries", True)):
            gdf = _repair_geometries(gdf)

        full_path = out_dir / "mn-precincts-full.geojson"

        gdf.to_file(
            full_path,
            driver="GeoJSON",
        )

        logger.info(f"Wrote full: {full_path}")

        web_geojson_name = "mn-precincts-web.geojson"
        web_geojson_path = out_dir / web_geojson_name

        shutil.copy2(
            full_path,
            web_geojson_path,
        )

        logger.info(f"Wrote web GeoJSON: {web_geojson_path}")

        topo_name: str | None = None

        if bool(build_cfg.get("write_topojson", False)):
            executable = _which_mapshaper()

            if executable:
                topo_path = out_dir / "mn-precincts-web.topojson"

                topo_output = _write_topojson(
                    executable,
                    web_geojson_path,
                    topo_path,
                    simplify_pct=_clamped_pct(build_cfg.get("simplify_pct", 0)),
                )

                topo_name = topo_output.name if topo_output else None
            else:
                logger.warning("mapshaper not found; skipping TopoJSON.")

        _write_metadata(
            full_path=full_path,
            web_name=web_geojson_name,
            topo_name=topo_name,
            out_dir=out_dir,
            version=version,
            snapshot_date=snapshot_date,
        )

        logger.info("Build completed.")
        return 0

    except BuildError as exc:
        logger.error(f"Build failed: {exc}")
        return 1

def build_index_main() -> int:
    """Scan data-out for GeoJSON, compute metadata, generate index and manifest.

    Workflow:
    1. Scans the output directory for all GeoJSON files.
    2. For each file, computes its bounding box and feature count.
    3. Writes a flat index of all files to 'index.json'.
    4. Writes a dataset-level manifest to 'manifest.json'.
    5. Writes a state-level index to 'states/minnesota/index.json'.

    Returns:
        int: 0 if the index build succeeds, 1 if an error occurs.
    """
    try:
        out_root = get_data_out_dir()
        flat_index: list[dict[str, Any]] = []

        logger.info(f"Scanning {out_root} for GeoJSONs...")
        for geojson in out_root.rglob("*.geojson"):
            bbox = _compute_bbox(geojson)
            nfeat = _compute_feature_count(geojson)
            flat_index.append(
                {
                    "path": geojson.relative_to(out_root).as_posix(),
                    "bbox": bbox,
                    "features": nfeat,
                }
            )

        # data-out/index.json
        index_file = out_root / "index.json"
        with index_file.open("w", encoding="utf-8") as f:
            json.dump(flat_index, f, indent=2)
        logger.info(f"Flat index written: {index_file} ({len(flat_index)} files)")

        # data-out/manifest.json (dataset-level)
        _write_manifest(out_root, flat_index)

        # data-out/states/minnesota/index.json (layer pointers)
        _write_state_index(out_root)

        return 0

    except Exception as e:  # noqa: BLE001
        logger.error(f"Index build failed: {e}")
        return 1

def main() -> int:
    """Execute the main entry point for building the index.

    Returns:
        int: 0 if the index was built successfully, 1 if an error occurs.
    """
    return build_index_main()

def main() -> int:
    """Download, build, validate, and index the current Minnesota precinct snapshot."""
    try:
        source_url = _load_source_url()

        logger.info(f"Fetching Minnesota precinct source: {source_url}")
        data = _download(source_url)

        source_date, version = _read_source_metadata(data)

        logger.info(f"Source date: {source_date}")
        logger.info(f"Snapshot version: {version}")

        if _already_current(version, source_date):
            logger.info("Snapshot is already current.")
            return 0

        input_path = (
            get_data_in_dir() / "states" / "minnesota" / f"precincts_{version}.json"
        )
        input_path.parent.mkdir(parents=True, exist_ok=True)
        input_path.write_bytes(data)

        if (
            build_layer.main(
                version=version,
                input_path=input_path,
                snapshot_date=source_date,
            )
            != 0
        ):
            raise RefreshError("Build failed")

        if validate.main(version=version) != 0:
            raise RefreshError("Validation failed")

        if index_mod.main() != 0:
            raise RefreshError("Index generation failed")

        logger.info(f"Refresh completed successfully for {version}.")
        return 0

    except (RefreshError, OSError, URLError, ValueError, yaml.YAMLError) as exc:
        logger.error(f"Refresh failed: {exc}")
        return 1

def main(version: str) -> int:
    """Validate MN precincts outputs for the specified snapshot version."""
    try:
        out_dir = _out_dir(version)
        _require_files(out_dir, REQUIRED_FILES)

        full_path = out_dir / "mn-precincts-full.geojson"
        gdf = _load_gdf(full_path)

        _require_columns(gdf, REQUIRED_COLUMNS)
        _check_precinct_id_consistency(gdf, "precinct_id")

        logger.info("Validation passed.")
        return 0
    except ValidateError as exc:
        logger.error(f"Validation failed: {exc}")
        return 1