#!/usr/bin/python3
"""
Copyright (c) 2023 NIBIO <http://www.nibio.no/>.
This file is part of VIPSCore-Python-Common.
VIPSCore-Python-Common is free software: you can redistribute it and/or modify
it under the terms of the NIBIO Open Source License as published by
NIBIO, either version 1 of the License, or (at your option) any
later version.
VIPSCore-Python-Common is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
NIBIO Open Source License for more details.
You should have received a copy of the NIBIO Open Source License
along with VIPSCore-Python-Common. If not, see <http://www.nibio.no/licenses/>.
"""
#
# @author Tor-Einar Skog <tor-einar.skog@nibio.no>
# https://www.pedaldrivenprogramming.com/2020/12/fastapi-fundamentals-pydantic/
from datetime import datetime
from shapely.geometry import Point, Polygon
from pydantic import BaseModel, validator, constr, Field
from typing import Any, ClassVar, Optional
import json
import pytz
class Result(BaseModel):
"""
Represents a set of DSS model result values for a given point in space (Point, Polygon, MultiPolygon) and time (Period or immediate)
"""
valid_time_start: datetime = Field(alias="validTimeStart")
valid_time_end: Optional[datetime] = Field(alias="validTimeEnd")
valid_geometry: Optional[Any] = Field(alias="validGeometry")
warning_status: int = Field(alias="warningStatus")
all_values: str = Field("{}", alias="allValues")
class Config:
allow_population_by_field_name = True
WARNING_STATUS_NO_WARNING: ClassVar[int] = 0
WARNING_STATUS_NO_WARNING_MISSING_DATA: ClassVar[int] = 1
WARNING_STATUS_NO_RISK: ClassVar[int] = 2
WARNING_STATUS_MINOR_RISK: ClassVar[int] = 3
WARNING_STATUS_HIGH_RISK: ClassVar[int] = 4
def get_keys(self):
return set(json.loads(self.all_values.keys)) if self.all_values is not None else set()
@validator("valid_time_start")
def ensure_timezone(cls, v):
if v.tzinfo is None or v.tzinfo.utcoffset(v) is None:
raise ValueError("%s must be timezone aware" % v)
return v
@validator("valid_time_end")
def ensure_none_or_timezone(cls, v):
if v is None:
return v
if v.tzinfo is None or v.tzinfo.utcoffset(v) is None:
raise ValueError("%s must be timezone aware" % v)
return v
@validator("valid_geometry")
def ensure_geometry(cls,v):
if v is not None and not isinstance(v, Point) and not isinstance(v, Polygon):
raise ValueError("%s is not a " % v)
def set_value(self, namespace, key, value):
temp_all_values = json.loads(self.all_values)
temp_all_values["%s.%s" %(namespace, key)] = value
self.all_values = json.dumps(temp_all_values)
def get_value(self, namespace, key):
temp_all_values = json.loads(self.all_values)
return temp_all_values.get("%s.%s" %(namespace, key))
def set_all_values(self, values_dict):
self.all_values = json.dumps(values_dict)
class ModelConfiguration(BaseModel):
"""All the input data for the model."""
model_id: constr(min_length=10, max_length=10) = Field(alias="modelId")
config_parameters: dict = Field(alias="configParameters")
class Config:
allow_population_by_field_name = True
# Can we do this and still serialize the object??
def get_config_parameter_as_date(self, param_name: str, required=True) -> datetime:
"""
Input check and typing of date (ISO format)
"""
param_value = self.config_parameters.get(param_name, None)
if required and param_value is None:
raise ModelConfigurationException("%s is required" % param_name)
try:
return datetime.fromisoformat(param_value)
except ValueError:
raise ModelConfigurationException("%s=%s, which is not a valid date." % (param_name, param_value))
# Can we do this and still serialize the object??
def get_config_parameter_as_string(self, param_name: str, required=True) -> datetime:
"""
Input check
"""
param_value = self.config_parameters.get(param_name, None)
if required and param_value is None:
raise ModelConfigurationException("%s is required" % param_name)
return param_value
# Can we do this and still serialize the object??
def get_config_parameter_as_timezone(self, param_name: str, required=True) -> datetime:
"""
Input check
"""
param_value = self.config_parameters.get(param_name, None)
if required and param_value is None:
raise ModelConfigurationException("%s is required" % param_name)
try:
return pytz.timezone(param_value)
except pytz.exceptions.UnknownTimeZoneError:
raise ModelConfigurationException("%s=%s, which is not a valid timezone." % (param_name, param_value))
class ModelConfigurationException(Exception):
"""
Should be raised when there's something wrong with the VIPS model configuration
passed to VIPSModel.set_configuration
"""
class WeatherObservation(BaseModel):
"""Data class for a single weather observation"""
elementMeasurementTypeId: str
logIntervalId: int
timeMeasured: datetime
value: float
# Log interval categories
LOG_INTERVAL_ID_1M: ClassVar[int] = 8
LOG_INTERVAL_ID_5M: ClassVar[int] = 9
LOG_INTERVAL_ID_10M: ClassVar[int] = 6
LOG_INTERVAL_ID_15M: ClassVar[int] = 5
LOG_INTERVAL_ID_30M: ClassVar[int] = 4
LOG_INTERVAL_ID_1H: ClassVar[int] = 1
LOG_INTERVAL_ID_3H: ClassVar[int] = 3
LOG_INTERVAL_ID_6H: ClassVar[int] = 7
LOG_INTERVAL_ID_1D: ClassVar[int] = 2
@validator("timeMeasured")
def ensure_timezone(cls, v):
if v.tzinfo is None or v.tzinfo.utcoffset(v) is None:
raise ValueError("%s must be timezone aware" % v)
return v
class WeatherElements():
# Mean temp (Celcius)
TEMPERATURE_MEAN = "TM"
# Minimum temperature (Celcius)
TEMPERATURE_MINIMUM = "TN"
# Maximum temperature (Celcius)
TEMPERATURE_MAXIMUM = "TX"
# Instantaneous temperature (Celcius)
TEMPERATURE_INSTANTANEOUS = "TT"
#Soil temperatures at various depths
SOIL_TEMPERATURE_5CM_MEAN ="TJM5"
SOIL_TEMPERATURE_10CM_MEAN ="TJM10"
SOIL_TEMPERATURE_25CM_MEAN ="TJM25"
SOIL_TEMPERATURE_50CM_MEAN ="TJM50"
# Dew point temperature (Celcius)
DEW_POINT_TEMPERATURE = "TD"
# Aggregated rainfall (millimeters)
PRECIPITATION = "RR"
# Average relative humidity (%)
RELATIVE_HUMIDITY_MEAN = "UM"
# Instantaneous relative humidity (%)
RELATIVE_HUMIDITY_INSTANTANEOUS = "UU"
# Global radiation (W/sqm)
GLOBAL_RADIATION = "Q0"
# Immmediate calculated clear sky solar radiation (clear sky)
GLOBAL_RADIATION_CLEAR_SKY_CALCULATED = "Q0c"
# Leaf wetness (Minutes per hour)
LEAF_WETNESS_DURATION = "BT"
# Leaf wetness at ground level (Minutes per hour)
LEAF_WETNESS_DURATION_GROUND_LEVEL = "BTg"
# Soil water content at various depths
SOIL_WATER_CONTENT_5CM = "VAN5"
SOIL_WATER_CONTENT_10CM = "VAN10"
SOIL_WATER_CONTENT_25CM = "VAN25"
SOIL_WATER_CONTENT_50CM = "VAN50"
# Soil conductivity at various depths
SOIL_CONDUCTIVITY_5CM = "LEJ5"
# Average wind speed (meters / second) for the last 60 minutes, measured at 2 meter height
WIND_SPEED_2M = "FM2"
# Average wind speed (meters / second) for the last 60 minutes, measured at 10 meter height
WIND_SPEED_10M = "FM"
# Average wind speed (meters / second) for the last 10 minutes, measured at 2 meter height
WIND_SPEED_10MIN_2M = "FF2"
# Average wind speed (meters / second) for the last 10 minutes, measured at 10 meter height
WIND_SPEED_10MIN_10M = "FF"
# Average wind direction for the last 10 minutes, measured at 2 meter height
WIND_DIR_10MIN_2M = "DD2"
# Average wind direction for the last 10 minutes, measured at 10 meter height
WIND_DIR_10MIN_10M = "DD"
# Potential evapotranspiration by the Penman equation
POTENTIAL_EVAPORATION = "EPP"