# Pleasant-bicycling score per way (local parquet of aggregated metrics).
name: pleasant_bicycling
description: " Get the density of vehicles on the road "
enable: false
class_name: pleasantBicyclingMapper
debug:
  endpoint: pleasant-bicycling
data_type: dynamic
source:
  input: none
  mode: single
  check_metadata:
    enable: true
    keys: ["content_type"]
  file_path: data/pleasant_bicycle/aggregated_metrics.parquet
  fetch: local
  save_local: true
  destination: tmp/pleasant_bicycle/pleasant.csv
  response_type: csv
  multi_fetch:
    enable: false
    strategy: expand_params
job:
  executor: process
  trigger:
    type:
      name: interval
      start_date: 2025-12-21T11:15:00
      config:
        hours: 10
mapping:
  enable: true
  table_name: pleasant_mapping
  strategy:
    # Custom LEFT JOIN LATERAL — every way_base gets a row; unmatched ways
    # receive connection_id = NULL. SQL built in PleasantBicyclingMapper.mapping_db_query().
    type: custom
  base_table:
    column_name: "pleasant_score"
    column_type: Float
storage:
  persistent: true
  # expires_after removed: dataset takes ~50 min to ETL (4M rows); aggressive
  # expiry + container restarts cause a reset loop. Freshness = 10h job interval.
  staging:
    table_name: pleasant_staging
    table_class: PleasantStagingTable
  enrichment:
    table_name: pleasant_enrichment
    table_class: PleasantEnrichmentTable

# Materialized view for this datasource (replaces mv_configs/mv_pleasant.yaml).
# One row per way segment. SPI values are stored as a 24-element float array,
# one entry per hour of the day (index 0 = 00:00-01:00, index 23 = 23:00-24:00).
# Missing hours (no bicycle data for that hour) are stored as -1. refresh defaults
# to `concurrently` (mv_defaults), valid here because of the unique id index.
materialized_view:
  name: mv_pleasant
  description: >
    Per-way hourly SPI (speed performance index) as a 24-element float array.
    spi_hourly[h] is the average speed performance index for hour h (0-23); -1 means no data.
  depends_on:
    tables:
      - { name: ways_base }
      - { name: pleasant_mapping }
      - { name: pleasant_enrichment }
  definition:
    select_sql: |
      WITH hours AS (
          SELECT generate_series(0, 23) AS hour
      )
      SELECT
          w.id,
          w.way_id,
          w.way_link_index,
          array_agg(
              COALESCE(pe.avg_speed_performance_index, -1)
              ORDER BY h.hour
          )                                               AS spi_hourly
      FROM {schema}.ways_base w
      CROSS JOIN hours h
      LEFT JOIN {schema}.pleasant_mapping pm
          ON pm.way_id = w.id
      LEFT JOIN {schema}.pleasant_enrichment pe
          ON pe.connection_id = pm.connection_id
         AND pe.hour          = h.hour
      GROUP BY w.id, w.way_id, w.way_link_index
  indexes:
    - { name: idx_mv_pleasant_id, columns: [id], unique: true }
    - { name: idx_mv_pleasant_way_id, columns: [way_id, way_link_index] }
    - { name: idx_mv_pleasant_way_id_only, columns: [way_id] }