proposed revision of api.py

pipeline/APIClient.py

@@ -0,0 +1,51 @@
+
+import requests 
+
+API_URL = "https://api.data.gov.sg/v1/transport/traffic-images"
+
+class APIClient():
+    def __init__(self, url="No API URL"):
+        self.url = url
+        self.timestamp = None
+        self.api_status = "Unverified"
+        self.metadata = None
+
+        # 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}")
+
+    def extract_image(self, camera_id):
+        # Debugging step to ensure camera_id is correctly parsed
+        #print(f"Searching for camera_id: {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"]:
+                # Debugging step to print the camera_id being checked
+                #print(f"Checking camera_id: {camera['camera_id']}")
+
+                if camera["camera_id"] == str(camera_id):
+                    #print(f"Found camera_id: {camera_id}")
+                    return camera["image"]  # Return the image URL if the camera ID matches
+
+        # If camera ID is not found
+        return f"Camera ID {camera_id} not found."
+
+#Debugging
+#client = APIClient(API_URL)
+#image_url = client.extract_image(1111)
+#print("Camera Image URL:", image_url)
+
+
+
+

pipeline/Congestion.py

@@ -0,0 +1,25 @@
+import APIClient
+import Rating
+
+class Congestion:
+    def __init__(self, congestion_rating=None, captured_on=None):
+        self.congestion_rating = congestion_rating
+        self.captured_on = captured_on
+
+
+    def add_datetime(self, apiclient_instance):
+        self.captured_on = apiclient_instance.timestamp
+
+    def add_congestion_rating(self, rating):
+        self.congestion_rating = rating
+
+    def is_heavy_traffic(self):
+        if self.congestion_rating == None:
+            return "Error: No congestion rating has been set!"
+        elif self.congestion_rating < 0.5:
+            return False
+        else:
+            return True
+
+
+

pipeline/Rating.py

@@ -0,0 +1,14 @@
+class Rating:
+    def __init__(self, rated_on=None, rated_by=None, rating_value=None):
+        self.rated_on = rated_on
+        self.rated_by = rated_by
+        self.rating_value = rating_value
+
+    def add_rating_datetime(self, rated_on):
+        self.rated_on = rated_on
+
+    def add_rater(self, rated_by):
+        self.rated_by = rated_by
+
+    def add_rating_value(self, rating):
+        self.rating_value = rating

pipeline/TrafficImage.py

@@ -0,0 +1,13 @@
+import APIClient
+
+class TrafficImage:
+    def __init__(self, image_url,processed=False, congestion_rating=None):
+        self.image_url = image_url
+        self.processed = processed
+        self.conestion_rating = congestion_rating
+
+    def is_processed(self):
+        self.processed = True
+
+    def add_congestion_rating(self, congestion_rating):
+        self.conestion_rating = congestion_rating