# -*- coding: utf-8 -*- """ Pydantic schemas for TOAR database """ from typing import List from pydantic import BaseModel, Field, BaseConfig, Json, validator from geoalchemy2 import WKTElement from geoalchemy2.shape import to_shape import datetime as dt from .models import CZ_enum, CV_enum, ST_enum, TA_enum, TC_enum, \ TR_enum, DL_enum from toardb.generic.models import RC_enum, RS_enum from toardb.contacts.models import Contact # the following class was taken from: # https://github.com/tiangolo/fastapi/issues/312 class Coordinates(BaseModel): lat: float = Field(0, gte=-90, lte=90) lng: float = Field(0, gte=-180, lte=180) alt: float # ======== StationmetaCore ========= class StationmetaCoreStub(BaseModel): id: int = None name: str class StationmetaCoreBase(BaseModel): id: int = None codes: List[str] = [] name: str coordinates: Coordinates country: str state: str coordinate_validation_status: str coordinate_validation_date: dt.datetime type_of_environment: str type_of_area: str timezone: str additional_metadata: Json coordinate_validator_id: int class Config(BaseConfig): orm_mode = True # arbitrary_types_allowed = True @validator('coordinate_validation_status') def check_coordinate_validation_status(cls, v): return tuple(filter(lambda x: x.value == int(v), CV_enum))[0].string @validator('type_of_environment') def check_type_of_environment(cls, v): return tuple(filter(lambda x: x.value == int(v), ST_enum))[0].string @validator('type_of_area') def check_type_of_area(cls, v): return tuple(filter(lambda x: x.value == int(v), TA_enum))[0].string class StationmetaCoreCreate(StationmetaCoreBase): pass @validator('coordinate_validation_status') def check_coordinate_validation_status(cls, v): if tuple(filter(lambda x: x.string == v, CV_enum)): return v else: raise ValueError(f"coordinate validation status not known: {v}") @validator('type_of_environment') def check_type_of_environment(cls, v): if tuple(filter(lambda x: x.string == v, ST_enum)): return v else: raise ValueError(f"type of environment not known: {v}") @validator('type_of_area') def check_type_of_area(cls, v): if tuple(filter(lambda x: x.string == v, TA_enum)): return v else: raise ValueError(f"type of area not known: {v}") class StationmetaCore(StationmetaCoreBase): id: int class Config: orm_mode = True def get_geom_from_coordinates(coordinates: Coordinates) -> str: geom_wkte = f'SRID=4326;POINTZ({coordinates.lng} {coordinates.lat} {coordinates.alt})' return geom_wkte def get_coordinates_from_geom(geom: WKTElement) -> Coordinates: shply_geom = to_shape(geom) coordinates = Coordinates(lng=shply_geom.x, lat=shply_geom.y, alt=shply_geom.z) return coordinates # ======== StationmetaAnnotation ========= class StationmetaAnnotationBase(BaseModel): id: int = None kind: int text: str date_added: dt.datetime approved: bool contributor_id: int class StationmetaAnnotationCreate(StationmetaAnnotationBase): pass class StationmetaAnnotation(StationmetaAnnotationBase): id: int class Config: orm_mode = True # ======== StationmetaAux ========= # -------- StationmetaAuxDoc --------- class StationmetaAuxDocBase(BaseModel): id: int = None resource_description: str date_added: dt.datetime resource: str station_id: int class StationmetaAuxDocCreate(StationmetaAuxDocBase): pass class StationmetaAuxDoc(StationmetaAuxDocBase): id: int class Config: orm_mode = True # -------- StationmetaAuxImage --------- class StationmetaAuxImageBase(BaseModel): id: int = None resource_description: str date_added: dt.datetime resource: str image_height: int image_width: int station_id: int class StationmetaAuxImageCreate(StationmetaAuxImageBase): pass class StationmetaAuxImage(StationmetaAuxImageBase): id: int class Config: orm_mode = True # -------- StationmetaAuxUrl --------- class StationmetaAuxUrlBase(BaseModel): id: int = None resource_description: str date_added: dt.datetime resource: str station_id: int class StationmetaAuxUrlCreate(StationmetaAuxUrlBase): pass class StationmetaAuxUrl(StationmetaAuxUrlBase): id: int class Config: orm_mode = True # ======== StationmetaGlobal ========= class StationmetaGlobalBase(BaseModel): id: int = None population_density_year2010: float max_population_density_25km_year2010: float climatic_zone: str nightlight_1km_year2013: float nightlight_5km_year2013: float max_nightlight_25km_year2013: float wheat_production_year2000: float rice_production_year2000: float edgar_htap_v2_nox_emissions_year2010: float omi_no2_column_years2011to2015: float htap_region_tier1: str etopo_alt: float etopo_min_alt_5km: float etopo_relative_alt: float dominant_landcover_year2012: str toar1_category: str station_id: int @validator('climatic_zone') def check_climatic_zone(cls, v): return tuple(filter(lambda x: x.value == int(v), CZ_enum))[0].string @validator('toar1_category') def check_toar1_category(cls, v): return tuple(filter(lambda x: x.value == int(v), TC_enum))[0].string @validator('htap_region_tier1') def check_htap_region_tier1(cls, v): return tuple(filter(lambda x: x.value == int(v), TR_enum))[0].string @validator('dominant_landcover_year2012') def check_dominant_landcover_year2012(cls, v): return tuple(filter(lambda x: x.value == int(v), DL_enum))[0].string class StationmetaGlobalCreate(StationmetaGlobalBase): pass @validator('climatic_zone') def check_climatic_zone(cls, v): if tuple(filter(lambda x: x.string == v, CZ_enum)): return v else: raise ValueError(f"climatic zone not known: {v}") @validator('toar1_category') def check_toar1_category(cls, v): if tuple(filter(lambda x: x.string == v, TC_enum)): return v else: raise ValueError(f"TOAR1 category not known: {v}") @validator('htap_region_tier1') def check_htap_region_tier1(cls, v): if tuple(filter(lambda x: x.string == v, TR_enum)): return v else: raise ValueError(f"HTAP region TIER1 not known: {v}") @validator('dominant_landcover_year2012') def check_dominant_landcover_year2012(cls, v): if tuple(filter(lambda x: x.string == v, DL_enum)): return v else: raise ValueError(f"dominant landcover (year2012) not known: {v}") class StationmetaGlobal(StationmetaGlobalBase): id: int class Config: orm_mode = True class StationmetaGlobalBaseNested(BaseModel): id: int = None population_density_year2010: float = None max_population_density_25km_year2010: float = None climatic_zone: str = None nightlight_1km_year2013: float = None nightlight_5km_year2013: float = None max_nightlight_25km_year2013: float = None wheat_production_year2000: float = None rice_production_year2000: float = None edgar_htap_v2_nox_emissions_year2010: float = None omi_no2_column_years2011to2015: float = None htap_region_tier1: int = None etopo_alt: float = None etopo_min_alt_5km: float = None etopo_relative_alt: float = None dominant_landcover_year2012: int = None toar1_category: int = None class StationmetaGlobalNestedCreate(StationmetaGlobalBaseNested): pass class StationmetaGlobalNested(StationmetaGlobalBaseNested): id: int class Config: orm_mode = True # ======== StationmetaGlobalService ========= class StationmetaGlobalServiceBase(BaseModel): id: int = None variable_name: str result_type: int result_nvalues: int service_valid_year: int service_url: str class StationmetaGlobalServiceCreate(StationmetaGlobalServiceBase): pass class StationmetaGlobalService(StationmetaGlobalServiceBase): id: int class Config: orm_mode = True # ======== StationmetaRole ========= class StationmetaRoleBase(BaseModel): id: int = None role: str status: str # contact: Contact contact_id: int @validator('role') def check_role(cls, v): return tuple(filter(lambda x: x.value == int(v), RC_enum))[0].string @validator('status') def check_status(cls, v): return tuple(filter(lambda x: x.value == int(v), RS_enum))[0].string class Config: orm_mode = True class StationmetaRoleCreate(StationmetaRoleBase): pass @validator('role') def check_role(cls, v): if tuple(filter(lambda x: x.string == v, RC_enum)): return v else: raise ValueError(f"role code not known: {v}") @validator('status') def check_status(cls, v): if tuple(filter(lambda x: x.string == v, RS_enum)): return v else: raise ValueError(f"role status not known: {v}") class StationmetaRole(StationmetaRoleBase): id: int class Config: orm_mode = True # ======== for nested view/upload ========= class StationmetaBase(StationmetaCoreBase): roles: List[StationmetaRole] = None annotations: List[StationmetaAnnotation] = None aux_images: List[StationmetaAuxImage] = None aux_docs: List[StationmetaAuxDoc] = None aux_urls: List[StationmetaAuxUrl] = None globalmeta: StationmetaGlobal = None globalservice: StationmetaGlobalService = None class Config: orm_mode = True class StationmetaCreate(StationmetaCoreCreate): roles: List[StationmetaRoleBase] = None annotations: List[StationmetaAnnotation] = None aux_images: List[StationmetaAuxImage] = None aux_docs: List[StationmetaAuxDoc] = None aux_urls: List[StationmetaAuxUrl] = None globalmeta: StationmetaGlobalNestedCreate = None globalservice: StationmetaGlobalService = None class Config: orm_mode = True class Stationmeta(StationmetaBase): id: int class Config: orm_mode = True