Skip to content

signals_implementation_plan.md

Latest commit

 

History

History
169 lines (141 loc) · 5.34 KB

signals_implementation_plan.md

File metadata and controls

169 lines (141 loc) · 5.34 KB
Raw

LocalPulse Action Engine - Real Data V0 Implementation Plan

Overview

This document outlines the implementation of the LocalPulse Action Engine with real CSV data sources for traffic and weather/air quality data. The system will compute a Capacity Index (0-100) and provide rule-based action recommendations.

Architecture

graph TD
    A[Traffic CSV] --> B[ingest_csv.py]
    C[Weather/Air CSV] --> B
    B --> D[index_v0.py]
    D --> E[routes_signals.py]
    E --> F[/signals/current endpoint]
    E --> G[/actions endpoint]
    F --> H[Static Demo Page]
    G --> H
Loading

Data Flow

sequenceDiagram
    participant Client
    participant API
    participant Cache
    participant Ingest
    participant Scorer
    
    Client->>API: GET /signals/current
    API->>Cache: Check cache (signals:*)
    alt Cache hit
        Cache-->>API: Return cached data
    else Cache miss
        API->>Ingest: load_traffic_csv()
        API->>Ingest: load_weather_csv()
        Ingest->>Scorer: compute_flow_score()
        Ingest->>Scorer: compute_env_score()
        Scorer->>Scorer: compute_capacity_index()
        Scorer-->>API: Return scores
        API->>Cache: Store in cache
    end
    API-->>Client: JSON response with signals
Loading

Implementation Details

1. packages/signals Module Structure

packages/signals/init.py

  • Module initialization with version info
  • Export key functions for easy import

packages/signals/ingest_csv.py

Functions:

  • load_traffic_csv(path, tz) -> list[dict]
    • Parse CSV with columns: 群組名稱, 統計時間_時, 類別, 數量
    • Normalize to: {group, ts, kind, count}
    • Convert timestamps to ISO8601 with timezone
  • load_weather_csv(path, tz) -> list[dict]
    • Parse CSV with columns: 裝置, 回報時間, 溫度, 濕度, 風速, PM2.5, PM10, 噪音
    • Normalize to: {device, ts, temp_c, rh, wind_ms, pm25, pm10, noise_db}
    • Aggregate to hourly means
  • latest_common_hour(traffic, weather) -> datetime
  • aggregate_hour(traffic, hour) -> dict
    • Apply weights: {PEOPLE:1.0, MOTORCYCLE:0.6, BIKE:0.5, AUTOCAR:0.7, BUS:0.8, TRUCK:0.3}
  • aggregate_weather_hour(weather, hour) -> dict
  • robust_minmax(values: list[float]) -> (lo, hi)
    • 5th/95th percentile with fallback to min/max

packages/signals/index_v0.py

Functions:

  • score_linear(x, lo, hi) -> int (0-100)
  • score_peak(x, lo, opt, hi) -> int (triangular)
  • compute_flow_score(hour_total, history_totals) -> int
  • compute_env_score(weather_data) -> int
    • Components: temp, rh, wind, air quality
  • compute_capacity_index(flow, env) -> int
    • Formula: 0.65flow + 0.35env
  • band(idx) -> str ("low", "mid", "high", "peak")
  • generate_actions(idx, weather_data) -> list[dict]
    • Rule-based actions with overrides for pm25 and wind

2. API Routes Implementation

apps/action-engine-api/app/routes_signals.py

New endpoints:

  • GET /signals/current
    • Returns current capacity index with metrics
    • Response format as specified
  • GET /actions?limit=3
    • Returns top N actions based on current signals
    • Supports limit parameter (1-5)
  • GET /demo
    • Serves static demo page

Environment Variables:

  • TRAFFIC CSV_PATH (default: packages/datasets/traffic_sample.csv)
  • WEATHER_CSV_PATH (default: packages/datasets/weather_sample.csv)
  • TIMEZONE (default: "Asia/Taipei")

3. Sample Data

packages/datasets/traffic_sample.csv

群組名稱,統計時間_時,類別,數量
西門町,2025-10-19 10:00:00,PEOPLE,120
西門町,2025-10-19 10:00:00,MOTORCYCLE,45
西門町,2025-10-19 10:00:00,BIKE,15
...

packages/datasets/weather_sample.csv

裝置,回報時間,溫度,濕度,風速,PM2.5,PM10,噪音
台北站,2025-10-19 10:15:00,24.5,65,2.3,18,25,62.3
...

4. Testing Strategy

Unit Tests (apps/action-engine-api/tests/unit/test_index_v0_logic.py)

  • Test CSV parsing and aggregation
  • Test scoring functions with known inputs/outputs
  • Test action generation rules
  • Test edge cases (empty data, extreme values)

Contract Tests (apps/action-engine-api/tests/contract/test_signals_and_actions_contract.py)

  • Test API response formats
  • Test rate limiting headers
  • Test error handling
  • Test with temporary CSV files

5. Static Demo Page

apps/action-engine-api/app/static/demo.html

  • Single-page HTML with embedded CSS/JS
  • Fetches /signals/current and /actions
  • Displays capacity index with visual indicators
  • Shows top 3 actions with reasons
  • Simple sparkline for historical trend

Implementation Order

  1. Create module structure and CSV ingestion
  2. Implement scoring and action logic
  3. Create API routes with caching
  4. Generate sample CSV files
  5. Write unit tests
  6. Write contract tests
  7. Create demo page
  8. Update documentation

Key Design Decisions

  1. Pure Functions: All data processing functions are pure with no external dependencies
  2. Caching: Leverage existing cache infrastructure with TTL
  3. Error Handling: Graceful degradation when CSV files are missing/invalid
  4. Timezone Handling: All timestamps converted to ISO8601 with proper timezone
  5. Rate Limiting: Reuse existing rate limiting middleware
  6. Testing Strategy: Both unit and contract tests with temporary CSV fixtures

Dependencies

No new third-party dependencies required. Using only existing:

  • pandas (for CSV processing)
  • FastAPI (for API)
  • pytest (for testing)
  • Standard library modules