Merge pull request #72 from GispoCoding/66-add-export-to-json-script

66 add export to json script

fvh3t/core/gate_layer.py

@@ -67,12 +67,14 @@ def as_line_layer(self, traveler_class: str, start_time: QDateTime, end_time: QD
 
         line_layer.addAttribute(QgsField("fid", QVariant.Int))
         line_layer.addAttribute(QgsField("name", QVariant.String))
-        line_layer.addAttribute(QgsField("traveler_class", QVariant.String))
-        line_layer.addAttribute(QgsField("start_time", QVariant.Time))
-        line_layer.addAttribute(QgsField("end_time", QVariant.Time))
+        line_layer.addAttribute(QgsField("class", QVariant.String))
+        line_layer.addAttribute(QgsField("interval_start", QVariant.DateTime))
+        line_layer.addAttribute(QgsField("interval_end", QVariant.DateTime))
         line_layer.addAttribute(QgsField("counts_negative", QVariant.Bool))
         line_layer.addAttribute(QgsField("counts_positive", QVariant.Bool))
-        line_layer.addAttribute(QgsField("trajectory_count", QVariant.Int))
+        line_layer.addAttribute(QgsField("vehicle_count", QVariant.Int))
+        line_layer.addAttribute(QgsField("speed_avg", QVariant.Double))
+        line_layer.addAttribute(QgsField("acceleration_avg", QVariant.Double))
 
         fields = line_layer.fields()
 
@@ -89,6 +91,8 @@ def as_line_layer(self, traveler_class: str, start_time: QDateTime, end_time: QD
                     gate.counts_negative(),
                     gate.counts_positive(),
                     gate.trajectory_count(),
+                    gate.average_speed(),
+                    gate.average_acceleration(),
                 ]
             )
             feature.setGeometry(gate.geometry())

fvh3t/fvh3t_processing/export_to_json.py

@@ -0,0 +1,103 @@
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from qgis.core import (
+    QgsProcessing,
+    QgsProcessingAlgorithm,
+    QgsProcessingContext,
+    QgsProcessingFeedback,
+    QgsProcessingParameterFileDestination,
+    QgsProcessingParameterVectorLayer,
+)
+from qgis.PyQt.QtCore import QCoreApplication, QDateTime, QVariant
+
+
+class ExportToJSON(QgsProcessingAlgorithm):
+    INPUT_GATES = "INPUT_GATES"
+    OUTPUT_JSON = "OUTPUT_JSON"
+
+    def __init__(self) -> None:
+        super().__init__()
+
+        self._name = "export_to_json"
+        self._display_name = "Export to JSON"
+
+    def tr(self, string) -> str:
+        return QCoreApplication.translate("Processing", string)
+
+    def createInstance(self):  # noqa N802
+        return ExportToJSON()
+
+    def name(self) -> str:
+        return self._name
+
+    def displayName(self) -> str:  # noqa N802
+        return self.tr(self._display_name)
+
+    def initAlgorithm(self, config=None):  # noqa N802
+        self.addParameter(
+            QgsProcessingParameterVectorLayer(
+                name=self.INPUT_GATES,
+                description="Input gates",
+                types=[QgsProcessing.TypeVectorLine],
+            )
+        )
+
+        self.addParameter(
+            QgsProcessingParameterFileDestination(
+                name=self.OUTPUT_JSON,
+                description="Output JSON file",
+                fileFilter="JSON files (*.json)",
+            )
+        )
+
+    def processAlgorithm(  # noqa N802
+        self,
+        parameters: dict[str, Any],
+        context: QgsProcessingContext,
+        feedback: QgsProcessingFeedback,
+    ) -> dict:
+        """
+        Here is where the processing itself takes place.
+        """
+
+        # Initialize feedback if it is None
+        if feedback is None:
+            feedback = QgsProcessingFeedback()
+
+        gate_layer = self.parameterAsVectorLayer(parameters, self.INPUT_GATES, context)
+        output_json_path = self.parameterAsFile(parameters, self.OUTPUT_JSON, context)
+
+        fields_to_exclude = ["fid", "counts_negative", "counts_positive"]
+
+        features_data = []
+
+        for feature in gate_layer.getFeatures():
+            feature_dict = {}
+
+            for field_name, field_value in zip(gate_layer.fields().names(), feature.attributes()):
+                if field_name not in fields_to_exclude:
+                    if field_name == "name":
+                        field_name = "channel"  # noqa: PLW2901
+
+                    # Convert speed km/h -> m/s
+                    value = field_value / 3.6 if field_name == "speed_avg" else field_value
+
+                    # Convert QVariant nulls to None
+                    if isinstance(field_value, QVariant):
+                        value = None if value.isNull() else value.value()
+
+                    # Convert QTime objects to string
+                    if isinstance(field_value, QDateTime):
+                        value = value.toString("yyyy-MM-dd HH-mm-ss")
+
+                    feature_dict[field_name] = value
+
+            features_data.append(feature_dict)
+
+        with open(output_json_path, "w") as json_file:
+            json.dump(features_data, json_file, indent=2)
+
+        return {self.OUTPUT_JSON: output_json_path}

