# Hourly weather forecast (DWD via Bright Sky) for the stations in `expand`.
name: "weather_forecast_bright_sky"
description: "get weather forecast for the dwd_stations in params with hourly forecast "
enable: true
class_name: weatherMapper
debug:
  endpoint: weather-forecast
data_type: dynamic
source:
  fetch: http
  mode: multi
  check_metadata:
    enable: false
  url: https://api.brightsky.dev/weather
  multi_fetch:
    enable: true
    strategy: expand_params
    expand:
      dwd_station_id: ["00399", "00403", "00400", "00410", "00420", "00427", "00430", "00433"]
    params:
      date: >
        ${{
        datetime.now(ZoneInfo("Europe/Berlin")).isoformat()
        }}
  save_local: true
  destination: tmp/brightsky/weather-data/weather_forecast.json
  response_type: json
  header:
    Accept: "application/json"
job:
  trigger:
    type:
      name: interval
      start_date: 2025-12-21T11:15:00
      config:
        hours: 10
mapping:
  # No per-way mapping: weather_enrichment is consumed directly by mv_weather.
  enable: false
storage:
  force_create: false
  persistent: true
  expires_after: 1d
  enrichment:
    table_name: weather_enrichment
    table_class: WeatherEnrichmentTable
  staging:
    table_name: weather_staging
    table_class: WeatherStagingTable
    persistent: false

# Materialized view for this datasource (replaces mv_configs/mv_weather.yaml).
# One row per way segment. Wind/visibility forecasts are stored as float arrays,
# one entry per hour of the forecast window; index i = forecast_start + i hours,
# -1 means no data. bearing_deg is the fixed segment travel bearing.
# Only the forecast datasource fires this MV (auto-filled from this datasource);
# the station mapping is read via depends_on.tables. depends_on.datasources is
# kept explicit so the station datasource is also tracked as a dependency.
# refresh defaults to `concurrently` (mv_defaults): one row per way + unique id.
materialized_view:
  name: mv_weather
  description: >
    Per-way hourly wind forecast as float arrays (wind_speed_hourly,
    wind_direction_hourly, visibility_hourly). Index i = forecast_start + i hours;
    -1 means no data. bearing_deg is the fixed segment travel bearing.
  depends_on:
    datasources:
      - weather_forecast_bright_sky
      - weather_station_bright_sky
    tables:
      - { name: weather_enrichment }
      - { name: dwd_station_locations_mapping }
      - { name: ways_base }
  definition:
    select_sql: |
      WITH bounds AS (
          SELECT
              date_trunc('hour', MIN("timestamp")) AS forecast_start,
              date_trunc('hour', MAX("timestamp")) AS forecast_end
          FROM {schema}.weather_enrichment
      ),
      current_slot AS (
          SELECT GREATEST(0, (EXTRACT(EPOCH FROM (
              date_trunc('hour', NOW() AT TIME ZONE 'Europe/Berlin')
              - (SELECT forecast_start FROM bounds)
          )) / 3600)::int) AS idx
      ),
      slots AS (
          SELECT
              gs                                                       AS slot_ts,
              (EXTRACT(EPOCH FROM (gs - b.forecast_start)) / 3600)::int AS slot_idx
          FROM bounds b
          CROSS JOIN generate_series(b.forecast_start, b.forecast_end, INTERVAL '1 hour') AS gs
      )
      SELECT
          w.id,
          w.way_id,
          w.way_link_index,
          m.bearing_degree                       AS bearing_deg,
          (SELECT forecast_start FROM bounds)    AS forecast_start,
          (SELECT forecast_end   FROM bounds)    AS forecast_end,
          agg.wind_speed_hourly,
          agg.wind_direction_hourly,
          agg.visibility_hourly,
          NULLIF(agg.wind_speed_hourly[1 + (SELECT idx FROM current_slot)],     -1) AS wind_speed,
          NULLIF(agg.wind_direction_hourly[1 + (SELECT idx FROM current_slot)], -1) AS wind_direction,
          NULLIF(agg.visibility_hourly[1 + (SELECT idx FROM current_slot)],     -1) AS visibility
      FROM {schema}.ways_base w
      LEFT JOIN {schema}.dwd_station_locations_mapping m
          ON m.way_id = w.id
      LEFT JOIN LATERAL (
          SELECT
              array_agg(COALESCE(e.wind_speed, -1)     ORDER BY s.slot_idx) AS wind_speed_hourly,
              array_agg(COALESCE(e.wind_direction, -1) ORDER BY s.slot_idx) AS wind_direction_hourly,
              array_agg(COALESCE(e.visibility, -1)     ORDER BY s.slot_idx) AS visibility_hourly
          FROM slots s
          LEFT JOIN {schema}.weather_enrichment e
              ON e.dwd_station_id::INTEGER = m.dwd_station_id::INTEGER
             AND date_trunc('hour', e."timestamp") = s.slot_ts
      ) agg ON TRUE
  indexes:
    - { name: idx_mv_weather_id, columns: [id], unique: true }
    - { name: idx_mv_weather_way_id, columns: [way_id, way_link_index] }
    - { name: idx_mv_weather_way_id_only, columns: [way_id] }