Skrypt do konwersji pliku JSON ze szlakami PTTK to formatu GeoJSON.

pttk_trails_to_geojson.py

# /// script
# requires-python = ">=3.14"
# dependencies = [
#     "pydantic>=2.13.4,<3",
#     "geojson-pydantic>=2.1.1,<3",
#     "pyreqwest>=0.12.0",
# ]
# ///

from collections import defaultdict
from datetime import timedelta
from pathlib import Path
from typing import Optional

from geojson_pydantic import Feature, FeatureCollection
from geojson_pydantic.geometries import MultiLineString
from pydantic import BaseModel, Field, model_validator
from pyreqwest.client import SyncClientBuilder

MultiLineStringFeature = Feature[MultiLineString, dict]
MultiLineStringFeatureCollection = FeatureCollection[MultiLineStringFeature]


class Trail(BaseModel):
    id: int
    name: Optional[str] = None
    type: str
    color: str
    pk1: str = Field(..., description="Starting point")
    pk2: str = Field(..., description="Ending point")
    content: Optional[str] = None


class Node(BaseModel):
    id: int
    coord: tuple[float, float] | tuple[float, float, float] = Field(
        ..., description="[longitude, latitude] or [longitude, latitude, elevation]"
    )


class Coordinate3D(BaseModel):
    lon: float
    lat: float
    elevation: float

    @model_validator(mode="before")
    @classmethod
    def _from_list(cls, data):
        if isinstance(data, (list, tuple)):
            lon, lat, elevation = data
            return {"lon": lon, "lat": lat, "elevation": elevation}
        return data


class Segment(BaseModel):
    id: int
    ns_id: int = Field(..., description="Node start ID")
    ne_id: int = Field(..., description="Node end ID")
    geom: list[Coordinate3D]


class Relation(BaseModel):
    t_id: int = Field(..., description="Trail ID")
    s_id: int = Field(..., description="Segment ID")
    seq: int = Field(..., description="Sequence number")
    r: int = Field(..., description="Direction flag")


class TrailData(BaseModel):
    trails: list[Trail]
    nodes: list[Node]
    segments: list[Segment]
    relations: list[Relation]


def trails_to_geojson(trail_data: TrailData) -> MultiLineStringFeatureCollection:
    segments_by_id = {segment.id: segment for segment in trail_data.segments}

    relations_by_trail = defaultdict(list)
    for relation in trail_data.relations:
        relations_by_trail[relation.t_id].append(relation)

    features = []
    for trail in trail_data.trails:
        relations = sorted(relations_by_trail[trail.id], key=lambda relation: relation.seq)
        if not relations:
            print(f"Skipping trail: id={trail.id} name={trail.name} due to lack of relations that would connect it to geometry.")
            continue

        lines = []
        for relation in relations:
            segment = segments_by_id[relation.s_id]
            coords = [(coord.lon, coord.lat, coord.elevation) for coord in segment.geom]
            if relation.r == 1:
                coords.reverse()
            lines.append(coords)

        t = trail.model_dump()
        features.append(
            MultiLineStringFeature(
                type="Feature",
                id=t.pop("id"),
                geometry=MultiLineString.create(coordinates=lines),
                properties=t,
            )
        )

    return MultiLineStringFeatureCollection(type="FeatureCollection", features=features)


def download_file(url: str) -> bytes:
    client = SyncClientBuilder().timeout(timeout=timedelta(minutes=3)).error_for_status().build()
    request = client.get(url=url).build()
    response = request.send()
    return response.bytes().to_bytes()


def main(input_file: str, output_file: str) -> None:
    print("Hello from convert2geojson.py!")
    if input_file.startswith("http"):
        print("Downloading file...")
        content = download_file(url=input_file)
    else:
        print("Reading file from disk...")
        content = Path(input_file).read_bytes()
    print("File read. Parsing...")
    trail_data = TrailData.model_validate_json(json_data=content)
    print("Converting to GeoJSON...")
    feature_collection = trails_to_geojson(trail_data=trail_data)
    print("Writing output...")
    Path(output_file).write_text(feature_collection.model_dump_json())


if __name__ == "__main__":
    import sys

    if len(sys.argv) != 3:
        print("Run with exactly two parameters: input path (or url) and output path.")
        sys.exit(1)
        

    main(input_file=sys.argv[1], output_file=sys.argv[2])