73 lines
2.1 KiB
Python
73 lines
2.1 KiB
Python
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) | <future providers>
|
|
"""
|
|
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()
|