# -*- coding: utf-8 -*- """ Pydantic schemas for TOAR database """ from typing import List from pydantic import BaseModel, Json, validator import datetime as dt from toardb.generic.models import RS_enum, RC_enum from .models import DA_enum, SF_enum, AT_enum, DS_enum, MM_enum from toardb.variables.schemas import Variable #from toardb.stationmeta.schemas import StationmetaCoreCreate # ======== Timeseries ========= class TimeseriesCoreBase(BaseModel): id: int = None label: str order: int access_rights: str sampling_frequency: str aggregation: str source: str data_start_date: dt.datetime data_end_date: dt.datetime measurement_method: str sampling_height: float date_added: dt.datetime date_modified: dt.datetime station_id: int variable_id: int programme_id: int additional_metadata: Json @validator('access_rights') def check_access_rights(cls, v): return tuple(filter(lambda x: x.value == int(v), DA_enum))[0].string @validator('sampling_frequency') def check_sampling_frequency(cls, v): return tuple(filter(lambda x: x.value == int(v), SF_enum))[0].string @validator('aggregation') def check_aggregation(cls, v): return tuple(filter(lambda x: x.value == int(v), AT_enum))[0].string @validator('source') def check_source(cls, v): return tuple(filter(lambda x: x.value == int(v), DS_enum))[0].string @validator('measurement_method') def check_measurement_method(cls, v): return tuple(filter(lambda x: x.value == int(v), MM_enum))[0].string class TimeseriesCoreCreate(TimeseriesCoreBase): pass @validator('access_rights') def check_access_rights(cls, v): if tuple(filter(lambda x: x.string == v, DA_enum)): return v else: raise ValueError(f"data access rights not known: {v}") @validator('sampling_frequency') def check_sampling_frequency(cls, v): if tuple(filter(lambda x: x.string == v, SF_enum)): return v else: raise ValueError(f"sampling frequency not known: {v}") @validator('aggregation') def check_aggregation(cls, v): if tuple(filter(lambda x: x.string == v, AT_enum)): return v else: raise ValueError(f"aggregation type not known: {v}") @validator('source') def check_source(cls, v): if tuple(filter(lambda x: x.string == v, DS_enum)): return v else: raise ValueError(f"data source not known: {v}") @validator('measurement_method') def check_measurement_method(cls, v): if tuple(filter(lambda x: x.string == v, MM_enum)): return v else: raise ValueError(f"measurement method not known: {v}") class TimeseriesCore(TimeseriesCoreBase): id: int class Config: orm_mode = True # ======== TimeseriesRole ========= class TimeseriesRoleBase(BaseModel): id: int = None role: str status: str person_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 TimeseriesRoleCreate(TimeseriesRoleBase): 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 TimeseriesRole(TimeseriesRoleBase): id: int class Config: orm_mode = True # ======== TimeseriesAnnotation ========= class TimeseriesAnnotationBase(BaseModel): id: int = None kind: int text: str date_added: dt.datetime approved: bool contributor_id: int timeseries_id: int class TimeseriesAnnotationCreate(TimeseriesAnnotationBase): pass class TimeseriesAnnotation(TimeseriesAnnotationBase): id: int class Config: orm_mode = True # ======== TimeseriesProgramme ========= class TimeseriesProgrammeBase(BaseModel): id: int = None name: str longname: str homepage: str description: str class TimeseriesProgrammeCreate(TimeseriesProgrammeBase): pass class TimeseriesProgramme(TimeseriesProgrammeBase): id: int class Config: orm_mode = True # ======== for nested view/upload ========= class TimeseriesBase(TimeseriesCoreBase): roles: List[TimeseriesRole] = None annotations: List[TimeseriesAnnotation] = None variable: Variable # station: StationmetaCoreCreate programme: TimeseriesProgramme class Config: orm_mode = True class TimeseriesCreate(TimeseriesCoreCreate): roles: List[TimeseriesRoleCreate] = None annotations: List[TimeseriesAnnotation] = None class Config: orm_mode = True class Timeseries(TimeseriesBase): id: int class Config: orm_mode = True