fvh3t/fvh3t_processing/traffic_trajectory_toolkit_provider.py

@@ -1,6 +1,7 @@
 from qgis.core import QgsProcessingProvider
 
 from fvh3t.fvh3t_processing.count_trajectories import CountTrajectories
+from fvh3t.fvh3t_processing.export_to_json import ExportToJSON
 
 
 class TTTProvider(QgsProcessingProvider):
@@ -40,5 +41,5 @@ def loadAlgorithms(self) -> None:  # noqa N802
         """
         Adds individual processing algorithms to the provider.
         """
-        alg = CountTrajectories()
-        self.addAlgorithm(alg)
+        self.addAlgorithm(CountTrajectories())
+        self.addAlgorithm(ExportToJSON())

tests/processing/test_count_trajectories.py

@@ -198,9 +198,9 @@ def test_count_trajectories(
     assert gate2.geometry().asWkt() == "LineString (0 1, 0 2)"
     assert gate3.geometry().asWkt() == "LineString (1 1, 2 1, 2 2)"
 
-    assert gate1.attribute("trajectory_count") == 1
-    assert gate2.attribute("trajectory_count") == 1
-    assert gate3.attribute("trajectory_count") == 2
+    assert gate1.attribute("vehicle_count") == 1
+    assert gate2.attribute("vehicle_count") == 1
+    assert gate3.attribute("vehicle_count") == 2
 
     traj1: QgsFeature = output_trajectories.getFeature(1)
     traj2: QgsFeature = output_trajectories.getFeature(2)
@@ -331,9 +331,9 @@ def test_count_trajectories(
     assert case2gate2.geometry().asWkt() == "LineString (0 1, 0 2)"
     assert case2gate3.geometry().asWkt() == "LineString (1 1, 2 1, 2 2)"
 
-    assert case2gate1.attribute("trajectory_count") == 0
-    assert case2gate2.attribute("trajectory_count") == 1
-    assert case2gate3.attribute("trajectory_count") == 1
+    assert case2gate1.attribute("vehicle_count") == 0
+    assert case2gate2.attribute("vehicle_count") == 1
+    assert case2gate3.attribute("vehicle_count") == 1
 
     case2traj1: QgsFeature = case2_output_trajectories.getFeature(1)
     case2traj2: QgsFeature = case2_output_trajectories.getFeature(2)

tests/processing/test_export_to_json.py

@@ -0,0 +1,66 @@
+try:
+    import processing
+except ImportError:
+    from qgis import processing
+
+import json
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from qgis.core import QgsVectorLayer
+
+from fvh3t.fvh3t_processing.traffic_trajectory_toolkit_provider import TTTProvider
+from tests.processing.test_count_trajectories import (  # noqa: F401
+    input_gate_layer_for_algorithm,
+    input_point_layer_for_algorithm,
+)
+
+
+def test_export_to_json(
+    qgis_app,
+    qgis_processing,  # noqa: ARG001
+    input_point_layer_for_algorithm,  # noqa: F811
+    input_gate_layer_for_algorithm,  # noqa: F811
+):
+    provider = TTTProvider()
+    qgis_app.processingRegistry().removeProvider(provider)
+    qgis_app.processingRegistry().addProvider(provider)
+
+    params = {
+        "INPUT_POINTS": input_point_layer_for_algorithm,
+        "INPUT_LINES": input_gate_layer_for_algorithm,
+        "TRAVELER_CLASS": "car",
+        "START_TIME": None,
+        "END_TIME": None,
+        "OUTPUT_GATES": "TEMPORARY_OUTPUT",
+        "OUTPUT_TRAJECTORIES": "TEMPORARY_OUTPUT",
+    }
+
+    result = processing.run(
+        "traffic_trajectory_toolkit:count_trajectories",
+        params,
+    )
+
+    output_gates: QgsVectorLayer = result["OUTPUT_GATES"]
+
+    params = {
+        "INPUT_GATES": output_gates,
+        "OUTPUT_JSON": "TEMPORARY_OUTPUT",
+    }
+
+    result = processing.run(
+        "traffic_trajectory_toolkit:export_to_json",
+        params,
+    )
+
+    output_json = result["OUTPUT_JSON"]
+
+    with open(output_json) as file:
+        json_data = json.load(file)
+
+    assert len(json_data) == 3  # Number of gates
+    data = json_data[0]
+    assert len(data) == 7  # Number of fields per gate
+
+    keys = {"channel", "class", "interval_start", "interval_end", "vehicle_count", "speed_avg", "acceleration_avg"}
+    assert set(data.keys()) == keys