# Air quality grid (DCAITI / TU Berlin). Paged JSON fetch via url_template.
name: "air_quality_data_download"
description: "Air quality data from the dcaiti Tu Berlin server"
enable: false
class_name: airQualityDataMapper
debug:
  endpoint: air-quality
data_type: dynamic
source:
  fetch: http
  mode: multi
  check_metadata:
    enable: true
    keys: ["last_modified"]
  url: https://werkzeug.dcaiti.tu-berlin.de/fairqberlin/
  multi_fetch:
    enable: true
    strategy: url_template
    url_template: https://werkzeug.dcaiti.tu-berlin.de/fairqberlin/inwt_fairq_cache_skip_{skip}_limit_100000.json.gz
    template_params:
      skip: [0, 100000, 200000, 300000]
  save_local: true
  destination: tmp/dcaiti/airqualityAPI/fairq.gz
  response_type: json.gz
  header:
    Accept:
job:
  trigger:
    type:
      name: interval
      config:
        hours: 8
mapping:
  enable: true
  table_name: air_pollution_grid_mapping
  # One unified block. base_geometry_column default (geometry_25833) is overridden
  # by base_geometry_sql below; enrichment uses geom_25833 (not the default).
  strategy:
    type: idw
    # Inverse-distance-weighted interpolation of the (continuous) pollutant field
    # onto each way. Smooths out the Voronoi steps the old nearest-1 mapping had.
    k: 4
    power: 2
    # ways_base may carry only geometry (4326); fall back to a transform.
    base_geometry_sql: "COALESCE({base_alias}.geometry_25833, ST_Transform({base_alias}.geometry, 25833))"
    enrichment_geometry_column: geom_25833
    distance_alias: nearest_distance_m
    # The k neighbours must share a forecast origin so element i lines up across
    # cells — pin them to the latest forecast_time.
    enrichment_filter_sql: >
      e.no2 IS NOT NULL
      AND e.forecast_time = (SELECT MAX(ee.forecast_time) FROM {enrichment_table} ee)
    value_columns:
      - { name: no2,  type: array }
      - { name: pm10, type: array }
      - { name: pm25, type: array }
  base_table:
    column_name: "air_pollution"
    column_type: Integer
storage:
  persistent: true
  expires_after: 6h
  staging:
    table_name: air_pollution_grid
    table_class: AirPollutionGridStagingTable
  enrichment:
    table_name: air_pollution_grid_enrichment
    table_class: AirPollutionGridEnrichmentTable

# Materialized view for this datasource (replaces mv_configs/mv_air_pollution.yaml).
# One row per way segment (same length as ways_base). no2/pm10/pm25 are float
# arrays; element i maps to forecast_start + (i-1) hours. Values are inverse-
# distance-weighted (IDW, k=4, p=2) interpolations of the k nearest grid cells,
# computed element-wise per hour. Ways with no intersecting grid cells get NULL
# arrays. The Java AirQualityRouteScorer reads this by way_id.
materialized_view:
  name: mv_air_pollution
  description: >
    Per-way air pollution forecast (NO2, PM10, PM2.5) packed as float arrays.
    One row per way segment (same length as ways_base). Array element i maps to
    forecast_start + (i-1) hours. Values are inverse-distance-weighted (IDW)
    interpolations of the k nearest grid cells, computed element-wise per hour.
  refresh:
    mode: normal
  depends_on:
    tables:
      - { name: ways_base }
      - { name: air_pollution_grid_enrichment }
      - { name: air_pollution_grid_mapping }
  definition:
    select_sql: |
      WITH forecast_meta AS (
          SELECT
              timezone('UTC', MIN(forecast_time))                                              AS forecast_start,
              timezone('UTC',
                  MIN(forecast_time) + (MAX(array_length(no2, 1)) - 1) * INTERVAL '1 hour'
              )                                                                                AS forecast_end
          FROM {schema}.air_pollution_grid_enrichment
          WHERE no2 IS NOT NULL
            AND forecast_time IS NOT NULL
      )
      SELECT
          w.id,
          w.way_id,
          w.way_link_index,
          fm.forecast_start,
          fm.forecast_end,
          m.no2,
          m.pm10,
          m.pm25
      FROM {schema}.ways_base w
      CROSS JOIN forecast_meta fm
      LEFT JOIN {schema}.air_pollution_grid_mapping m ON m.way_id = w.id
  indexes:
    - { name: mv_air_pollution_way_id_idx, columns: [way_id] }
    - { name: mv_air_pollution_id_idx, columns: [id], unique: true }