Revert "feat(pipeline): Refactor APIClient, Add TrafficImage & Model …

…APIs (#13)"

This reverts commit 3725fde.

pipeline/api.py

@@ -1,51 +1,70 @@
-import requests
-
-
-class APIClient:
-    def __init__(self, url="https://api.data.gov.sg/v1/transport/traffic-images"):
-        self.url = url
-        self.timestamp = None
-        self.api_status = "Unverified"
-        self.camera_id_array = []
-
-        # Get API response
-        response = requests.get(self.url)
-        response_json = response.json()
-        self.metadata = response_json
-
-        # Get and set API status
-        self.api_status = self.metadata["api_info"]["status"]
-
-        # Get and set timestamp
-        self.timestamp = self.metadata["items"][0]["timestamp"]
-
-        print(f"The API status is: {self.api_status}")
-        print(f"The API was called at: {self.timestamp}")
-
-        for item in self.metadata["items"]:
-            for camera in item["cameras"]:
-                self.camera_id_array.append(camera["camera_id"])
-
-    def extract_image(self, camera_id):
-        # Loop through the items and cameras to find the correct camera_id
-        for item in self.metadata["items"]:
-            for camera in item["cameras"]:
-                if camera["camera_id"] == str(camera_id):
-                    return camera[
-                        "image"
-                    ]  # Return the image URL if the camera ID matches
-        # If camera ID is not found
-        raise RuntimeError(f"Camera ID {camera_id} not found.")
-
-    def extract_latlon(self, camera_id):
-        for item in self.metadata["items"]:
-            for camera in item["cameras"]:
-                if camera["camera_id"] == str(camera_id):
-                    longitude = camera["location"]["longitude"]
-                    latitude = camera["location"]["latitude"]
-                    return (
-                        longitude,
-                        latitude,
-                    )  # Return both longitude and latitude as a tuple
-        # If camera ID is not found
-        raise RuntimeError(f"Camera ID {camera_id} not found.")
+#
+# Flowmotion
+# Pipeline
+# Traffic Images API Client
+#
+
+import asyncio
+
+import httpx
+
+from model import Camera, Location
+
+
+class TrafficImageAPI:
+    """Data.gov.sg Traffic Images API Client."""
+
+    API_URL = "https://api.data.gov.sg/v1/transport/traffic-images"
+
+    def __init__(self):
+        self._sync = httpx.Client()
+        self._async = httpx.AsyncClient()
+
+    def get_cameras(self) -> list[Camera]:
+        """Get Traffic Camera metadata from traffic images API endpoint.
+
+        Returns:
+            Parsed traffic camera metadata.
+        """
+        # fetch traffic-images api endpoint
+        response = self._sync.get(TrafficImageAPI.API_URL)
+        response.raise_for_status()
+        meta = response.json()
+        return parse_cameras(meta)
+
+        # parse traffic camera metadata
+
+    def get_images(self, cameras: list[Camera]) -> list[bytes]:
+        """Get Traffic Camera images from given Traffic Cameras.
+
+        Args:
+            cameras:
+                List of traffic cameras to retrieve traffic images from.
+        Returns:
+            List of JPEG image bytes camera captured by each given Camera.
+        """
+
+        async def fetch():
+            responses = [self._async.get(camera.image_url) for camera in cameras]
+            images = [(await r).aread() for r in responses]
+            return await asyncio.gather(*images)
+
+        return asyncio.run(fetch())
+
+
+def parse_cameras(meta: dict) -> list[Camera]:
+    meta = meta["items"][0]
+    retrieved_on = meta["timestamp"]
+    return [
+        Camera(
+            id=c["camera_id"],
+            retrieved_on=retrieved_on,
+            captured_on=c["timestamp"],
+            image_url=c["image"],
+            location=Location(
+                longitude=c["location"]["longitude"],
+                latitude=c["location"]["latitude"],
+            ),
+        )
+        for c in meta["cameras"]
+    ]

pipeline/db.py

@@ -11,7 +11,7 @@
 from google.cloud.firestore import DocumentReference
 from pydantic import BaseModel
 
-from data import to_json_dict
+from model import to_json_dict
 
 
 class DatabaseClient:

pipeline/model.py

@@ -1,22 +1,48 @@
 #
 # Flowmotion
 # Pipeline
-# ML Model
+# Models
 #
 
-from TrafficImage import TrafficImage
+import json
+from datetime import datetime
 
+from pydantic import BaseModel
 
-class Model:
-    def __init__(self):
-        """Initialise ML model for traffic congestion inference"""
-        # TODO: initialise model
-        pass
 
-    def predict(self, images: list[TrafficImage]):
-        """Predict the traffic congestion rating (0.0-1.0) in the given images.
+class Location(BaseModel):
+    """Geolocation consisting of longitude and latitude."""
 
-        Set the predicted congestion rating for each TrafficImage using TrafficImage.set_processed()
-        """
-        # TODO: make prediction and TrafficImage.set_processed()
-        pass
+    longitude: float
+    latitude: float
+
+
+class Rating(BaseModel):
+    """Traffic Congestion rating performed by a model"""
+
+    rated_on: datetime
+    model_id: str
+    value: float
+
+
+class Camera(BaseModel):
+    """Traffic Camera capturing traffic images."""
+
+    id: str
+    image_url: str
+    captured_on: datetime
+    retrieved_on: datetime
+    location: Location
+
+
+class Congestion(BaseModel):
+    """Traffic Congestion data."""
+
+    camera: Camera
+    rating: Rating
+    updated_on: datetime
+
+
+def to_json_dict(model: BaseModel):
+    """Convert given pydantic model into the its JSON dict representation"""
+    return json.loads(model.model_dump_json())

pipeline/pipeline.py

@@ -1,18 +1,8 @@
-from api import APIClient
-from TrafficImage import TrafficImage
+#
+# Flowmotion
+# Pipeline
+#
 
 if __name__ == "__main__":
-    apiclient = APIClient("https://api.data.gov.sg/v1/transport/traffic-images")
-    active_cameraIDs = apiclient.camera_id_array
-    current_traffic_camera_objects = []  # array of TrafficImage objects
-
-    # populating array of TrafficImage objects
-    for camera_id in active_cameraIDs:
-        image_url = apiclient.extract_image(camera_id)
-        longitude, latitude = apiclient.extract_latlon(camera_id)
-        traffic_camera_obj = TrafficImage(
-            camera_id=camera_id, image=image_url, longitude=longitude, latitude=latitude
-        )
-        current_traffic_camera_objects.append(traffic_camera_obj)
-
-    # MLmodel(current_traffic_camera_objects)
+    # TODO: pipeline code starts here
+    pass

pipeline/test_api.py

@@ -9,60 +9,24 @@
 
 import pytest
 
-from api import APIClient
-from TrafficImage import TrafficImage
+from api import TrafficImageAPI, parse_cameras
+from model import Camera
 
 
 @pytest.fixture
-def api() -> APIClient:
-    return APIClient(url="https://api.data.gov.sg/v1/transport/traffic-images")
+def api() -> TrafficImageAPI:
+    return TrafficImageAPI()
 
 
 @pytest.fixture
-def sample_camera_id() -> str:
-    # Load camera ID from a sample JSON file for testing
+def cameras() -> list[Camera]:
     with open(Path(__file__).parent / "resources" / "traffic_images.json") as f:
-        cameras_data = json.loads(f.read())
-        return cameras_data["items"][0]["cameras"][0][
-            "camera_id"
-        ]  # Grab a camera_id from the file
+        return parse_cameras(json.loads(f.read()))
 
 
-def test_get_camera_ids(api: APIClient):
-    assert len(api.camera_id_array) > 0  # Ensure the APIClient fetched camera IDs
+def test_get_cameras(api: TrafficImageAPI):
+    assert len(api.get_cameras()) > 0
 
 
-def test_extract_image(api: APIClient, sample_camera_id: str):
-    image_url = api.extract_image(sample_camera_id)
-    assert isinstance(image_url, str)
-    assert image_url.startswith("http")  # Ensure a valid image URL is returned
-
-
-def test_extract_latlon(api: APIClient, sample_camera_id: str):
-    latlon = api.extract_latlon(sample_camera_id)
-    assert isinstance(latlon, tuple)
-    assert len(latlon) == 2  # Ensure we get a tuple with (longitude, latitude)
-    longitude, latitude = latlon
-    assert isinstance(longitude, float)
-    assert isinstance(latitude, float)
-
-
-def test_traffic_image_class():
-    # Example test for TrafficImage class
-    traffic_image = TrafficImage(
-        image="some_image_url",
-        processed=False,
-        congestion_rating=None,
-        camera_id="camera_123",
-        longitude=103.851959,
-        latitude=1.290270,
-    )
-
-    assert traffic_image.image == "some_image_url"
-    assert traffic_image.camera_id == "camera_123"
-    assert not traffic_image.processed  # Ensure it's not processed initially
-
-    # Simulate processing the image
-    traffic_image.set_processed(0.5)
-    assert traffic_image.processed  # Now it should be processed
-    assert traffic_image.congestion_rating == 0.5
+def test_get_images(api: TrafficImageAPI, cameras: list[Camera]):
+    assert len(api.get_images(cameras[:1])[0]) > 0

pipeline/test_db.py

@@ -16,8 +16,8 @@
 import pytest
 from pydantic import BaseModel
 
-from data import to_json_dict
 from db import DatabaseClient
+from model import to_json_dict
 
 
 class Model(BaseModel):

pipeline/test_data.py → pipeline/test_model.py

@@ -1,6 +1,6 @@
 #
 # Flowmotion
-# Data Models
+# Models
 # Unit Tests
 #
 
@@ -11,7 +11,7 @@
 
 from jsonschema import validate
 
-from data import Camera, Congestion, Location, Rating, to_json_dict
+from model import Camera, Congestion, Location, Rating, to_json_dict
 
 CONGESTION_SCHEMA = Path(__file__).parent.parent / "schema" / "congestion.schema.json"