[converter] ED-318 conversion

README.md

@@ -1 +1,13 @@
 # geospatial-utils
+
+This repository contains a suite of tools to manage geospatial data
+
+## Getting started
+
+To run the tool, install Docker. Then, run the following command: `./geospatial-utils/run_locally.sh --help`
+
+## Tools
+
+### ED-269 to ED-318 converter
+
+A converter to transform ED-269 to ED-318 is provided by using the `convert` command.

geospatial-utils/config.py

@@ -0,0 +1,72 @@
+from __future__ import annotations
+
+from datetime import datetime
+
+from implicitdict import ImplicitDict, StringBasedDateTime
+from uas_standards.eurocae_ed318 import TextShortType
+
+
+class ConverterConfiguration(ImplicitDict):
+    ed318_additions: ED318Additions
+
+
+class ED318Additions(ImplicitDict):
+    default_lang: str
+    provider: list[TextShortType]
+    description: list[TextShortType]
+    technicalLimitation: list[TextShortType]
+    issued: StringBasedDateTime
+    otherGeoid: str
+    feature_collection_bbox: list[float]
+    collection_name: str
+
+
+FOCA = ED318Additions(
+    default_lang="en-GB",
+    provider=[
+        TextShortType(lang="de-CH", text="BAZL"),
+        TextShortType(lang="fr-CH", text="OFAC"),
+        TextShortType(lang="it-CH", text="UFAC"),
+        TextShortType(lang="en-GB", text="FOCA"),
+    ],
+    description=[  # TODO: To validate
+        TextShortType(
+            lang="de-CH",
+            text="Schweizerische UAS Geozones, herausgegeben vom Bundesamt für Zivilluftfahrt (BAZL). Umwandlung aus dem Modell ED-269.",
+        ),
+        TextShortType(
+            lang="fr-CH",
+            text="UAS Geozones suisses publiées par l'Office fédéral de l'aviation civile (OFAC). Conversion à partir du modèle ED-269",
+        ),
+        TextShortType(
+            lang="it-CH",
+            text="Geozones UAS svizzere emesse dall'Ufficio federale dell'aviazione civile (UFAC). Conversione dal modello ED-269",
+        ),
+        TextShortType(
+            lang="en-GB",
+            text="Swiss UAS Geozones issued by the Federal Office of Civil Aviation (FOCA). Conversion from the ED-269 model",
+        ),
+    ],
+    technicalLimitation=[  # TODO: To validate
+        TextShortType(
+            lang="de-CH",
+            text="Der Datensatz entsteht durch die Umwandlung der Originaldaten des ED-269-Modells ins neue ED-318. Für die Umwandlung sind einige Datenänderungen nötig. Diese Datei wurde in INTERLIS 2.4 erstellt.",
+        ),
+        TextShortType(
+            lang="fr-CH",
+            text="Le fichier a été créé en convertissant les données originales du modèle ED-269 dans le nouveau ED-318. La conversion nécessite des modifications des données. Ce fichier a été créé en INTERLIS 2.4",
+        ),
+        TextShortType(
+            lang="it-CH",
+            text="Il dataset è stato creato convertendo i dati originali del modello ED-269 nel nuovo ED-318. Per la conversione alcune modifiche dei dati sono necessarie. Questo file è stato creato in INTERLIS 2.4",
+        ),
+        TextShortType(
+            lang="en-GB",
+            text="The dataset was created by converting the original data from the ED-269 model to the new ED-318. Some data modifications are necessary for conversion. This file was created in INTERLIS 2.4",
+        ),
+    ],
+    issued=StringBasedDateTime(datetime.now()),
+    otherGeoid="CHGeo2004",
+    collection_name="Swiss UAS Geozones according to ED-318 converted from the ED-269 data model",
+    feature_collection_bbox=[2485410.215, 1075268.136, 2833857.724, 1295933.698],
+)

geospatial-utils/convert.py

