gisaf-backend/src/gisaf/config.py

from os import environ
import logging
from pathlib import Path
from typing import Any, Type, Tuple

from pydantic_settings import (BaseSettings,
                               PydanticBaseSettingsSource,
                               SettingsConfigDict)
from pydantic import ConfigDict
from pydantic.v1.utils import deep_update
from yaml import safe_load

from gisaf._version import __version__
#from sqlalchemy.ext.asyncio.engine import AsyncEngine
#from sqlalchemy.orm.session import sessionmaker

logger = logging.getLogger(__name__)
ENV = environ.get('env', 'prod')

config_files = [
    Path(Path.cwd().root) / 'etc' / 'gisaf' / ENV,
    Path.home() / '.local' / 'gisaf' / ENV
]

class DashboardHome(BaseSettings):
    title: str
    content_file: str
    footer_file: str

class GisafConfig(BaseSettings):
    title: str
    windowTitle: str
    debugLevel: str
    dashboard_home: DashboardHome
    redirect: str = ''
    use_pretty_errors: bool = False

class SpatialSysRef(BaseSettings):
    author: str
    ellps: str
    k: int
    lat_0: float
    lon_0: float
    no_defs: bool
    proj: str
    towgs84: str
    units: str
    x_0: float
    y_0: float

class RawSurvey(BaseSettings):
    spatial_sys_ref: SpatialSysRef
    srid: int

class Geo(BaseSettings):
    raw_survey: RawSurvey
    simplify_geom_factor: int
    simplify_preserve_topology: bool = False
    srid: int
    srid_for_proj: int

class Flask(BaseSettings):
    secret_key: str
    debug: int

class MQTT(BaseSettings):
    broker: str = 'localhost'
    port: int = 1883

class GisafLive(BaseSettings):
    hostname: str
    port: int
    scheme: str
    redis: str
    mqtt: MQTT

class DefaultSurvey(BaseSettings):
    surveyor_id: int
    equipment_id: int

class Survey(BaseSettings):
    model_config = ConfigDict(extra='ignore')
    db_schema_raw: str
    db_schema: str
    default: DefaultSurvey

class Crypto(BaseSettings):
    secret: str
    algorithm: str
    expire: float

class DB(BaseSettings):
    uri: str
    host: str
    port: int = 5432
    user: str
    db: str
    password: str
    debug: bool
    info: bool
    pool_size: int = 10
    max_overflow: int = 10
    echo: bool = False

    def get_sqla_url(self):
        return f'postgresql+asyncpg://{self.user}:{self.password}@{self.host}:{self.port}/{self.db}'

    def get_pg_url(self):
        return f'postgresql://{self.user}:{self.password}@{self.host}:{self.port}/{self.db}'


class Log(BaseSettings):
    level: str

class OGCAPILicense(BaseSettings):
    name: str
    url: str

class OGCAPIProvider(BaseSettings):
    name: str
    url: str

class OGCAPIServerContact(BaseSettings):
    name: str
    address: str
    city: str
    stateorprovince: str
    postalcode: int
    country: str
    email: str

class OGCAPIIdentification(BaseSettings):
    title: str
    description: str
    keywords: list[str]
    keywords_type: str
    terms_of_service: str
    url: str

class OGCAPIMetadata(BaseSettings):
    identification: OGCAPIIdentification
    license: OGCAPILicense
    provider: OGCAPIProvider
    contact: OGCAPIServerContact

class ServerBind(BaseSettings):
    host: str
    port: int

class OGCAPIServerMap(BaseSettings):
    url: str
    attribution: str

class OGCAPIServer(BaseSettings):
    bind: ServerBind
    url: str
    mimetype: str
    encoding: str
    language: str
    pretty_print: bool
    limit: int
    map: OGCAPIServerMap

class OGCAPI(BaseSettings):
    base_url: str
    bbox: list[float]
    log: Log
    metadata: OGCAPIMetadata
    server: OGCAPIServer

class TileServer(BaseSettings):
     baseDir: str
     useRequestUrl: bool = False
     spriteBaseDir: str
     spriteUrl: str
     spriteBaseUrl: str
     openMapTilesKey: str | None = None

class Map(BaseSettings):
    tileServer: TileServer | None = None
    zoom: int
    pitch: int
    lat: float
    lng: float
    bearing: float
    style: str
    opacity: float
    attribution: str
    status: list[str]
    defaultStatus: list[str] # FIXME: should be str
    tagKeys: list[str]

class Measures(BaseSettings):
    defaultStore: str

class BasketDefault(BaseSettings):
    surveyor: str
    equipment: str
    project: str | None
    status: str
    store: str | None

class BasketOldDef(BaseSettings):
    base_dir: str

class Basket(BaseSettings):
    base_dir: str
    default: BasketDefault

class Plot(BaseSettings):
    maxDataSize: int

