feat: added location extraction to gbfs process

Author: cka-y

functions-python/gbfs_validator/requirements.txt

@@ -20,6 +20,7 @@ google-cloud-pubsub
 google-api-core
 google-cloud-firestore
 google-cloud-datastore
+google-cloud-tasks
 cloudevents~=1.10.1
 
 # Configuration

functions-python/gbfs_validator/src/gbfs_data_processor.py

@@ -5,6 +5,7 @@
 from http import HTTPMethod
 from typing import Dict, Any, Optional, List, Tuple
 from jsonpath_ng import parse
+from google.cloud import tasks_v2
 import requests
 from requests.exceptions import RequestException
 from google.cloud import storage
@@ -16,6 +17,7 @@
     BUCKET_NAME,
     VALIDATOR_URL,
 )
+import os
 from shared.database_gen.sqlacodegen_models import (
     Gbfsnotice,
     Gbfsversion,
@@ -26,6 +28,7 @@
 )
 from sqlalchemy.orm import Session
 from shared.database.database import with_db_session
+from shared.helpers.utils import create_http_task
 
 
 class GBFSDataProcessor:
@@ -53,6 +56,8 @@ def process_gbfs_data(self, autodiscovery_url: str) -> None:
         # Update database entities
         self.update_database_entities()
 