@@ -0,0 +1,217 @@
+from config import ED318Additions
+from uas_standards.eurocae_ed318 import (
+    Authority,
+    CodeVerticalReferenceType,
+    CodeWeekDayType,
+    CodeZoneReasonType,
+    CodeZoneType,
+    DailyPeriod,
+    DatasetMetadata,
+    ED318Schema,
+    ExtentCircle,
+    Feature,
+    Geometry,
+    GeometryCollection,
+    Point,
+    Polygon,
+    TextShortType,
+    TimePeriod,
+    UASZone,
+    UomDistance,
+    VerticalLayer,
+)
+from uas_standards.eurocae_ed269 import (
+    ApplicableTimePeriod,
+    ED269Schema,
+    HorizontalProjectionType,
+    Reason,
+    Restriction,
+    UASZoneAirspaceVolume,
+    UASZoneAuthority,
+    UomDimensions,
+)
+
+COUNTRY_REGION_MAPPING = {"CHE": 0, "LIE": 27}
+
+
+def _convert_restriction(restriction: Restriction) -> CodeZoneType:
+    if restriction is Restriction.REQ_AUTHORISATION:
+        return CodeZoneType.REQ_AUTHORIZATION
+    return CodeZoneType(restriction)
+
+
+def _convert_uom(uom_dimensions: UomDimensions) -> UomDistance:
+    return UomDistance(uom_dimensions.lower())
+
+
+def _convert_authority(za: UASZoneAuthority, default_lang: str) -> Authority:
+    return Authority(
+        name=[TextShortType(text=za.name, lang=default_lang)] if "name" in za else [],
+        service=[TextShortType(text=za.service, lang=default_lang)]
+        if "service" in za
+        else [],
+        contactName=[TextShortType(text=za.contactName, lang=default_lang)]
+        if "contactName" in za
+        else [],
+        siteURL=za.siteURL if "siteURL" in za else None,
+        email=za.email if "email" in za else None,
+        phone=za.phone if "phone" in za and za.phone else None,
+        purpose=za.purpose if "purpose" in za else None,
+        intervalBefore=za.intervalBefore if "intervalBefore" in za else None,
+    )
+
+
+def _convert_geometry(g: UASZoneAirspaceVolume) -> Geometry:
+    vertical_layer = VerticalLayer(
+        upper=g.upperLimit if "upperLimit" in g else None,
+        upperReference=CodeVerticalReferenceType(g.upperVerticalReference),
+        lower=g.lowerLimit if "lowerLimit" in g else None,
+        lowerReference=CodeVerticalReferenceType(g.lowerVerticalReference),
+        uom=_convert_uom(g.uomDimensions),
+    )
+
+    hp = g.horizontalProjection
+    if hp.type is HorizontalProjectionType.Circle:
+        return Point(
+            coordinates=hp.center if "center" in hp else None,
+            extent=ExtentCircle(radius=hp.radius) if "radius" in hp else None,
+            layer=vertical_layer,
+        )
+
+    return Polygon(
+        type="Polygon",
+        coordinates=hp.coordinates if "coordinates" in hp else None,
+        layer=vertical_layer,
+    )
+
+
+def _convert_reasons(reason: list[Reason] | None) -> list[CodeZoneReasonType] | None:
+    reason_types: list[CodeZoneReasonType] = []
+    for r in reason or []:
+        if r == Reason.FOREIGN_TERRITORY:
+            raise NotImplementedError(
+                "Reason FOREIGN_TERRITORY is not supported yet. (Value inexistent in ED-318)"
+            )
+        reason_types.append(CodeZoneReasonType(r))
+    return reason_types if len(reason_types) > 0 else None
+
+
+def _convert_applicability(a: ApplicableTimePeriod) -> TimePeriod | None:
+    schedule: list[DailyPeriod] = []
+    if "schedule" in a:
+        for d in a.schedule or []:
+            daily_period = DailyPeriod(
+                day=CodeWeekDayType(d.day) if "day" in d else None,
+                startTime=d.startTime if "startTime" in d else None,
+                startEvent=None,
+                endTime=d.endTime if "endTime" in d else None,
+                endEvent=None,
+            )
+
+            if len(daily_period) > 0:
+                schedule.append(daily_period)
+
+    time_period = TimePeriod(
+        startDateTime=a.startDateTime if "startDateTime" in a else None,
+        endDateTime=a.endDateTime if "endDateTime" in a else None,
+        schedule=schedule if len(schedule) > 0 else None,
+    )
+    return time_period if len(time_period) > 0 else None
+
+
+def from_ed269_to_ed318(ed269_data: ED269Schema, config: ED318Additions) -> ED318Schema:
+    """Convert ED269 data to ED318 data.
+    Missing data in the new format is provided as a config."""
+
+    dataset_metadata = DatasetMetadata(
+        validFrom=None,
+        validTo=None,
+        provider=config.provider,
+        description=config.description,
+        issued=config.issued,
+        otherGeoid=config.otherGeoid,
+    )
+
+    features: list[Feature] = []
+    for i, zv in enumerate(ed269_data.features):
+        zone_authority: list[Authority] = []
+        for za in zv.zoneAuthority:
+            zone_authority.append(_convert_authority(za, config.default_lang))
+
+        geometries: list[Geometry] = []
+        for g in zv.geometry:
+            geometries.append(_convert_geometry(g))
+
+        if len(geometries) > 1:
+            geometry = GeometryCollection(
+                type="GeometryCollection", geometries=geometries
+            )
+        elif len(geometries) == 1:
+            geometry = geometries[0]
+        else:
+            raise ValueError(f"No geometry found for geozone {zv.name}")
+
+        limited_applicability = [_convert_applicability(a) for a in zv.applicability]
+
+        # Ensures it is not a table of None since permanent zone may be represented like this.
+        if (
+            sum(
+                [
+                    1 if la is not None and len(la) > 0 else 0
+                    for la in limited_applicability
+                ]
+            )
+            == 0
+        ):
+            limited_applicability = None
+
+        # Ensures the converter accepts either an optional string as specified in the standard
+        # definition or a list of str of 0 or 1 item as provided in the jsonschema in the standard.
+        restriction_conditions: str | None = None
+        if "restrictionConditions" in zv and zv.restrictionConditions is not None:
+            if isinstance(zv.restrictionConditions, dict):
+                if len(zv.restrictionConditions) == 0:
+                    restriction_conditions = None
+                elif len(zv.restrictionConditions) == 1:
+                    restriction_conditions = zv.restrictionConditions[0]
+                else:
+                    raise ValueError("Unexpected array with more than one item.")
+            else:
+                restriction_conditions = str(zv.restrictionConditions)
+
+        feature = Feature(
+            id=str(i),
+            type="Feature",
+            properties=UASZone(
+                identifier=zv.identifier,
+                country=zv.country,
+                name=[TextShortType(text=zv.name, lang=config.default_lang)],
+                type=_convert_restriction(zv.restriction),
+                variant=zv.type,
+                restrictionConditions=restriction_conditions,
+                region=COUNTRY_REGION_MAPPING[zv.country],
+                reason=_convert_reasons(zv.reason) if "reason" in zv else None,
+                otherReasonInfo=[
+                    TextShortType(text=zv.otherReasonInfo, lang=config.default_lang)
+                ],
+                regulationExemption=zv.regulationExemption,
+                message=[TextShortType(text=zv.message, lang=config.default_lang)]
+                if "message" in zv
+                else [],
+                extendedProperties=zv.extendedProperties
+                if "extendedProperties" in zv
+                else None,
+                limitedApplicability=limited_applicability,
+                zoneAuthority=zone_authority,
+                dataSource=None,
+            ),
+            geometry=geometry,
+        )
+
+        features.append(feature)
+
+    return ED318Schema(
+        type="FeatureCollection",
+        metadata=dataset_metadata,
+        features=features,
+    )