class Dashboard(BaseSettings):
    base_source_url: str
    base_storage_dir: str
    base_storage_url: str = '/dashboard-attachment/'

class Widgets(BaseSettings):
    footer: str

class Admin(BaseSettings):
    basket: Basket

class Attachments(BaseSettings):
    base_dir: str

class Job(BaseSettings):
    id: str
    func: str
    trigger: str
    minutes: int | None = 0
    seconds: int | None = 0

class Crs(BaseSettings):
    '''
    Handy definitions for crs-es
    '''
    db: str
    geojson: str
    for_proj: str
    survey: str
    web_mercator: str

class Config(BaseSettings):
    model_config = SettingsConfigDict(
        #env_prefix='gisaf_',
        env_nested_delimiter='__',
    )

    @classmethod
    def settings_customise_sources(
        cls,
        settings_cls: Type[BaseSettings],
        init_settings: PydanticBaseSettingsSource,
        env_settings: PydanticBaseSettingsSource,
        dotenv_settings: PydanticBaseSettingsSource,
        file_secret_settings: PydanticBaseSettingsSource,
    ) -> Tuple[PydanticBaseSettingsSource, ...]:
        return env_settings, init_settings, file_secret_settings, config_file_settings

    # def __init__(self, **kwargs):
    #     super().__init__(**kwargs)
    #     self.crs = {
    #         'db': f'epsg:{conf.srid}',
    #         'geojson': f'epsg:{conf.geojson_srid}',
    #         'for_proj': f'epsg:{conf.srid_for_proj}',
    #         'survey': f'epsg:{conf.raw_survey_srid}',
    #         'web_mercator': 'epsg:3857',
    #     }

    admin: Admin
    attachments: Attachments
    basket: BasketOldDef
    # crs: Crs
    crypto: Crypto
    dashboard: Dashboard
    db: DB
    flask: Flask
    geo: Geo
    gisaf: GisafConfig
    gisaf_live: GisafLive
    jobs: list[Job]
    map: Map
    measures: Measures
    ogcapi: OGCAPI
    plot: Plot
    plugins: dict[str, dict[str, Any]]
    survey: Survey
    version: str
    weather_station: dict[str, dict[str, Any]]
    widgets: Widgets
    #engine: AsyncEngine
    #session_maker: sessionmaker

    @property
    def crs(self) -> Crs:
        return Crs(
            db=f'epsg:{self.geo.srid}',
            geojson=f'epsg:{self.geo.srid}',
            for_proj=f'epsg:{self.geo.srid_for_proj}',
            survey=f'epsg:{self.geo.raw_survey.srid}',
            web_mercator='epsg:3857',
        )

def config_file_settings() -> dict[str, Any]:
    config: dict[str, Any] = {}
    for p in config_files:
        for suffix in {".yaml", ".yml"}:
            path = p.with_suffix(suffix)
            if not path.is_file():
                logger.info(f"No file found at `{path.resolve()}`")
                continue
            logger.debug(f"Reading config file `{path.resolve()}`")
            if path.suffix in {".yaml", ".yml"}:
                config = deep_update(config, load_yaml(path))
            else:
                logger.info(f"Unknown config file extension `{path.suffix}`")
    return config


def load_yaml(path: Path) -> dict[str, Any]:
    with Path(path).open("r") as f:
        config = safe_load(f)
    if not isinstance(config, dict):
        raise TypeError(
            f"Config file has no top-level mapping: {path}"
        )
    return config


conf = Config(version=__version__)

# def set_app_config(app) -> None:
#     raw_configs = []
#     with open(Path(__file__).parent / 'defaults.yml') as cf:
#         raw_configs.append(cf.read())
#     for cf_path in (
#         Path(Path.cwd().root) / 'etc' / 'gisaf' / ENV,
#         Path.home() / '.local' / 'gisaf' / ENV
#     ):
#         try:
#             with open(cf_path.with_suffix('.yml')) as cf:
#                 raw_configs.append(cf.read())
#         except FileNotFoundError:
#             pass

#     yaml_config = safe_load('\n'.join(raw_configs))

#     conf.app = yaml_config['app']
#     conf.postgres = yaml_config['postgres']
#     conf.storage = yaml_config['storage']
#     conf.map = yaml_config['map']
#     conf.security = yaml_config['security']
#     # create_dirs()


# def create_dirs():
#     """
#     Create the directories needed for a proper functioning of the app
#     """
#     ## Avoid circular imports
#     from treetrail.api_v1 import attachment_types
#     for type in attachment_types:
#         base_dir = Path(conf.storage['root_attachment_path']) / type
#         base_dir.mkdir(parents=True, exist_ok=True)
#     logger.info(f'Cache dir: {get_cache_dir()}')
#     get_cache_dir().mkdir(parents=True, exist_ok=True)


# def get_cache_dir() -> Path:
#     return Path(conf.storage['root_cache_path'])