+        self.trigger_location_extraction()
+
     @with_db_session()
     def record_autodiscovery_request(
         self, autodiscovery_url: str, db_session: Session
@@ -376,3 +381,53 @@ def extract_endpoints_for_all_versions(self):
                 self.gbfs_endpoints[version.version] = endpoints
             else:
                 logging.error(f"No endpoints found for version {version.version}.")
+
+    def trigger_location_extraction(self):
+        """Trigger the location extraction process."""
+        latest_version = self.get_latest_version()
+        if not latest_version:
+            logging.error("No latest version found.")
+            return
+        endpoints = self.gbfs_endpoints.get(latest_version, [])
+        # Find the station_information_url endpoint
+        station_information_url = next(
+            (
+                endpoint.url
+                for endpoint in endpoints
+                if endpoint.name == "station_information"
+            ),
+            None,
+        )
+        # If station_information_url is not found, use vehicle_status_url
+        vehicle_status_url = next(
+            (
+                endpoint.url
+                for endpoint in endpoints
+                if endpoint.name == "vehicle_status"
+            ),
+            None,
+        )
+        if not station_information_url and not vehicle_status_url:
+            logging.error("No station_information_url or vehicle_status_url found.")
+            return
+        client = tasks_v2.CloudTasksClient()
+        body = json.dumps(
+            {
+                "stable_id": self.stable_id,
+                "data_type": "gbfs",
+                "station_information_url": station_information_url,
+                "vehicle_status_url": vehicle_status_url
+            }
+        ).encode()
+        project_id = os.getenv("PROJECT_ID")
+        gcp_region = os.getenv("GCP_REGION")
+        queue_name = os.getenv("QUEUE_NAME")
+        create_http_task(
+            client,
+            body,
+            f"https://{gcp_region}-{project_id}.cloudfunctions.net/reverse-geolocation-processor",
+            project_id,
+            gcp_region,
+            queue_name
+        )
+

functions-python/helpers/requirements.txt

@@ -22,6 +22,7 @@ google-cloud-datastore
 cloudevents~=1.10.1
 google-cloud-bigquery
 google-api-core
+google-cloud-tasks
 google-cloud-firestore
 google-cloud-bigquery
 

functions-python/helpers/utils.py

@@ -13,7 +13,7 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 #
-
+from google.cloud import tasks_v2
 import hashlib
 import logging
 import os
@@ -131,3 +131,28 @@ def download_and_get_hash(
         if os.path.exists(file_path):
             os.remove(file_path)
         raise e
+
+
+def create_http_task(
+        client: tasks_v2.CloudTasksClient,
+        body: bytes,
+        url: str,
+        project_id: str,
+        gcp_region: str,
+        queue_name: str,
+) -> None:
+    """Creates a GCP Cloud Task."""
+    task = tasks_v2.Task(
+        http_request=tasks_v2.HttpRequest(
+            url=url,
+            http_method=tasks_v2.HttpMethod.POST,
+            oidc_token=tasks_v2.OidcToken(
+                service_account_email=os.getenv("SERVICE_ACCOUNT_EMAIL")
+            ),
+            body=body,
+            headers={"Content-Type": "application/json"},
+        )
+    )
+    client.create_task(
+        parent=client.queue_path(project_id, gcp_region, queue_name), task=task
+    )

functions-python/reverse_geolocation/requirements.txt

@@ -27,6 +27,7 @@ pycountry
 shapely
 gtfs-kit
 matplotlib
+jsonpath_ng
 
 # Configuration
 python-dotenv==1.0.0

functions-python/reverse_geolocation/src/location_group_utils.py

@@ -1,18 +1,11 @@
-import os
 from typing import List, Optional
 
 import matplotlib.pyplot as plt
 import pycountry
 from geoalchemy2.shape import to_shape
-from google.cloud import tasks_v2
 
 from shared.database_gen.sqlacodegen_models import Geopolygon, Osmlocationgroup
 
-queue_name = os.getenv("QUEUE_NAME")
-project_id = os.getenv("PROJECT_ID")
-gcp_region = os.getenv("GCP_REGION")
-
-
 ERROR_STATUS_CODE = 299  # Custom error code for the function to avoid retries
 
 
@@ -29,24 +22,6 @@ def generate_color(
     return f"rgba({int(rgba[0] * 255)}, {int(rgba[1] * 255)}, {int(rgba[2] * 255)}, {rgba[3]})"
 
 
-def create_http_task(client: tasks_v2.CloudTasksClient, body: bytes, url: str) -> None:
-    """Creates a GCP Cloud Task."""
-    task = tasks_v2.Task(
-        http_request=tasks_v2.HttpRequest(
-            url=url,
-            http_method=tasks_v2.HttpMethod.POST,
-            oidc_token=tasks_v2.OidcToken(
-                service_account_email=os.getenv("SERVICE_ACCOUNT_EMAIL")
-            ),
-            body=body,
-            headers={"Content-Type": "application/json"},
-        )
-    )
-    client.create_task(
-        parent=client.queue_path(project_id, gcp_region, queue_name), task=task
-    )
-
-
 class GeopolygonAggregate:
     """
     A class to represent an aggregate of geopolygon object to represent a location

functions-python/reverse_geolocation/src/parse_request.py

@@ -0,0 +1,140 @@
+import io
+import logging
+from typing import Tuple, Optional
+
+import flask
+import pandas as pd
+import requests
+from jsonpath_ng import parse
+
+
+def parse_request_parameters(
+    request: flask.Request,
+) -> Tuple[pd.DataFrame, str, Optional[str], str, str]:
+    """
+    Parse the request parameters and return a DataFrame with the stops data.
+    @:returns Tuple: A tuple containing the stops DataFrame, stable ID, and dataset ID.
+    """
+    logging.info("Parsing request parameters.")
+    request_json = request.get_json(silent=True)
+    logging.info(f"Request JSON: {request_json}")
+
+    if (
+            not request_json
+            or (
+                ("stops_url" not in request_json or "dataset_id" not in request_json) and
+                "station_information_url" not in request_json and
+                "vehicle_status_url" not in request_json
+            )
+            or "stable_id" not in request_json
+    ):
+        raise ValueError(
+            "Missing required parameters: [stops_url, dataset_id  | station_information_url | vehicle_status_url], "
+            "stable_id."
+        )
+
+    data_type = request_json.get("data_type", "gtfs")
+    logging.info(f"Data type: {data_type}")
+    if data_type == "gtfs":
+        df, stable_id, dataset_id, url = parse_request_parameters_gtfs(request_json)
+    elif data_type == "gbfs":
+        df, stable_id, dataset_id, url = parse_request_parameters_gbfs(request_json)
+    else:
+        raise ValueError(
+            f"Invalid data_type '{data_type}'. Supported types are 'gtfs' and 'gbfs'."
+        )
+    return df, stable_id, dataset_id, data_type, url
+
+
+def parse_request_parameters_gtfs(request_json: dict) -> Tuple[pd.DataFrame, str, Optional[str], str]:
+    """ Parse the request parameters for GTFS data. """
+    if (
+            not request_json
+            or "stops_url" not in request_json
+            or "stable_id" not in request_json
+            or "dataset_id" not in request_json
+    ):
+        raise ValueError(
+            "Invalid request: missing 'stops_url', 'dataset_id' or 'stable_id' parameter."
+        )
+
+    stable_id = request_json["stable_id"]
+    dataset_id = request_json["dataset_id"]
+
+    # Read the stops from the URL
+    try:
+        s = requests.get(request_json["stops_url"]).content
+        stops_df = pd.read_csv(io.StringIO(s.decode("utf-8")))
+    except Exception as e:
+        raise ValueError(
+            f"Error reading stops from URL {request_json['stops_url']}: {e}"
+        )
+    return stops_df, stable_id, dataset_id, request_json["stops_url"]
+
+
+def parse_station_information_url(station_information_url) -> pd.DataFrame:
+    """ Parse the station information URL and return a DataFrame with the stops' data. """
+    response = requests.get(station_information_url)
+    response.raise_for_status()
+    data = response.json()
+
+    lat_expr = parse('data.stations[*].lat')
+    lon_expr = parse('data.stations[*].lon')
+    station_id_expr = parse('data.stations[*].station_id')
+
+    lats = [match.value for match in lat_expr.find(data)]
+    lons = [match.value for match in lon_expr.find(data)]
+    station_ids = [match.value for match in station_id_expr.find(data)]
+
+    stations_info = [
+        {"station_id": sid, "stop_lat": lat, "stop_lon": lon}
+        for sid, lat, lon in zip(station_ids, lats, lons)
+    ]
+    return pd.DataFrame(stations_info)
+
+
+def parse_vehicle_status_url(vehicle_status_url) -> pd.DataFrame:
+    """ Parse the vehicle status URL and return a DataFrame with vehicle_id, lat, and lon. """
+    response = requests.get(vehicle_status_url)
+    response.raise_for_status()
+    data = response.json()
+
+    lat_expr = parse('data.vehicles[*].lat')
+    lon_expr = parse('data.vehicles[*].lon')
+    vehicle_id_expr = parse('data.vehicles[*].vehicle_id')
+
+    lats = [match.value for match in lat_expr.find(data)]
+    lons = [match.value for match in lon_expr.find(data)]
+    vehicle_ids = [match.value for match in vehicle_id_expr.find(data)]
+
+    vehicles_info = [
+        {"vehicle_id": vid, "stop_lat": lat, "stop_lon": lon}
+        for vid, lat, lon in zip(vehicle_ids, lats, lons)
+    ]
+
+    return pd.DataFrame(vehicles_info)
+
+
+def parse_request_parameters_gbfs(request_json: dict) -> Tuple[pd.DataFrame, str, Optional[str], str]:
+    """ Parse the request parameters for GBFS data. """
+    if (
+            not request_json
+            or ("station_information_url" not in request_json and "vehicle_status_url" not in request_json)
+            or "stable_id" not in request_json
+    ):
+        raise ValueError(
+            "Invalid request: missing ['station_information_url' | 'vehicle_status_url'], 'dataset_id' or 'stable_id' "
+            "parameter."
+        )
+
+    stable_id = request_json["stable_id"]
+    station_information_url = request_json.get("station_information_url")
+    vehicle_status_url = request_json.get("vehicle_status_url")
+    if station_information_url:
+        logging.info('Parsing station information URL')
+        stops_df = parse_station_information_url(station_information_url)
+    else:
+        logging.info('Parsing vehicle status URL')
+        stops_df = parse_vehicle_status_url(vehicle_status_url)
+    return stops_df, stable_id, None, station_information_url or vehicle_status_url
+

functions-python/reverse_geolocation/src/reverse_geolocation.py

@@ -8,9 +8,9 @@
 from google.cloud import storage
 from google.cloud import tasks_v2
 
-from location_group_utils import create_http_task, project_id, gcp_region
 from shared.helpers.logger import Logger
 from shared.helpers.parser import jsonify_pubsub
+from shared.helpers.utils import create_http_task
 
 
 def init(request: CloudEvent) -> None:
@@ -110,8 +110,15 @@ def create_http_processor_task(
     body = json.dumps(
         {"stable_id": stable_id, "stops_url": stops_url, "dataset_id": dataset_id}
     ).encode()
+    queue_name = os.getenv("QUEUE_NAME")
+    project_id = os.getenv("PROJECT_ID")
+    gcp_region = os.getenv("GCP_REGION")
+
     create_http_task(
         client,
         body,
         f"https://{gcp_region}-{project_id}.cloudfunctions.net/reverse-geolocation-processor",
+        project_id,
+        gcp_region,
+        queue_name,
     )