Create Kuba sync and external table (#4083)

* Create Kuba bucket on staging

* Add KubaHook to communicate with Proxima API

* Add KubaToGCS operator

* Add sync_kuba dag

* Add production kuba bucket

* Fix lint complaints

* Fix terraform Kuba bucket complaint

* Initialize KubaToGCSOperator and fix misspelled name

* Update setting on yml file to fix creation of Kuba external table

---------

Co-authored-by: Erika Pacheco <[email protected]>

Author: ohrite

airflow/.test.env

@@ -22,6 +22,7 @@ CALITP_BUCKET__GTFS_SCHEDULE_UNZIPPED=gs://calitp-staging-pytest
 CALITP_BUCKET__GTFS_SCHEDULE_UNZIPPED_HOURLY=gs://calitp-staging-pytest
 CALITP_BUCKET__GTFS_SCHEDULE_VALIDATION=gs://calitp-staging-pytest
 CALITP_BUCKET__GTFS_SCHEDULE_VALIDATION_HOURLY=gs://calitp-staging-pytest
+CALITP_BUCKET__KUBA=gs://calitp-staging-pytest
 CALITP_BUCKET__LITTLEPAY_PARSED=gs://calitp-staging-pytest
 CALITP_BUCKET__LITTLEPAY_PARSED_V3=gs://calitp-staging-pytest
 CALITP_BUCKET__LITTLEPAY_RAW=gs://calitp-staging-pytest

airflow/dags/create_external_tables/kuba/device_properties.yml

@@ -0,0 +1,200 @@
+operator: operators.ExternalTable
+bucket: "{{ env_var('CALITP_BUCKET__KUBA') }}"
+post_hook: |
+  SELECT *
+  FROM `{{ env_var('GOOGLE_CLOUD_PROJECT') }}`.external_kuba.device_properties
+  LIMIT 1;
+source_objects:
+  - "device_properties/*.jsonl.gz"
+destination_project_dataset_table: "external_kuba.device_properties"
+source_format: NEWLINE_DELIMITED_JSON
+use_bq_client: true
+hive_options:
+  mode: CUSTOM
+  require_partition_filter: false
+  source_uri_prefix: "device_properties/{dt:DATE}/{ts:TIMESTAMP}/"
+schema_fields:
+  - name: device
+    type: RECORD
+    fields:
+      - name: fo_device_logical_id
+        type: STRING
+      - name: fo_device_type
+        type: STRING
+      - name: fo_device_type_model
+        type: STRING
+      - name: fo_device_serial_number
+        type: STRING
+      - name: fo_device_description
+        type: STRING
+      - name: fo_device_location_id
+        type: STRING
+      - name: fo_device_location
+        type: STRING
+      - name: fo_device_last_connection
+        type: STRING
+  - name: device_replicator_info
+    type: RECORD
+    fields:
+      - name: software_version
+        type: STRING
+      - name: software_last_connection
+        type: STRING
+      - name: cd_version
+        type: STRING
+      - name: cd_last_connection
+        type: STRING
+      - name: dataset_version
+        type: STRING
+      - name: dataset_last_connection
+        type: STRING
+      - name: denylist_version
+        type: STRING
+      - name: denylist_last_connection
+        type: STRING
+      - name: acceptlist_version
+        type: STRING
+      - name: acceptlist_last_connection
+        type: STRING
+      - name: binlist_version
+        type: STRING
+      - name: binlist_last_connection
+        type: STRING
+      - name: asset_last_connection
+        type: STRING
+      - name: monitoring_last_connection
+        type: STRING
+      - name: ud_last_transaction_time
+        type: STRING
+  - name: device_monitor_info
+    type: RECORD
+    fields:
+      - name: application__isdisabled
+        type: STRING
+      - name: application__isinservice
+        type: STRING
+      - name: application__servicestatus
+        type: RECORD
+        fields:
+          - name: Rows
+            type: RECORD
+            mode: REPEATED
+            fields:
+              - name: Id
+                type: STRING
+      - name: gps__position
+        type: RECORD
+        fields:
+          - name: altitude
+            type: STRING
+          - name: dateTime
+            type: STRING
+          - name: direction
+            type: STRING
+          - name: gpsFix
+            type: STRING
+          - name: groundSpeed
+            type: STRING
+          - name: hasAltitude
+            type: STRING
+          - name: hasDateTime
+            type: STRING
+          - name: hasDirection
+            type: STRING
+          - name: hasGpsFix
+            type: STRING
+          - name: hasGroundSpeed
+            type: STRING
+          - name: hasLatitude
+            type: STRING
+          - name: hasLongitude
+            type: STRING
+          - name: latitude
+            type: STRING
+          - name: longitude
+            type: STRING
+          - name: numberSatelites
+            type: STRING
+          - name: properties
+            type: RECORD
+            fields:
+              - name: Id
+                type: STRING
+      - name: location__location__servicestatus
+        type: STRING
+      - name: location__location__source
+        type: STRING
+      - name: os__uptime
+        type: STRING
+      - name: smartmedium__emv3000__batterylevel
+        type: STRING
+      - name: ngwifi__gps__apistatus
+        type: RECORD
+        fields:
+          - name: status
+            type: STRING
+      - name: ngwifi__gps__position
+        type: RECORD
+        fields:
+          - name: altitude
+            type: STRING
+          - name: fix
+            type: STRING
+          - name: heading
+            type: STRING
+          - name: latitude
+            type: STRING
+          - name: longitude
+            type: STRING
+          - name: speed
+            type: STRING
+          - name: success
+            type: STRING
+          - name: timestamp
+            type: STRING
+      - name: location__location__info
+        type: RECORD
+        fields:
+          - name: dataid
+            type: STRING
+          - name: dataversion
+            type: STRING
+          - name: locationProviderSource
+            type: STRING
+          - name: type
+            type: STRING
+          - name: properties
+            type: RECORD
+            fields:
+              - name: Id
+                type: STRING
+          - name: current
+            type: RECORD
+            fields:
+              - name: zoneinfo
+                type: RECORD
+                fields:
+                  - name: name
+                    type: STRING
+                  - name: reference
+                    type: STRING
+              - name: stopinfo
+                type: RECORD
+                fields:
+                  - name: abbreviation
+                    type: STRING
+                  - name: farematrixreference
+                    type: STRING
+                  - name: name
+                    type: STRING
+                  - name: reference
+                    type: STRING
+                  - name: shortname
+                    type: STRING
+                  - name: type
+                    type: STRING
+              - name: properties
+                type: RECORD
+                fields:
+                  - name: Id
+                    type: STRING

airflow/dags/sync_kuba/METADATA.yml

@@ -0,0 +1,17 @@
+description: "Capture the Kuba API to GCS"
+schedule_interval: "0 0 * * *"
+tags:
+  - all_gusty_features
+default_args:
+    owner: airflow
+    depends_on_past: False
+    start_date: "2025-07-14"
+    catchup: False
+    email:
+      - "[email protected]"
+    email_on_failure: True
+    pool: default_pool
+    concurrency: 50
+wait_for_defaults:
+    timeout: 3600
+latest_only: True

airflow/dags/sync_kuba/README.md

@@ -0,0 +1,5 @@
+# `sync_kuba`
+
+Type: [Now / Scheduled](https://docs.calitp.org/data-infra/airflow/dags-maintenance.html)
+
+This DAG orchestrates the syncing of raw Kuba data.

airflow/dags/sync_kuba/device_monitoring.yml

@@ -0,0 +1,7 @@
+operator: operators.KubaToGCSOperator
+
+product: device_properties
+endpoint: monitoring/deviceproperties/v1/ForLocations/all
+parameters:
+    location_type: 1
+bucket: "{{ env_var('CALITP_BUCKET__KUBA') }}"

airflow/docker-compose.yaml

@@ -109,6 +109,7 @@ x-airflow-common:
     CALITP_BUCKET__NTD_XLSX_DATA_PRODUCTS__CLEAN: "gs://calitp-staging-ntd-xlsx-products-clean"
     CALITP_BUCKET__NTD_REPORT_VALIDATION: "gs://calitp-staging-ntd-report-validation"
     CALITP_BUCKET__STATE_GEOPORTAL_DATA_PRODUCTS: "gs://calitp-staging-state-geoportal-scrape"
+    CALITP_BUCKET__KUBA: "gs://calitp-staging-kuba"
 
     DBT_TARGET: staging_service_account
 

airflow/plugins/hooks/kuba_hook.py

@@ -0,0 +1,46 @@
+from airflow.hooks.base import BaseHook
+from airflow.hooks.http_hook import HttpHook
+
+
+class KubaHook(BaseHook):
+    _http_conn_id: str
+    _method: str
+    _session_id: str
+
+    def __init__(self, method: str = "GET", http_conn_id: str = "kuba_default"):
+        super().__init__()
+        self._http_conn_id: str = http_conn_id
+        self._method: str = method
+        self._session_id: str = None
+        self._get_connection()
+
+    def _get_connection(self):
+        conn = self.get_connection(self._http_conn_id)
+        data = {
+            "UserName": conn.login,
+            "Password": conn.password,
+            "OperatorIdentifier": int(conn.schema),
+        }
+        headers = {"Content-Type": "application/json"}
+        kuba_api = HttpHook(method="POST", http_conn_id=self._http_conn_id)
+        authenticate_path = "monitoring/authenticate/v1"
+        result = kuba_api.run(
+            endpoint=authenticate_path, headers=headers, json=data
+        ).json()
+        assert result["Error"] is None
+        self._session_id = result["SessionId"]
+
+    def get_headers(self):
+        return {
+            "Content-Type": "application/json",
+            "Accept": "application/json",
+            "Cookie": f"session-id={self._session_id}",
+        }
+
+    def run(self, endpoint, data=None, headers=None):
+        if headers is None:
+            headers = self.get_headers()
+
+        kuba_api = HttpHook(method=self._method, http_conn_id=self._http_conn_id)
+        response = kuba_api.run(endpoint=endpoint, headers=headers, data=data)
+        return response.json()

airflow/plugins/operators/__init__.py

@@ -8,6 +8,7 @@
 from operators.external_table import ExternalTable
 from operators.gtfs_csv_to_jsonl import GtfsGcsToJsonlOperator
 from operators.gtfs_csv_to_jsonl_hourly import GtfsGcsToJsonlOperatorHourly
+from operators.kuba_to_gcs_operator import KubaToGCSOperator
 from operators.littlepay_raw_sync import LittlepayRawSync
 from operators.littlepay_raw_sync_feed_v3 import LittlepayRawSyncV3
 from operators.littlepay_to_jsonl import LittlepayToJSONL