131 lines
4.2 KiB
Python
131 lines
4.2 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any, Iterable
|
|
|
|
from app.models import GtfsRoute, GtfsStop, OsmFeature, RouteMatch, RoutePattern
|
|
from app.osm_storage import osm_feature_public_id
|
|
|
|
|
|
def feature_collection(features: Iterable[dict[str, Any]]) -> dict[str, Any]:
|
|
return {"type": "FeatureCollection", "features": list(features)}
|
|
|
|
|
|
def gtfs_route_feature(route: GtfsRoute, extra: dict[str, Any] | None = None) -> dict[str, Any] | None:
|
|
if not route.geometry_geojson:
|
|
return None
|
|
props = {
|
|
"id": route.id,
|
|
"dataset_id": route.dataset_id,
|
|
"route_id": route.route_id,
|
|
"mode": route.mode,
|
|
"route_scope": route.route_scope,
|
|
"ref": route.short_name,
|
|
"name": route.long_name,
|
|
"operator": route.operator_name,
|
|
"source": "gtfs",
|
|
}
|
|
if extra:
|
|
props.update(extra)
|
|
return {"type": "Feature", "geometry": json.loads(route.geometry_geojson), "properties": props}
|
|
|
|
|
|
def osm_feature_feature(feature: OsmFeature, extra: dict[str, Any] | None = None) -> dict[str, Any] | None:
|
|
if not feature.geometry_geojson:
|
|
return None
|
|
props = {
|
|
"id": osm_feature_public_id(feature),
|
|
"row_id": feature.id,
|
|
"dataset_id": feature.dataset_id,
|
|
"osm_type": feature.osm_type,
|
|
"osm_id": feature.osm_id,
|
|
"kind": feature.kind,
|
|
"mode": feature.mode,
|
|
"route_scope": feature.route_scope,
|
|
"ref": feature.ref,
|
|
"name": feature.name,
|
|
"operator": feature.operator,
|
|
"network": feature.network,
|
|
"source": "osm",
|
|
}
|
|
if extra:
|
|
props.update(extra)
|
|
return {"type": "Feature", "geometry": json.loads(feature.geometry_geojson), "properties": props}
|
|
|
|
|
|
def route_pattern_feature(pattern: RoutePattern, extra: dict[str, Any] | None = None) -> dict[str, Any] | None:
|
|
if not pattern.geometry_geojson:
|
|
return None
|
|
props = {
|
|
"id": pattern.id,
|
|
"route_pattern_id": pattern.id,
|
|
"route_ref": pattern.route_ref,
|
|
"ref": pattern.route_ref,
|
|
"name": pattern.route_name,
|
|
"mode": pattern.mode,
|
|
"route_scope": pattern.route_scope,
|
|
"operator": pattern.operator_name,
|
|
"source": "route_layer",
|
|
"source_kind": pattern.source_kind,
|
|
"status": pattern.status,
|
|
"confidence": pattern.confidence,
|
|
"osm_feature_id": pattern.osm_feature_id,
|
|
"gtfs_route_id": pattern.gtfs_route_id,
|
|
"gtfs_shape_id": pattern.gtfs_shape_id,
|
|
}
|
|
if extra:
|
|
props.update(extra)
|
|
return {"type": "Feature", "geometry": json.loads(pattern.geometry_geojson), "properties": props}
|
|
|
|
|
|
def gtfs_stop_feature(stop: GtfsStop) -> dict[str, Any] | None:
|
|
if stop.lon is None or stop.lat is None:
|
|
return None
|
|
return {
|
|
"type": "Feature",
|
|
"geometry": {"type": "Point", "coordinates": [stop.lon, stop.lat]},
|
|
"properties": {
|
|
"id": stop.id,
|
|
"dataset_id": stop.dataset_id,
|
|
"stop_id": stop.stop_id,
|
|
"name": stop.name,
|
|
"source": "gtfs",
|
|
},
|
|
}
|
|
|
|
|
|
def match_row(match: RouteMatch) -> dict[str, Any]:
|
|
route = match.gtfs_route
|
|
feature = match.osm_feature
|
|
return {
|
|
"id": match.id,
|
|
"status": match.status,
|
|
"confidence": match.confidence,
|
|
"rule_source": match.rule_source,
|
|
"gtfs": {
|
|
"id": route.id,
|
|
"dataset_id": route.dataset_id,
|
|
"route_id": route.route_id,
|
|
"mode": route.mode,
|
|
"route_scope": route.route_scope,
|
|
"ref": route.short_name,
|
|
"name": route.long_name,
|
|
"operator": route.operator_name,
|
|
},
|
|
"osm": None
|
|
if feature is None
|
|
else {
|
|
"id": feature.id,
|
|
"dataset_id": feature.dataset_id,
|
|
"osm_type": feature.osm_type,
|
|
"osm_id": feature.osm_id,
|
|
"mode": feature.mode,
|
|
"route_scope": feature.route_scope,
|
|
"ref": feature.ref,
|
|
"name": feature.name,
|
|
"operator": feature.operator,
|
|
"network": feature.network,
|
|
},
|
|
"reasons": json.loads(match.reasons_json or "{}"),
|
|
}
|