diff --git a/.gitignore b/.gitignore index e267765..9a3d6e9 100644 --- a/.gitignore +++ b/.gitignore @@ -36,6 +36,7 @@ Makefile # Python __pycache__/ *.pyc +/cache/ # Demand data (large, contact authors per README) /LivingCity/berkeley_2018 diff --git a/README.md b/README.md index 40a8873..330d9fc 100755 --- a/README.md +++ b/README.md @@ -52,6 +52,31 @@ python3 viz/server.py --network data/networks/sf_bay_area # Open http://localhost:8080 ``` +## Generate a Custom OpenStreetMap Network + +With Python 3.9 or newer, install the optional network-generation dependencies, +then export a place directly to LPSim's current CSV schema: + +```bash +pip install -e ".[network]" +python3 tools/generate_network.py \ + --place "Berkeley, California, USA" \ + --output data/networks/berkeley +``` + +For reproducible or offline conversion, pass an existing OSMnx GraphML file: + +```bash +python3 tools/generate_network.py \ + --graphml berkeley.graphml \ + --output data/networks/berkeley +``` + +The exporter keeps the largest strongly connected component by default, +normalizes missing lane/speed attributes, reduces parallel OSM edges to the +fastest free-flow edge, and writes dense sequential node/edge identifiers. +Generate demand and optional GPU partitions with the existing tools below. + ## Generate Synthetic Demand ```bash diff --git a/pyproject.toml b/pyproject.toml index b21cdf3..f23d6c9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -27,7 +27,8 @@ dependencies = [ ] [project.optional-dependencies] -dev = ["pytest", "black", "ruff"] +dev = ["pytest", "black", "ruff", "networkx>=3.1"] +network = ["networkx>=3.1", "osmnx>=2.0,<3"] tools = ["matplotlib", "pandas", "seaborn"] [project.urls] @@ -37,6 +38,7 @@ Issues = "https://github.com/Xuan-1998/LPSim/issues" [project.scripts] lpsim-demand = "tools.generate_demand:main" +lpsim-network = "tools.generate_network:main" lpsim-partition = "tools.partition_network:main" [tool.setuptools.packages.find] diff --git a/tests/test_generate_network.py b/tests/test_generate_network.py new file mode 100644 index 0000000..84f9058 --- /dev/null +++ b/tests/test_generate_network.py @@ -0,0 +1,330 @@ +"""Acceptance tests for the OpenStreetMap network exporter proposed in #106. + +These tests intentionally exercise only an offline GraphML input. Live +Overpass requests do not belong in CI: they are slow, rate-limited, and +non-deterministic. The same exporter entry point can add place/bbox download +support while keeping normalization and CSV serialization testable here. +""" + +from __future__ import annotations + +import csv +import math +import subprocess +import sys +from pathlib import Path + +import networkx as nx +import pytest + +REPOSITORY_ROOT = Path(__file__).resolve().parents[1] +GENERATOR = REPOSITORY_ROOT / "tools" / "generate_network.py" +DEMAND_GENERATOR = REPOSITORY_ROOT / "tools" / "generate_demand.py" +PARTITIONER = REPOSITORY_ROOT / "tools" / "partition_network.py" + +NODE_COLUMNS = ["osmid", "x", "y", "ref", "highway", "index"] +EDGE_COLUMNS = [ + "uniqueid", + "osmid_u", + "osmid_v", + "length", + "lanes", + "speed_mph", + "u", + "v", +] + +LARGE_OSM_ID = 5_464_992_110 # larger than uint32, like current OSM data + + +def _run(*args: object) -> subprocess.CompletedProcess[str]: + command = [sys.executable, *(str(arg) for arg in args)] + return subprocess.run( + command, + cwd=REPOSITORY_ROOT, + capture_output=True, + text=True, + check=False, + timeout=30, + ) + + +def _assert_success(result: subprocess.CompletedProcess[str]) -> None: + assert result.returncode == 0, ( + f"command failed with exit code {result.returncode}\n" + f"stdout:\n{result.stdout}\n" + f"stderr:\n{result.stderr}" + ) + + +def _write_graphml_fixture(path: Path) -> None: + graph = nx.MultiDiGraph() + graph.graph["crs"] = "epsg:4326" + + # The three-node component is strongly connected and should be retained. + graph.add_node(LARGE_OSM_ID, x=-122.2700, y=37.8700, ref="", highway="") + graph.add_node(2_002, x=-122.2680, y=37.8710, ref="", highway="traffic_signals") + graph.add_node(3_003, x=-122.2660, y=37.8720, ref="7", highway="motorway_junction") + + # Parallel edges expose an important LPSim limitation: its C++ graph is + # keyed by (u, v), so the exporter must select one edge deterministically. + # The longer edge is faster in free-flow time and is therefore expected to + # win: 150 m / 50 mph < 100 m / 30 mph. + graph.add_edge( + LARGE_OSM_ID, + 2_002, + key=0, + length=100.0, + lanes="2", + maxspeed="30 mph", + highway="primary", + oneway=True, + ) + graph.add_edge( + LARGE_OSM_ID, + 2_002, + key=1, + length=150.0, + lanes="1", + maxspeed="50 mph", + highway="primary", + oneway=True, + ) + + # OSM maxspeed without an explicit unit is kph. Missing lanes and speed + # exercise the CLI defaults without depending on regional OSM coverage. + graph.add_edge( + 2_002, + 3_003, + length=80.0, + maxspeed="80", + highway="secondary", + oneway=True, + ) + graph.add_edge( + 3_003, + LARGE_OSM_ID, + length=90.0, + lanes="2;3", + highway="tertiary", + oneway=True, + ) + graph.add_edge( + 2_002, + LARGE_OSM_ID, + length=105.0, + lanes="1", + maxspeed="30 mph", + highway="primary", + oneway=True, + ) + graph.add_edge( + 3_003, + 2_002, + length=85.0, + lanes="1", + maxspeed="40 mph", + highway="secondary", + oneway=True, + ) + graph.add_edge( + LARGE_OSM_ID, + 3_003, + length=95.0, + lanes="1", + maxspeed="25 mph", + highway="tertiary", + oneway=True, + ) + + # A smaller disconnected component must not leak into generated OD data. + graph.add_node(9_001, x=-121.9000, y=37.5000) + graph.add_node(9_002, x=-121.8990, y=37.5010) + graph.add_edge( + 9_001, 9_002, length=25.0, lanes="1", maxspeed="20 mph", highway="residential" + ) + graph.add_edge( + 9_002, 9_001, length=25.0, lanes="1", maxspeed="20 mph", highway="residential" + ) + + nx.write_graphml(graph, path) + + +def _generate_network(graphml: Path, output: Path) -> subprocess.CompletedProcess[str]: + return _run( + GENERATOR, + "--graphml", + graphml, + "--output", + output, + "--connectivity", + "strong", + "--default-speed-kph", + 40, + "--default-lanes", + 1, + ) + + +def _read_csv(path: Path) -> tuple[list[str], list[dict[str, str]]]: + with path.open(newline="", encoding="utf-8") as stream: + reader = csv.DictReader(stream) + return list(reader.fieldnames or []), list(reader) + + +@pytest.fixture() +def generated_network(tmp_path: Path) -> Path: + graphml = tmp_path / "source.graphml" + output = tmp_path / "network" + _write_graphml_fixture(graphml) + result = _generate_network(graphml, output) + _assert_success(result) + return output + + +def test_cli_documents_download_and_offline_input_modes() -> None: + result = _run(GENERATOR, "--help") + _assert_success(result) + + assert "--place" in result.stdout + assert "--bbox" in result.stdout + assert "--graphml" in result.stdout + assert "--output" in result.stdout + + +def test_exports_exact_current_schema_with_dense_ids(generated_network: Path) -> None: + node_columns, nodes = _read_csv(generated_network / "nodes.csv") + edge_columns, edges = _read_csv(generated_network / "edges.csv") + + assert node_columns == NODE_COLUMNS + assert edge_columns == EDGE_COLUMNS + + node_indices = [int(row["index"]) for row in nodes] + edge_ids = [int(row["uniqueid"]) for row in edges] + assert node_indices == list(range(len(nodes))) + assert edge_ids == list(range(len(edges))) + + node_index_set = set(node_indices) + assert all(int(row["u"]) in node_index_set for row in edges) + assert all(int(row["v"]) in node_index_set for row in edges) + assert all(int(row["u"]) != int(row["v"]) for row in edges) + + osmids = {int(row["osmid"]) for row in nodes} + assert LARGE_OSM_ID in osmids + assert {int(row["osmid_u"]) for row in edges} <= osmids + assert {int(row["osmid_v"]) for row in edges} <= osmids + + +def test_keeps_largest_strong_component_and_removes_parallel_edges( + generated_network: Path, +) -> None: + _, nodes = _read_csv(generated_network / "nodes.csv") + _, edges = _read_csv(generated_network / "edges.csv") + + assert {int(row["osmid"]) for row in nodes} == {LARGE_OSM_ID, 2_002, 3_003} + + endpoint_pairs = [(int(row["u"]), int(row["v"])) for row in edges] + assert len(endpoint_pairs) == len(set(endpoint_pairs)) + + exported = nx.DiGraph() + exported.add_nodes_from(int(row["index"]) for row in nodes) + exported.add_edges_from(endpoint_pairs) + assert nx.is_strongly_connected(exported) + + selected = next( + row + for row in edges + if int(row["osmid_u"]) == LARGE_OSM_ID and int(row["osmid_v"]) == 2_002 + ) + assert float(selected["length"]) == pytest.approx(150.0) + assert float(selected["speed_mph"]) == pytest.approx(50.0, abs=0.05) + + +def test_normalizes_speed_units_lanes_and_numeric_constraints( + generated_network: Path, +) -> None: + _, edges = _read_csv(generated_network / "edges.csv") + + metric_speed_edge = next( + row + for row in edges + if int(row["osmid_u"]) == 2_002 and int(row["osmid_v"]) == 3_003 + ) + assert float(metric_speed_edge["speed_mph"]) == pytest.approx( + 80 / 1.609344, + abs=0.05, + ) + assert int(metric_speed_edge["lanes"]) == 1 + + default_speed_edge = next( + row + for row in edges + if int(row["osmid_u"]) == 3_003 and int(row["osmid_v"]) == LARGE_OSM_ID + ) + assert float(default_speed_edge["speed_mph"]) == pytest.approx( + 40 / 1.609344, + abs=0.05, + ) + # Semicolon-separated OSM lane tags are ambiguous; either adjacent value + # is acceptable, but silently falling back to the CLI default is not. + assert int(default_speed_edge["lanes"]) in {2, 3} + + for row in edges: + assert math.isfinite(float(row["length"])) and float(row["length"]) > 0 + assert int(row["lanes"]) >= 1 + assert math.isfinite(float(row["speed_mph"])) and float(row["speed_mph"]) > 0 + + +def test_export_is_byte_for_byte_deterministic(tmp_path: Path) -> None: + graphml = tmp_path / "source.graphml" + first = tmp_path / "first" + second = tmp_path / "second" + _write_graphml_fixture(graphml) + + _assert_success(_generate_network(graphml, first)) + _assert_success(_generate_network(graphml, second)) + + assert (first / "nodes.csv").read_bytes() == (second / "nodes.csv").read_bytes() + assert (first / "edges.csv").read_bytes() == (second / "edges.csv").read_bytes() + + +def test_generated_network_works_with_existing_demand_and_partition_tools( + generated_network: Path, +) -> None: + demand_result = _run( + DEMAND_GENERATOR, + "--network", + generated_network, + "--num-trips", + 25, + "--start-hour", + 5, + "--end-hour", + 6, + "--model", + "uniform", + "--seed", + 7, + ) + _assert_success(demand_result) + + demand_columns, trips = _read_csv(generated_network / "od_demand.csv") + assert demand_columns == ["dep_time", "origin", "destination"] + assert len(trips) == 25 + assert all(5 * 3600 <= float(row["dep_time"]) < 6 * 3600 for row in trips) + assert all(row["origin"] != row["destination"] for row in trips) + + partition_result = _run( + PARTITIONER, + "--network", + generated_network, + "--num-parts", + 2, + "--method", + "spatial", + ) + _assert_success(partition_result) + + partitions = (generated_network / "partitions.txt").read_text().splitlines() + assert len(partitions) == 3 + assert set(partitions) <= {"0", "1"} diff --git a/tools/generate_network.py b/tools/generate_network.py new file mode 100755 index 0000000..cf0f3a6 --- /dev/null +++ b/tools/generate_network.py @@ -0,0 +1,465 @@ +#!/usr/bin/env python3 +# ruff: noqa: UP045 -- pyproject supports Python 3.8, before PEP 604 types. +"""Generate a current-schema LPSim road network from OpenStreetMap data. + +The exporter accepts a live OSM place/bounding-box query or an offline GraphML +file. It normalizes OSM attributes, reduces the MultiDiGraph to the single-edge +``(u, v)`` representation used by LPSim, assigns dense integer identifiers, and +writes ``nodes.csv`` and ``edges.csv``. +""" + +from __future__ import annotations + +import argparse +import ast +import csv +import math +import re +import sys +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Iterable, Optional, Sequence + +KPH_PER_MPH = 1.609344 +MPS_PER_MPH = 0.44704 +NUMBER_PATTERN = re.compile(r"[-+]?(?:\d+(?:\.\d*)?|\.\d+)") + +NODE_COLUMNS = ["osmid", "x", "y", "ref", "highway", "index"] +EDGE_COLUMNS = [ + "uniqueid", + "osmid_u", + "osmid_v", + "length", + "lanes", + "speed_mph", + "u", + "v", +] + + +@dataclass(frozen=True) +class NormalizedEdge: + """One edge after OSM attributes have been normalized.""" + + osmid_u: int + osmid_v: int + length: float + lanes: int + speed_mph: float + source_key: str + + @property + def free_flow_time_seconds(self) -> float: + return self.length / (self.speed_mph * MPS_PER_MPH) + + @property + def selection_rank(self) -> tuple[float, float, float, str]: + """Deterministic policy for reducing parallel OSM edges.""" + + return ( + self.free_flow_time_seconds, + self.length, + -self.speed_mph, + self.source_key, + ) + + +def _require_networkx() -> Any: + try: + import networkx as nx + except ModuleNotFoundError as error: + raise RuntimeError( + "Network generation dependencies are missing. " + "Install them with: pip install -e '.[network]'" + ) from error + return nx + + +def _require_osmnx() -> Any: + try: + import osmnx as ox + except ModuleNotFoundError as error: + raise RuntimeError( + "Live OpenStreetMap downloads require OSMnx. " + "Install it with: pip install -e '.[network]'" + ) from error + return ox + + +def _flatten_attribute(value: Any) -> list[Any]: + """Turn OSM scalar/list/list-like-string attributes into scalar tokens.""" + + if value is None: + return [] + if isinstance(value, float) and math.isnan(value): + return [] + if isinstance(value, set): + value = sorted(value, key=str) + if isinstance(value, (list, tuple)): + flattened: list[Any] = [] + for item in value: + flattened.extend(_flatten_attribute(item)) + return flattened + if not isinstance(value, str): + return [value] + + text = value.strip() + if not text: + return [] + + if text[:1] in "[(" and text[-1:] in ")]": + try: + parsed = ast.literal_eval(text) + except (SyntaxError, ValueError): + parsed = None + if parsed is not None and parsed != value: + return _flatten_attribute(parsed) + + return [part.strip() for part in re.split(r"[;,|]", text) if part.strip()] + + +def _first_positive_number(value: Any) -> Optional[float]: + for token in _flatten_attribute(value): + match = NUMBER_PATTERN.search(str(token)) + if match is None: + continue + number = float(match.group()) + if math.isfinite(number) and number > 0: + return number + return None + + +def _as_bool(value: Any) -> bool: + if isinstance(value, bool): + return value + if isinstance(value, (int, float)): + return bool(value) + return str(value).strip().lower() in {"1", "true", "yes", "y", "-1"} + + +def _parse_lanes(value: Any, default_lanes: int, oneway: bool) -> int: + explicit_value = _first_positive_number(value) + if explicit_value is None: + return default_lanes + + lane_count = max(1, round(explicit_value)) + if not oneway: + # OSM's lanes tag normally describes both directions on two-way ways. + lane_count = max(1, math.ceil(lane_count / 2)) + return lane_count + + +def _speed_tokens_to_mph(value: Any, default_unit: str) -> list[float]: + speeds: list[float] = [] + for token in _flatten_attribute(value): + text = str(token).strip().lower() + match = NUMBER_PATTERN.search(text) + if match is None: + continue + speed = float(match.group()) + if not math.isfinite(speed) or speed <= 0: + continue + + is_mph = "mph" in text or default_unit == "mph" + speeds.append(speed if is_mph else speed / KPH_PER_MPH) + return speeds + + +def _parse_speed_mph(edge_data: dict[str, Any], default_speed_kph: float) -> float: + candidates = ( + (edge_data.get("speed_mph"), "mph"), + (edge_data.get("speed_kph"), "kph"), + # Per OSM tagging conventions, numeric maxspeed values without a unit + # are kilometres per hour. Explicit "mph" strings remain mph. + (edge_data.get("maxspeed"), "kph"), + ) + for value, default_unit in candidates: + speeds = _speed_tokens_to_mph(value, default_unit) + if speeds: + return sum(speeds) / len(speeds) + return default_speed_kph / KPH_PER_MPH + + +def _stringify_attribute(value: Any) -> str: + values = _flatten_attribute(value) + return ";".join(str(item) for item in values) + + +def _format_float(value: float, decimal_places: int) -> str: + text = f"{value:.{decimal_places}f}".rstrip("0").rstrip(".") + return text if text not in {"", "-0"} else "0" + + +def _load_graphml(path: Path) -> Any: + if not path.is_file(): + raise ValueError(f"GraphML file does not exist: {path}") + + nx = _require_networkx() + try: + graph = nx.read_graphml(path, node_type=int, force_multigraph=True) + except (OSError, ValueError, nx.NetworkXError) as error: + raise ValueError(f"Unable to read GraphML file {path}: {error}") from error + return nx.MultiDiGraph(graph) + + +def _download_graph( + place: Optional[str], + bbox: Optional[Sequence[float]], + network_type: str, +) -> Any: + ox = _require_osmnx() + if place is not None: + return ox.graph.graph_from_place( + place, + network_type=network_type, + simplify=True, + retain_all=True, + ) + if bbox is not None: + return ox.graph.graph_from_bbox( + tuple(bbox), + network_type=network_type, + simplify=True, + retain_all=True, + ) + raise ValueError("one of --place, --bbox, or --graphml is required") + + +def _remove_unusable_edges(graph: Any) -> int: + unusable: list[tuple[Any, Any, Any]] = [] + for source, target, key, data in graph.edges(keys=True, data=True): + length = _first_positive_number(data.get("length")) + if source == target or length is None: + unusable.append((source, target, key)) + continue + data["_lpsim_length"] = length + + graph.remove_edges_from(unusable) + graph.remove_nodes_from(list(_isolated_nodes(graph))) + return len(unusable) + + +def _isolated_nodes(graph: Any) -> Iterable[Any]: + return (node for node, degree in graph.degree() if degree == 0) + + +def _select_component(graph: Any, connectivity: str) -> Any: + nx = _require_networkx() + if graph.number_of_nodes() == 0 or graph.number_of_edges() == 0: + raise ValueError("the source graph contains no usable road edges") + + if connectivity == "all": + return graph.copy() + if connectivity == "strong": + components = nx.strongly_connected_components(graph) + elif connectivity == "weak": + components = nx.weakly_connected_components(graph) + else: + raise ValueError(f"unsupported connectivity mode: {connectivity}") + + largest = max( + components, key=lambda component: (len(component), -min(map(int, component))) + ) + return graph.subgraph(largest).copy() + + +def _validate_node_coordinates(graph: Any) -> None: + for osmid, data in graph.nodes(data=True): + try: + x = float(data["x"]) + y = float(data["y"]) + except (KeyError, TypeError, ValueError) as error: + raise ValueError( + f"node {osmid} is missing numeric x/y coordinates" + ) from error + if not math.isfinite(x) or not math.isfinite(y): + raise ValueError(f"node {osmid} has non-finite x/y coordinates") + + +def _normalized_edges( + graph: Any, + default_speed_kph: float, + default_lanes: int, +) -> list[NormalizedEdge]: + selected: dict[tuple[int, int], NormalizedEdge] = {} + + for source, target, key, data in graph.edges(keys=True, data=True): + edge = NormalizedEdge( + osmid_u=int(source), + osmid_v=int(target), + length=float(data["_lpsim_length"]), + lanes=_parse_lanes( + data.get("lanes"), + default_lanes, + _as_bool(data.get("oneway", True)), + ), + speed_mph=_parse_speed_mph(data, default_speed_kph), + source_key=str(key), + ) + pair = (edge.osmid_u, edge.osmid_v) + previous = selected.get(pair) + if previous is None or edge.selection_rank < previous.selection_rank: + selected[pair] = edge + + return list(selected.values()) + + +def _write_network( + graph: Any, + edges: list[NormalizedEdge], + output: Path, +) -> None: + osmids = sorted((int(node) for node in graph.nodes()), key=int) + node_indices = {osmid: index for index, osmid in enumerate(osmids)} + node_data = {int(node): data for node, data in graph.nodes(data=True)} + + output.mkdir(parents=True, exist_ok=True) + nodes_path = output / "nodes.csv" + edges_path = output / "edges.csv" + + with nodes_path.open("w", newline="", encoding="utf-8") as stream: + writer = csv.DictWriter(stream, fieldnames=NODE_COLUMNS, lineterminator="\n") + writer.writeheader() + for osmid in osmids: + data = node_data[osmid] + writer.writerow( + { + "osmid": osmid, + "x": _format_float(float(data["x"]), 10), + "y": _format_float(float(data["y"]), 10), + "ref": _stringify_attribute(data.get("ref")), + "highway": _stringify_attribute(data.get("highway")), + "index": node_indices[osmid], + } + ) + + ordered_edges = sorted( + edges, + key=lambda edge: ( + node_indices[edge.osmid_u], + node_indices[edge.osmid_v], + edge.selection_rank, + ), + ) + with edges_path.open("w", newline="", encoding="utf-8") as stream: + writer = csv.DictWriter(stream, fieldnames=EDGE_COLUMNS, lineterminator="\n") + writer.writeheader() + for unique_id, edge in enumerate(ordered_edges): + writer.writerow( + { + "uniqueid": unique_id, + "osmid_u": edge.osmid_u, + "osmid_v": edge.osmid_v, + "length": _format_float(edge.length, 3), + "lanes": edge.lanes, + "speed_mph": _format_float(edge.speed_mph, 6), + "u": node_indices[edge.osmid_u], + "v": node_indices[edge.osmid_v], + } + ) + + +def generate_network( + graph: Any, + output: Path, + connectivity: str = "strong", + default_speed_kph: float = 40.0, + default_lanes: int = 1, +) -> tuple[int, int]: + """Normalize a graph and write LPSim CSV files. + + Returns the number of exported nodes and directed edges. + """ + + if not math.isfinite(default_speed_kph) or default_speed_kph <= 0: + raise ValueError("--default-speed-kph must be a positive finite number") + if default_lanes < 1: + raise ValueError("--default-lanes must be at least 1") + + graph = graph.copy() + dropped_edges = _remove_unusable_edges(graph) + graph = _select_component(graph, connectivity) + _validate_node_coordinates(graph) + edges = _normalized_edges(graph, default_speed_kph, default_lanes) + if not edges: + raise ValueError("the selected component contains no exportable road edges") + + _write_network(graph, edges, output) + print( + f"Generated LPSim network: {graph.number_of_nodes():,} nodes, " + f"{len(edges):,} directed edges -> {output}" + ) + if dropped_edges: + print(f"Dropped {dropped_edges:,} self-loop or invalid-length edges") + return graph.number_of_nodes(), len(edges) + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description="Generate current-schema LPSim network CSVs from OpenStreetMap" + ) + source = parser.add_mutually_exclusive_group(required=True) + source.add_argument( + "--place", help="OSM place query, e.g. 'Berkeley, California, USA'" + ) + source.add_argument( + "--bbox", + nargs=4, + type=float, + metavar=("LEFT", "BOTTOM", "RIGHT", "TOP"), + help="OSM bounding box in longitude/latitude order", + ) + source.add_argument("--graphml", type=Path, help="Offline GraphML input file") + + parser.add_argument( + "--output", type=Path, required=True, help="Output network directory" + ) + parser.add_argument( + "--network-type", + default="drive", + help="OSMnx network type for live downloads (default: drive)", + ) + parser.add_argument( + "--connectivity", + choices=["strong", "weak", "all"], + default="strong", + help="Component selection before export (default: strong)", + ) + parser.add_argument( + "--default-speed-kph", + type=float, + default=40.0, + help="Fallback speed for edges without maxspeed (default: 40)", + ) + parser.add_argument( + "--default-lanes", + type=int, + default=1, + help="Fallback per-direction lane count (default: 1)", + ) + return parser + + +def main(argv: Optional[Sequence[str]] = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + + try: + if args.graphml is not None: + graph = _load_graphml(args.graphml) + else: + graph = _download_graph(args.place, args.bbox, args.network_type) + generate_network( + graph, + output=args.output, + connectivity=args.connectivity, + default_speed_kph=args.default_speed_kph, + default_lanes=args.default_lanes, + ) + except (RuntimeError, ValueError) as error: + parser.error(str(error)) + return 0 + + +if __name__ == "__main__": + sys.exit(main())