geospatial-utils/fileutils/ed269.py

@@ -1,5 +1,6 @@
 import json
 import pathlib
+
 from uas_standards.eurocae_ed269 import ED269Schema
 
 

geospatial-utils/main.py

@@ -1,9 +1,12 @@
 import argparse
 import json
-import pathlib
-import fileutils
 import os
+import pathlib
 import sys
+
+import config
+import convert
+import fileutils
 from fileutils import ed269
 from loguru import logger
 
@@ -39,13 +42,15 @@ def main():
         source = fileutils.get(args.input_url, int(args.ttl))
         logger.debug(f"Local input copy: {source.absolute()}")
 
-        data = ed269.loads(source)
-        # output = pathlib.Path(args.output_file)
-        # output.write_text(json.dumps(data, indent=2))
+        ed269_data = ed269.loads(source)
+        # TODO: Move hard-coded configuration to a json file.
+        logger.warning("Additional data not provided in ED269 is hard-coded with Swiss FOCA information. This will be moved to a configurable file in the near future.")
+        ed318_data = convert.from_ed269_to_ed318(ed269_data, config=config.FOCA)
+        output = pathlib.Path(args.output_file)
+        output.write_text(json.dumps(ed318_data), encoding="utf-8")
+
+        logger.info(f"Successful conversion. ED-318 saved to {output.absolute()}")
 
-        logger.debug(json.dumps(data, indent=2))
-        logger.error("Work in progress. The parser has not been fully implemented yet. The ED269 parser output was printed above for inspection. (debug log level required)")
-        sys.exit(1)
     else:
         parser.print_help()
         sys.exit(1)

geospatial-utils/run_locally.sh

@@ -17,10 +17,6 @@ cd geospatial-utils || exit 1
 make image
 )
 
-# https://stackoverflow.com/a/9057392
-# shellcheck disable=SC2124
-OTHER_ARGS=${@:2}
-
 if [ "$CI" == "true" ]; then
   docker_args=""
 else
@@ -38,4 +34,4 @@ docker run ${docker_args} --name geospatial-utils \
   -v ./geospatial-utils/output:/app/geospatial-utils/output \
   -w /app/geospatial-utils \
   interuss/geospatial-utils \
-  uv run main.py $OTHER_ARGS
+  uv run main.py ${@}

pyproject.toml

@@ -9,7 +9,7 @@ dependencies = [
     "loguru>=0.7.3",
     "requests>=2.32.5",
     "ruff",
-    "uas-standards>=4.0.0",
+    "uas-standards>=4.1.0",
 ]
 
 [tool.ruff]