feat: added location extraction to gbfs process (#1155)

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
@@ -112,7 +117,6 @@ def extract_gbfs_endpoints(
                 )
             except AttributeError:
                 language = None
-            print(language)
             endpoints += GBFSEndpoint.from_dict(feed_match.value, language)
         unique_endpoints = list(
             {
@@ -376,3 +380,52 @@ 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.warning("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/gbfs_validator/tests/test_gbfs_data_processor.py

@@ -68,6 +68,8 @@ def setUp(self):
         )
 
     @with_db_session(db_url=default_db_url)
+    @patch("gbfs_data_processor.create_http_task")
+    @patch("gbfs_data_processor.tasks_v2")
     @patch(
         "gbfs_data_processor.GBFSEndpoint.get_request_metadata",
         side_effect=mock_get_request_metadata,
@@ -76,9 +78,15 @@ def setUp(self):
     @patch("gbfs_data_processor.fetch_gbfs_data", side_effect=mock_fetch_gbfs_data)
     @patch("requests.post")
     @patch("requests.get")
-    @patch.dict(os.environ, {"FEEDS_DATABASE_URL": default_db_url})
+    @patch.dict(
+        os.environ,
+        {
+            "FEEDS_DATABASE_URL": default_db_url,
+            "GOOGLE_APPLICATION_CREDENTIALS": "test",
+        },
+    )
     def test_fetch_gbfs_files(
-        self, _, mock_post, __, mock_cloud_storage_client, ___, db_session
+        self, _, mock_post, __, mock_cloud_storage_client, ___, ____, _____, db_session
     ):
         autodiscovery_url = "http://example.com/gbfs.json"
         # Add GBFS feed to the database

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/tests/test_helpers.py

@@ -206,3 +206,18 @@ def test_download_url_content_failure(self, mock_get):
             "Failed to download",
             "Should raise the correct exception",
         )
+
+    @patch.dict(
+        os.environ,
+        {
+            "GOOGLE_APPLICATION_CREDENTIALS": "test",
+        },
+    )
+    def test_create_http_task(self):
+        from utils import create_http_task
+
+        client = MagicMock()
+        body = b"test"
+        url = "test"
+        create_http_task(client, body, url, "test", "test", "test")
+        client.create_task.assert_called_once()

functions-python/helpers/utils.py

@@ -13,7 +13,6 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 #
-
 import hashlib
 import logging
 import os
@@ -131,3 +130,30 @@ def download_and_get_hash(
         if os.path.exists(file_path):
             os.remove(file_path)
         raise e
+
+
+def create_http_task(
+    client,  # type: tasks_v2.CloudTasksClient
+    body: bytes,
+    url: str,
+    project_id: str,
+    gcp_region: str,
+    queue_name: str,
+) -> None:
+    """Creates a GCP Cloud Task."""
+    from google.cloud import tasks_v2
+
+    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/README.md

@@ -52,20 +52,25 @@ Currently, storing `stops.txt` in GCP is a temporary implementation and may be m
 
 ## 3. `reverse_geolocation_process` Function
 
-This function performs the core reverse geolocation logic. It processes each stop in `stops.txt` and determines its geographic location.
+This function performs the core reverse geolocation logic. It processes location data from GTFS or GBFS feeds to determine their geographic context and stores it accordingly.
 
 ### Parameters:
-- `stable_id`: Identifies the GTFS feed.
-- `dataset_id`: Identifies the dataset being processed.
-- `stops_url`: URL of the `stops.txt` file.
+- `stable_id`: Identifies the feed (GTFS or GBFS).
+- `dataset_id`: Required if `data_type` is not provided or is `gtfs`. Identifies the dataset being processed.
+- `stops_url`: Required if `data_type` is not provided or is `gtfs`. URL of the GTFS `stops.txt` file.
+- `station_information_url`: Required if `data_type` is `gbfs` and `vehicle_status_url` is omitted. URL of the GBFS `station_information.json` file.
+- `vehicle_status_url`: Required if `data_type` is `gbfs` and `station_information_url` is omitted. URL of the GBFS `vehicle_status.json` file.
+- `data_type`: Optional. Specifies the type of data being processed. Can be `gtfs` or `gbfs`. If not provided, the function will attempt to determine the type based on the URLs provided.
 
 ### Processing Steps:
 
-1. **Load Stop Data**  
-   - The function reads `stops.txt` into a Pandas DataFrame, ensuring unique longitude-latitude combinations to avoid redundant processing.
+1. **Load Location Data**  
+   - For GTFS: the function reads `stops.txt` into a Pandas DataFrame, ensuring unique longitude-latitude pairs.  
+   - For GBFS: location data is extracted from `station_information.json` (preferred) or `vehicle_status.json` (fallback), also ensuring uniqueness.
 
-2. **Update Dataset Bounding Box**  
-   - The dataset's bounding box is updated using only the extreme coordinate values, forming a rectangular boundary.
+2. **Updates Bounding Box**  
+   - For GTFS: the bounding box is derived from stop coordinates. The dataset's bounding box is updated in the database.
+   - For GBFS: it’s based on extracted station or vehicle coordinates. No database update is performed. We will use the term `stop` to refer to both GTFS stops and GBFS stations/vehicles.
 
 3. **Check for Previously Processed Stops**  
    - Stops are matched against existing `Stop` entities in PostgreSQL using geographic coordinates (not `stop_id`).
@@ -79,11 +84,17 @@ This function performs the core reverse geolocation logic. It processes each sto
 5. **Store Results in PostgreSQL**  
    - Unique location aggregates are identified, and stop counts per location are recorded.
    - `Location` entities are created based on the extracted administrative hierarchy.
+
 6. **GeoJSON Generation**  
-   - The function creates a **GeoJSON file** containing location aggregates and their corresponding stop counts.  
-   - The file is stored in **GCP Storage** following a consistent path format:  
-     - **`<feed_stable_id>/geolocation.geojson`**  
-   - This file always reflects the **latest dataset** results and is used for **location heatmap visualization on the front end**.
+   - A **GeoJSON file** is created representing the aggregated locations and their counts.  
+   - It is stored in GCS GTFS or GBFS buckets, depending on the data type under:  
+     - **`<stable_id>/geolocation.geojson`**  
+   - The file includes:
+     - Extracted locations,
+     - Timestamp of extraction,
+     - URL used for data extraction.  
+   - This file always reflects the most recent dataset/gbfs version results and powers the **location heatmap visualization** on the front end.
+
 ### Location Mapping:
 - **`country_code` / `country`**: Taken from the `Geopolygon` with an ISO 3166-1 code.
 - **`subdivision_name`**: Derived from the lowest administrative `Geopolygon` with an ISO 3166-2 code, but no ISO 3166-1 code.

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