"""Click CLI commands for osm-pt-geojson.""" import json import sys import click from osm_geojson.pt.core import ( GeoJson, OsmError, build_route_coords, fetch_relation_full, make_geojson, nearest_coord_index, node_name, parse_elements, ) def output_geojson(geojson: GeoJson, output_path: str | None) -> None: """Write GeoJSON to a file, or to stdout if output_path is None.""" text = json.dumps(geojson, ensure_ascii=False, indent=2) if output_path: with open(output_path, "w", encoding="utf-8") as f: f.write(text) click.echo(f"Wrote {output_path}", err=True) else: click.echo(text) @click.group() def cli() -> None: """OSM public transport route → GeoJSON tool.""" @cli.command("list-stations") @click.argument("relation_id", type=int) def list_stations(relation_id: int) -> None: """List all stations in an OSM public transport route relation.""" try: data = fetch_relation_full(relation_id) nodes, ways, stop_ids, way_ids, tags = parse_elements(data, relation_id) except OsmError as e: click.echo(f"Error: {e}", err=True) sys.exit(1) click.echo(f"Route: {tags.get('name', relation_id)}") click.echo(f"Stops ({len(stop_ids)}):") for i, sid in enumerate(stop_ids, 1): if sid in nodes: click.echo(f" {i:2}. {node_name(nodes[sid])}") else: click.echo(f" {i:2}. (node {sid} not in response)") @cli.command("route-between") @click.argument("relation_id", type=int) @click.argument("from_station") @click.argument("to_station") @click.option("--output", "-o", type=click.Path(), default=None, help="Output file (default: stdout)") @click.option("--no-stops", is_flag=True, default=False, help="Omit stop points from output.") def route_between( relation_id: int, from_station: str, to_station: str, output: str | None, no_stops: bool, ) -> None: """Output GeoJSON for the route segment between two named stations.""" try: data = fetch_relation_full(relation_id) nodes, ways, stop_ids, way_ids, tags = parse_elements(data, relation_id) except OsmError as e: click.echo(f"Error: {e}", err=True) sys.exit(1) route_coords = build_route_coords(way_ids, ways, nodes) def find_stop(name: str) -> int | None: """Return the node ID of the stop matching name (case-insensitive), or None.""" for sid in stop_ids: if sid in nodes and node_name(nodes[sid]).lower() == name.lower(): return sid return None sid_from = find_stop(from_station) sid_to = find_stop(to_station) errors = [] if sid_from is None: errors.append(f"Station not found: {from_station!r}") if sid_to is None: errors.append(f"Station not found: {to_station!r}") if errors: for msg in errors: click.echo(f"Error: {msg}", err=True) click.echo("Available stations:", err=True) for sid in stop_ids: if sid in nodes: click.echo(f" {node_name(nodes[sid])}", err=True) sys.exit(1) assert sid_from is not None and sid_to is not None idx_from = nearest_coord_index(nodes[sid_from]["lon"], nodes[sid_from]["lat"], route_coords) idx_to = nearest_coord_index(nodes[sid_to]["lon"], nodes[sid_to]["lat"], route_coords) geojson = make_geojson( route_coords, stop_ids, nodes, tags, idx_from=idx_from, idx_to=idx_to, no_stops=no_stops ) output_geojson(geojson, output) @cli.command("full-route") @click.argument("relation_id", type=int) @click.option("--output", "-o", type=click.Path(), default=None, help="Output file (default: stdout)") @click.option("--no-stops", is_flag=True, default=False, help="Omit stop points from output.") def full_route(relation_id: int, output: str | None, no_stops: bool) -> None: """Output GeoJSON for the entire route, end to end.""" try: data = fetch_relation_full(relation_id) nodes, ways, stop_ids, way_ids, tags = parse_elements(data, relation_id) except OsmError as e: click.echo(f"Error: {e}", err=True) sys.exit(1) route_coords = build_route_coords(way_ids, ways, nodes) geojson = make_geojson(route_coords, stop_ids, nodes, tags, no_stops=no_stops) output_geojson(geojson, output)