from __future__ import annotations from abc import ABC, abstractmethod from functools import lru_cache from typing import List, Dict, Any import os try: import pandas as pd # type: ignore except Exception: # pragma: no cover pd = None # for type hints only class DataProvider(ABC): """ Abstract provider interface for data access used across modules (SAMI, Sites, etc.). Implementations must live under pxy_de.providers.* and implement these methods. """ # ---------- Common ---------- @abstractmethod def health(self) -> Dict[str, Any]: ... # ---------- SAMI ---------- @abstractmethod def indicator(self, indicator: str, cities: List[str]) -> "pd.DataFrame": """ Return columns: city, value, N (N = population or scale variable) """ ... # ---------- Sites: competition (POIs) ---------- @abstractmethod def denue(self, city: str, business: str) -> "pd.DataFrame": """ Return columns: name, lat, lon, category """ ... # ---------- Sites: demand (population grid) ---------- @abstractmethod def popgrid(self, city: str) -> "pd.DataFrame": """ Return columns: cell_id, lat, lon, pop """ ... # ---------- Optional: city boundary (GeoJSON-like) ---------- @abstractmethod def city_boundary(self, city: str) -> Dict[str, Any]: """ Return a GeoJSON-like dict for city boundary, or {} if not available. """ ... @lru_cache(maxsize=1) def get_provider() -> DataProvider: """ Factory for data providers. Choose via env: DATA_PROVIDER = csv (default) | """ name = os.getenv("DATA_PROVIDER", "csv").strip().lower() if name == "csv": from .csv_provider import CsvDataProvider return CsvDataProvider() # Add more providers here in the future: # elif name == "postgres": from .pg_provider import PgDataProvider; return PgDataProvider(...) # elif name == "bigquery": ... # Fallback from .csv_provider import CsvDataProvider return CsvDataProvider()