420 lines
11 KiB
Python
420 lines
11 KiB
Python
from os import environ
|
|
import logging
|
|
from pathlib import Path
|
|
from typing import Any, Type, Tuple
|
|
from yaml import safe_load
|
|
|
|
from xdg import BaseDirectory
|
|
from pydantic_settings import (
|
|
BaseSettings,
|
|
PydanticBaseSettingsSource,
|
|
SettingsConfigDict,
|
|
)
|
|
from pydantic.v1.utils import deep_update
|
|
|
|
from importlib.metadata import version
|
|
|
|
logger = logging.getLogger(__name__)
|
|
ENV = environ.get("env", "prod")
|
|
|
|
config_files = [
|
|
Path(Path.cwd().root) / "etc" / "gisaf" / ENV,
|
|
Path(BaseDirectory.xdg_config_home) / "gisaf" / ENV,
|
|
]
|
|
|
|
|
|
class DashboardHome(BaseSettings):
|
|
title: str = "Gisaf - home/dashboards"
|
|
content_file: Path = (
|
|
Path(Path.cwd().root) / "etc" / "gisaf" / "dashboard_home_content.html"
|
|
)
|
|
footer_file: Path = (
|
|
Path(Path.cwd().root) / "etc" / "gisaf" / "dashboard_home_footer.html"
|
|
)
|
|
|
|
|
|
class GisafConfig(BaseSettings):
|
|
title: str = "Gisaf"
|
|
windowTitle: str = "Gisaf"
|
|
debugLevel: str = "INFO"
|
|
dashboard_home: DashboardHome = DashboardHome()
|
|
redirect: str = ""
|
|
use_pretty_errors: bool = False
|
|
|
|
|
|
class SpatialSysRef(BaseSettings):
|
|
author: str = "AVSM"
|
|
ellps: str = "WGS84"
|
|
k: int = 1
|
|
lat_0: float = 12.01605433
|
|
lon_0: float = 79.80998934
|
|
no_defs: bool = True
|
|
proj: str = "tmerc"
|
|
towgs84: str = "0,0,0,0,0,0,0"
|
|
units: str = "m"
|
|
x_0: float = 370455.630
|
|
y_0: float = 1328608.994
|
|
|
|
|
|
class RawSurvey(BaseSettings):
|
|
spatial_sys_ref: SpatialSysRef = SpatialSysRef()
|
|
srid: int = 910001
|
|
|
|
|
|
class Geo(BaseSettings):
|
|
raw_survey: RawSurvey = RawSurvey()
|
|
simplify_geom_factor: int = 10000000
|
|
simplify_preserve_topology: bool = False
|
|
srid: int = 4326
|
|
srid_for_proj: int = 32644
|
|
|
|
|
|
# class Flask(BaseSettings):
|
|
# secret_key: str
|
|
# debug: int
|
|
|
|
|
|
class MQTT(BaseSettings):
|
|
broker: str = "localhost"
|
|
port: int = 1883
|
|
|
|
|
|
class GisafLive(BaseSettings):
|
|
hostname: str = "localhost"
|
|
port: int = 80
|
|
scheme: str = "http"
|
|
redis: str = "redis://localhost"
|
|
mqtt: MQTT = MQTT()
|
|
|
|
|
|
class DefaultSurvey(BaseSettings):
|
|
surveyor_id: int = 1
|
|
equipment_id: int = 1
|
|
|
|
|
|
class Survey(BaseSettings):
|
|
db_schema_raw: str = "raw_survey"
|
|
db_schema: str = "survey"
|
|
default: DefaultSurvey = DefaultSurvey()
|
|
|
|
|
|
class Crypto(BaseSettings):
|
|
secret: str = "Gisaf big secret"
|
|
algorithm: str = "HS256"
|
|
expire: float = 21600
|
|
|
|
|
|
class DB(BaseSettings):
|
|
# uri: str
|
|
host: str = "localhost"
|
|
port: int = 5432
|
|
user: str = "gisaf"
|
|
db: str = "gisaf"
|
|
password: str = "secret"
|
|
debug: bool = False
|
|
info: bool = True
|
|
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 = "WARNING"
|
|
|
|
|
|
class OGCAPILicense(BaseSettings):
|
|
name: str = "CC-BY 4.0 license"
|
|
url: str = "https://creativecommons.org/licenses/by/4.0/"
|
|
|
|
|
|
class OGCAPIProvider(BaseSettings):
|
|
name: str = "Organization Name"
|
|
url: str = "https://pygeoapi.io"
|
|
|
|
|
|
class OGCAPIServerContact(BaseSettings):
|
|
name: str = "Lastname, Firstname"
|
|
position: str = "Position Title"
|
|
address: str = "Mailing Address"
|
|
city: str = "City"
|
|
stateorprovince: str = "Administrative Area"
|
|
postalcode: int = 0
|
|
country: str = "Country"
|
|
email: str = "you@example.org"
|
|
url: str | None = None
|
|
|
|
|
|
class OGCAPIIdentification(BaseSettings):
|
|
title: str = "pygeoapi default instance"
|
|
description: str = "pygeoapi provides an API to geospatial data"
|
|
keywords: list[str] = ["geospatial", "data", "api"]
|
|
keywords_type: str = "theme"
|
|
terms_of_service: str = "https://creativecommons.org/licenses/by/4.0/"
|
|
url: str = "http://example.org"
|
|
|
|
|
|
class OGCAPIMetadata(BaseSettings):
|
|
identification: OGCAPIIdentification = OGCAPIIdentification()
|
|
license: OGCAPILicense = OGCAPILicense()
|
|
provider: OGCAPIProvider = OGCAPIProvider()
|
|
contact: OGCAPIServerContact = OGCAPIServerContact()
|
|
|
|
|
|
class ServerBind(BaseSettings):
|
|
host: str = "0.0.0.0"
|
|
port: int = 5000
|
|
|
|
|
|
class OGCAPIServerMap(BaseSettings):
|
|
url: str = "https://maps.wikimedia.org/osm-intl/{z}/{x}/{y}.png"
|
|
attribution: str = (
|
|
"""<a href="https://wikimediafoundation.org/wiki/Maps_Terms_of_Use">Wikimedia maps</a> | Map data © <a href="https://openstreetmap.org/copyright">OpenStreetMap contributors</a>"""
|
|
)
|
|
|
|
|
|
class OGCAPIServer(BaseSettings):
|
|
bind: ServerBind = ServerBind()
|
|
url: str = "https://example.org/ogcapi"
|
|
mimetype: str = "application/json; charset=UTF-8"
|
|
encoding: str = "utf-8"
|
|
language: str = "en-US"
|
|
pretty_print: bool = False
|
|
limit: int = 1000
|
|
map: OGCAPIServerMap = OGCAPIServerMap()
|
|
|
|
|
|
class OGCAPI(BaseSettings):
|
|
base_url: str = "http://example.org/ogcapi"
|
|
bbox: list[float] = [-180, -90, 180, 90]
|
|
log: Log = Log()
|
|
metadata: OGCAPIMetadata = OGCAPIMetadata()
|
|
server: OGCAPIServer = OGCAPIServer()
|
|
|
|
|
|
class TileServer(BaseSettings):
|
|
baseDir: Path = Path(BaseDirectory.xdg_data_home) / "gisaf" / "mbtiles_files_dir"
|
|
useRequestUrl: bool = False
|
|
spriteBaseDir: Path = (
|
|
Path(BaseDirectory.xdg_data_home) / "gisaf" / "mbtiles_sprites_dir"
|
|
)
|
|
spriteUrl: str = "/tiles/sprite/sprite"
|
|
spriteBaseUrl: str = "https://gisaf.example.org"
|
|
openMapTilesKey: str | None = None
|
|
|
|
def __init__(self, *args, **kwargs) -> None:
|
|
super().__init__(*args, **kwargs)
|
|
self.baseDir.mkdir(parents=True, exist_ok=True)
|
|
self.spriteBaseDir.mkdir(parents=True, exist_ok=True)
|
|
|
|
|
|
class Map(BaseSettings):
|
|
tileServer: TileServer = TileServer()
|
|
zoom: int = 14
|
|
pitch: int = 45
|
|
lat: float = 12.0000
|
|
lng: float = 79.8106
|
|
bearing: float = 0
|
|
style: str = "OpenFreeMap"
|
|
opacity: float = 1
|
|
attribution: str = ""
|
|
status: list[str] = ["E", "F", "D"]
|
|
defaultStatus: list[str] = ["E"] # FIXME: should be str
|
|
tagKeys: list[str] = ["source"]
|
|
|
|
|
|
class Measures(BaseSettings):
|
|
defaultStore: str | None = None
|
|
|
|
|
|
class BasketDefault(BaseSettings):
|
|
surveyor: str = "Default surveyor"
|
|
equipment: str = "Default equipment"
|
|
project: str = "Default project"
|
|
status: str = "E"
|
|
store: str | None = None
|
|
|
|
|
|
# class BasketOldDef(BaseSettings):
|
|
# base_dir: str
|
|
|
|
|
|
class Basket(BaseSettings):
|
|
base_dir: str = "/var/local/gisaf/baskets"
|
|
default: BasketDefault = BasketDefault()
|
|
|
|
|
|
class Plot(BaseSettings):
|
|
maxDataSize: int = 10000
|
|
|
|
|
|
class Dashboard(BaseSettings):
|
|
base_source_url: str = "http://url.to.jupyter/lab/tree/"
|
|
base_storage_dir: str = "/var/lib/share/gisaf/dashboard"
|
|
base_storage_url: str = "/dashboard-attachment/"
|
|
|
|
|
|
class Widgets(BaseSettings):
|
|
footer: str = (
|
|
"""Generated by <span class='link' onclick="window.open('https://redmine.auroville.org.in/projects/gisaf/')">Gisaf</span>"""
|
|
)
|
|
|
|
|
|
class Admin(BaseSettings):
|
|
basket: Basket = Basket()
|
|
|
|
|
|
class Attachments(BaseSettings):
|
|
base_dir: str = "/var/local/gisaf/attachments"
|
|
|
|
|
|
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 # type: ignore
|
|
|
|
# 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 = Admin()
|
|
attachments: Attachments = Attachments()
|
|
# basket: BasketOldDef = BasketOldDef()
|
|
# crs: Crs
|
|
crypto: Crypto = Crypto()
|
|
dashboard: Dashboard = Dashboard()
|
|
db: DB = DB()
|
|
# flask: Flask
|
|
geo: Geo = Geo()
|
|
gisaf: GisafConfig = GisafConfig()
|
|
gisaf_live: GisafLive = GisafLive()
|
|
jobs: list[Job] = []
|
|
map: Map = Map()
|
|
measures: Measures = Measures()
|
|
ogcapi: OGCAPI = OGCAPI()
|
|
plot: Plot = Plot()
|
|
plugins: dict[str, dict[str, Any]] = {}
|
|
survey: Survey = Survey()
|
|
version: str = version("gisaf-backend")
|
|
weather_station: dict[str, dict[str, Any]] = {}
|
|
widgets: Widgets = Widgets()
|
|
|
|
@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()
|
|
|
|
# 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'])
|