""" REST API endpoints for BESS SCADA data and dispatch energy data. """ import json import logging import time from datetime import date, datetime from pathlib import Path logger = logging.getLogger(__name__) from fastapi import APIRouter, HTTPException, Query, Request from fastapi.responses import Response from app.config import ( ANALYTICS_TOKEN, DATA_START_DATE, DISPATCH_START_DATE, FPPMW_CUTOVER_DATE, MAX_DAYS_PER_REQUEST, ) from app.services.aemo_fetcher import AEMOFetchError, FetchResult, fetch_csv_for_date from app.services.analytics import get_stats, get_timing_estimate, log_request from app.services.dispatch_fetcher import DispatchFetchError, fetch_dispatch_csv_for_date from app.services.dispatch_processor import ( DispatchProcessingError, compute_dispatch_summary, filter_and_process_dispatch, to_csv_bytes as dispatch_to_csv_bytes, to_json_records as dispatch_to_json_records, to_parquet_bytes as dispatch_to_parquet_bytes, ) from app.services.gen_info_fetcher import fetch_bess_list from app.services.data_processor import ( DataProcessingError, compute_summary, filter_and_process, to_csv_bytes, to_json_records, to_parquet_bytes, ) router = APIRouter(prefix="/api") DATA_DIR = Path(__file__).parent.parent / "data" def _get_ip(request: Request) -> str: forwarded = request.headers.get("X-Forwarded-For") if forwarded: return forwarded.split(",")[0].strip() return request.client.host if request.client else "unknown" def _parse_date(date_str: str) -> date: try: return datetime.strptime(date_str, "%Y-%m-%d").date() except ValueError: raise HTTPException( status_code=400, detail="Invalid date format. Use YYYY-MM-DD." ) def _source_type(target_date: date) -> str: """'current' for SCADA dates from the Current directory, 'archive' otherwise.""" return "current" if target_date >= FPPMW_CUTOVER_DATE else "archive" # ── BESS list ───────────────────────────────────────────────────────────────── @router.get("/bess") async def get_bess_list(): """ Return in-service battery storage units grouped by state. Response shape: { "states": { "NSW": [...], "VIC": [...], ... }, "source": "live" | "mirror" | "snapshot", "fetched_at": ISO-8601 timestamp, "warnings": list of strings (may be empty) } """ result = await fetch_bess_list() return { "states": result.states, "source": result.source, "fetched_at": result.fetched_at.isoformat(timespec="seconds") + "Z", "warnings": result.warnings, } # ── Quality flags ───────────────────────────────────────────────────────────── @router.get("/quality-flags") def get_quality_flags(): """Return MW_QUALITY_FLAG descriptions.""" flags_file = DATA_DIR / "quality_flags.json" return json.loads(flags_file.read_text()) # ── SCADA data ──────────────────────────────────────────────────────────────── async def _fetch_day(target_date: date, duid: str) -> FetchResult: return await fetch_csv_for_date(target_date, duid) @router.get("/data") async def get_data( request: Request, duid: str = Query(..., description="BESS DUID identifier"), date: str = Query(..., description="Date in YYYY-MM-DD format"), ): """Fetch and return filtered 4-second SCADA data as JSON.""" target_date = _parse_date(date) ip = _get_ip(request) src = _source_type(target_date) t0 = time.monotonic() try: result = await _fetch_day(target_date, duid) df = filter_and_process(result.csv_chunks, duid, target_date) duration_ms = int((time.monotonic() - t0) * 1000) summary = compute_summary(df) records = to_json_records(df) log_request(ip, duid, date, "view", duration_ms=duration_ms, source_type=src) resp = { "duid": duid, "date": date, "total_rows": len(df), "displayed_rows": len(records), "summary": summary, "data": records, } if result.warnings: resp["warnings"] = result.warnings return resp except AEMOFetchError as e: raise HTTPException(status_code=503, detail=str(e)) except DataProcessingError as e: raise HTTPException(status_code=404, detail=str(e)) @router.get("/download/csv") async def download_csv( request: Request, duid: str = Query(...), date: str = Query(...), ): """Download full filtered SCADA data as CSV.""" target_date = _parse_date(date) ip = _get_ip(request) src = _source_type(target_date) t0 = time.monotonic() try: result = await _fetch_day(target_date, duid) df = filter_and_process(result.csv_chunks, duid, target_date) duration_ms = int((time.monotonic() - t0) * 1000) output = to_csv_bytes(df) log_request(ip, duid, date, "download_csv", duration_ms=duration_ms, source_type=src) filename = f"BESS_SCADA_{duid}_{date}.csv" headers = {"Content-Disposition": f'attachment; filename="{filename}"'} if result.warnings: headers["X-Data-Warnings"] = "; ".join(result.warnings) return Response(content=output, media_type="text/csv", headers=headers) except AEMOFetchError as e: raise HTTPException(status_code=503, detail=str(e)) except DataProcessingError as e: raise HTTPException(status_code=404, detail=str(e)) @router.get("/download/parquet") async def download_parquet( request: Request, duid: str = Query(...), date: str = Query(...), ): """Download full filtered SCADA data as Parquet.""" target_date = _parse_date(date) ip = _get_ip(request) src = _source_type(target_date) t0 = time.monotonic() try: result = await _fetch_day(target_date, duid) df = filter_and_process(result.csv_chunks, duid, target_date) duration_ms = int((time.monotonic() - t0) * 1000) output = to_parquet_bytes(df) log_request(ip, duid, date, "download_parquet", duration_ms=duration_ms, source_type=src) filename = f"BESS_SCADA_{duid}_{date}.parquet" headers = {"Content-Disposition": f'attachment; filename="{filename}"'} if result.warnings: headers["X-Data-Warnings"] = "; ".join(result.warnings) return Response(content=output, media_type="application/octet-stream", headers=headers) except AEMOFetchError as e: raise HTTPException(status_code=503, detail=str(e)) except DataProcessingError as e: raise HTTPException(status_code=404, detail=str(e)) # ── Dispatch / energy data ──────────────────────────────────────────────────── @router.get("/energy-data") async def get_energy_data( request: Request, duid: str = Query(..., description="BESS DUID identifier"), date: str = Query(..., description="Date in YYYY-MM-DD format"), ): """Fetch and return filtered 5-minute dispatch energy data as JSON.""" target_date = _parse_date(date) ip = _get_ip(request) t0 = time.monotonic() try: csv_bytes, src = await fetch_dispatch_csv_for_date(target_date) df = filter_and_process_dispatch(csv_bytes, duid, target_date) duration_ms = int((time.monotonic() - t0) * 1000) summary = compute_dispatch_summary(df) records = dispatch_to_json_records(df) log_request(ip, duid, date, "view", duration_ms=duration_ms, source_type=src) return { "duid": duid, "date": date, "total_rows": len(df), "summary": summary, "data": records, } except DispatchFetchError as e: raise HTTPException(status_code=503, detail=str(e)) except DispatchProcessingError as e: raise HTTPException(status_code=404, detail=str(e)) @router.get("/download/energy-csv") async def download_energy_csv( request: Request, duid: str = Query(...), date: str = Query(...), ): """Download full filtered dispatch energy data as CSV.""" target_date = _parse_date(date) ip = _get_ip(request) t0 = time.monotonic() try: csv_bytes, src = await fetch_dispatch_csv_for_date(target_date) df = filter_and_process_dispatch(csv_bytes, duid, target_date) duration_ms = int((time.monotonic() - t0) * 1000) output = dispatch_to_csv_bytes(df) log_request(ip, duid, date, "download_energy_csv", duration_ms=duration_ms, source_type=src) filename = f"BESS_Energy_{duid}_{date}.csv" return Response( content=output, media_type="text/csv", headers={"Content-Disposition": f'attachment; filename="{filename}"'}, ) except DispatchFetchError as e: raise HTTPException(status_code=503, detail=str(e)) except DispatchProcessingError as e: raise HTTPException(status_code=404, detail=str(e)) @router.get("/download/energy-parquet") async def download_energy_parquet( request: Request, duid: str = Query(...), date: str = Query(...), ): """Download full filtered dispatch energy data as Parquet.""" target_date = _parse_date(date) ip = _get_ip(request) t0 = time.monotonic() try: csv_bytes, src = await fetch_dispatch_csv_for_date(target_date) df = filter_and_process_dispatch(csv_bytes, duid, target_date) duration_ms = int((time.monotonic() - t0) * 1000) output = dispatch_to_parquet_bytes(df) log_request(ip, duid, date, "download_energy_parquet", duration_ms=duration_ms, source_type=src) filename = f"BESS_Energy_{duid}_{date}.parquet" return Response( content=output, media_type="application/octet-stream", headers={"Content-Disposition": f'attachment; filename="{filename}"'}, ) except DispatchFetchError as e: raise HTTPException(status_code=503, detail=str(e)) except DispatchProcessingError as e: raise HTTPException(status_code=404, detail=str(e)) # ── Admin ───────────────────────────────────────────────────────────────────── @router.get("/analytics") def analytics(token: str = Query(...)): """Admin-only analytics endpoint.""" if token != ANALYTICS_TOKEN: raise HTTPException(status_code=403, detail="Invalid token.") return get_stats() # ── App info ────────────────────────────────────────────────────────────────── @router.get("/info") def info(): """Return app metadata and timing estimates for the frontend.""" return { "data_start_date": DATA_START_DATE.isoformat(), "dispatch_start_date": DISPATCH_START_DATE.isoformat(), "cutover_date": FPPMW_CUTOVER_DATE.isoformat(), "max_days_per_request": MAX_DAYS_PER_REQUEST, "estimates": { "current": get_timing_estimate("current"), "archive": get_timing_estimate("archive"), "dispatch_current": get_timing_estimate("dispatch_current"), "dispatch_archive": get_timing_estimate("dispatch_archive"), }, }