Skip to content

gj_to_cj

Scripts to load icons from GeoJSON and export them to CityJSON.

Functions:

Name Description
load_geojson_icons

Load the icons from a GeoJSON file.

load_geojson_icons(gj_path, output_cj_path)

Load the icons from a GeoJSON file.

Parameters:

Name Type Description Default
gj_path pathlib.Path

Path of GeoJSON file storing the icons.

required
output_cj_path pathlib.Path

CityJSON path to export the icons to.

required

Raises:

Type Description
NotImplementedError

If the data is not a FeatureCollection.

NotImplementedError

If the CRS is not EPSG:7415.

RuntimeError

If a feature does not have a code.

NotImplementedError

If a feature is not a Point.

Source code in python/src/data_pipeline/cj_writing/gj_to_cj.py
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
def load_geojson_icons(gj_path: Path, output_cj_path: Path):
    """
    Load the icons from a GeoJSON file.

    Parameters
    ----------
    gj_path : Path
        Path of GeoJSON file storing the icons.
    output_cj_path : Path
        CityJSON path to export the icons to.

    Raises
    ------
    NotImplementedError
        If the data is not a FeatureCollection.
    NotImplementedError
        If the CRS is not EPSG:7415.
    RuntimeError
        If a feature does not have a code.
    NotImplementedError
        If a feature is not a Point.
    """
    with open(gj_path) as gj_file:
        gj_data: dict[str, Any] = json.load(gj_file)
    if gj_data.get("type", "") != "FeatureCollection":
        raise NotImplementedError(
            f"Only 'FeatureCollection' is supported for now, not '{gj_data.get("type", "")}'"
        )
    if (
        gj_data.get("crs", {}).get("properties", {}).get("name", "")
        != "urn:ogc:def:crs:EPSG::7415"
    ):
        raise NotImplementedError(
            f"Only 'EPSG:7415' is supported for now, not '{gj_data.get("crs", {})}'"
        )

    prefix = "Outdoor"
    outdoor_container = OutdoorObject(prefix=prefix)

    all_units: dict[str, list[OutdoorUnit]] = defaultdict(lambda: [])
    for feature in gj_data.get("features", []):
        # Extract the code
        attributes = feature["properties"]
        code_column = "Type Code [str]"
        if code_column not in attributes:
            raise RuntimeError(
                "At least one feature is missing the attribute 'code_column'"
            )
        unit_code = attributes[code_column]
        if unit_code is None:
            continue

        # Compute the id
        current_units_same_code = len(all_units[unit_code])
        unit_id = OutdoorUnit.unit_code_to_id(
            code=unit_code, prefix=prefix, number=current_units_same_code
        )

        # Get the icon position
        geometry = feature["geometry"]
        if geometry.get("type", "") != "Point":
            raise NotImplementedError(
                f"Only the 'Point' geometry is supported, not '{geometry.get("type", "")}"
            )

        icon_coordinates: list[float] = geometry["coordinates"]
        if len(geometry) == 2:
            icon_coordinates.append(0)
        icon_position = IconPosition.from_list(icon_coordinates)

        attributes = {}
        for key, value in feature["properties"].items():
            if key.endswith("]"):
                col_name, col_value = csv_get_row_value(row={key: value}, column=key)
                attributes[col_name] = col_value

        # Create the outdoor unit
        unit = OutdoorUnit(
            cj_key=unit_id,
            unit_code=unit_code,
            attributes=attributes,
            icon_position=icon_position,
        )
        all_units[unit_code].append(unit)

    # Create the unit containers
    unit_containers: list[OutdoorUnitContainer] = []
    for code, units in all_units.items():
        unit_container_id = OutdoorUnitContainer.unit_code_to_cj_key(
            code=code, prefix=prefix
        )
        unit_container = OutdoorUnitContainer(
            cj_key=unit_container_id, unit_code=code, attributes={}
        )
        unit_containers.append(unit_container)

        CityJSONObject.add_parent_child(parent=outdoor_container, child=unit_container)
        for unit in units:
            CityJSONObject.add_parent_child(parent=unit_container, child=unit)

    cj_file = CityJSONFile(
        scale=np.array([0.00001, 0.00001, 0.00001], dtype=np.float64),
        translate=np.array([0, 0, 0], dtype=np.float64),
    )

    cj_file.add_cityjson_objects([outdoor_container])
    cj_file.add_cityjson_objects(unit_containers)
    cj_file.add_cityjson_objects(
        [unit for units in all_units.values() for unit in units]
    )

    # Check the correctness of the hierarchy
    cj_file.check_objects_hierarchy()

    # Write to CityJSON
    output_cj_path.parent.mkdir(parents=True, exist_ok=True)
    file_json = cj_file.to_json()
    with open(output_cj_path, "w") as f:
        f.write(file_json)