# need to create the `__all__` attribute for documentation.
# if you want sphinx-apidoc to autodoc a module/class it should be
# included below.
# generate using ipython: from `shyft.hydrology._api import *` and `dir()`, then clean
# concatenate several lists from this file and from the list you create:
__all__ = [
'GoalFunctionModel',
'GoalFunctionModelVector',
'GoalFunctionClient',
'GoalFunctionServer',
'GeoCellDataServer',
'GeoCellDataClient',
'GeoCellDataModel',
'StateServer',
'StateClient',
'StateModel',
'ParameterServer',
'ParameterClient',
'ParameterModel',
'parse_cf_time',
'np_array',
'StrGeoCellData',
'ABS_DIFF',
'ARegionEnvironment',
'ActualEvapotranspirationCalculate_step',
'ActualEvapotranspirationParameter',
'ActualEvapotranspirationResponse',
'BTKParameter',
'CELL_CHARGE',
'CatchmentPropertyType',
'CellEnvironment',
'CellStateId',
'DISCHARGE',
'EXPONENTIAL',
'FlowAdjustResult',
'GAUSSIAN',
'GammaSnowCalculator',
'GammaSnowParameter',
'GammaSnowResponse',
'GammaSnowState',
'GeoCellData',
'GeoCellDataVector',
'GeoPointSource',
'GeoPointSourceVector',
'GlacierMeltParameter',
'HbvActualEvapotranspirationParameter',
'HbvActualEvapotranspirationResponse',
'HbvActualEvapotranspirationCalculate_step',
'HbvPhysicalSnowCalculator',
'HbvPhysicalSnowParameter',
'HbvPhysicalSnowResponse',
'HbvPhysicalSnowState',
'HbvSnowCalculator',
'HbvSnowParameter',
'HbvSnowResponse',
'HbvSnowState',
'HbvSoilCalculator',
'HbvSoilParameter',
'HbvSoilResponse',
'HbvSoilState',
'HbvTankCalculator',
'HbvTankParameter',
'HbvTankResponse',
'HbvTankState',
'SnowTilesParameter',
'IDWParameter',
'IDWPrecipitationParameter',
'IDWTemperatureParameter',
'InterpolationParameter',
'KLING_GUPTA',
'KalmanBiasPredictor',
'KalmanFilter',
'KalmanParameter',
'KalmanState',
'KirchnerCalculator',
'KirchnerParameter',
'KirchnerResponse',
'KirchnerState',
'LandTypeFractions',
'MethodStackParameter',
'NASH_SUTCLIFFE',
'OKCovarianceType',
'OKParameter',
'PenmanMonteithCalculator',
'PenmanMonteithParameter',
'PenmanMonteithResponse',
'PrecipitationCorrectionCalculator',
'PrecipitationCorrectionParameter',
'PrecipitationSource',
'PrecipitationSourceVector',
'PriestleyTaylorCalculator',
'PriestleyTaylorParameter',
'PriestleyTaylorResponse',
'RMSE',
'ROUTED_DISCHARGE',
'RadiationCalculator',
'RadiationParameter',
'RadiationResponse',
'RadiationSource',
'RadiationSourceVector',
'RelHumSource',
'RelHumSourceVector',
'River',
'RiverNetwork',
'RoutingInfo',
'SNOW_COVERED_AREA',
'SNOW_WATER_EQUIVALENT',
'SkaugenCalculator',
'SkaugenParameter',
'SkaugenResponse',
'SkaugenState',
'TargetSpecCalcType',
'TargetSpecificationPts',
'TargetSpecificationVector',
'TemperatureSource',
'TemperatureSourceVector',
'TsTransform',
'UHGParameter',
'WindSpeedSource',
'WindSpeedSourceVector',
'bayesian_kriging_temperature',
'catchment',
'cell',
'compute_geo_ts_values_at_time',
'create_precipitation_source_vector_from_np_array',
'create_radiation_source_vector_from_np_array',
'create_rel_hum_source_vector_from_np_array',
'create_temperature_source_vector_from_np_array',
'create_wind_speed_source_vector_from_np_array',
'glacier_melt_step',
'idw_precipitation',
'idw_radiation',
'idw_relative_humidity',
'idw_temperature',
'idw_wind_speed',
'make_uhg_from_gamma',
'ordinary_kriging',
'stat_scope',
'version',
'pt_gs_k',
'pt_hps_k',
'pt_hs_k',
'pt_st_k',
'pt_st_hbv',
'pt_ss_k',
'r_pm_gs_k',
'r_pt_gs_k',
'CalibrationOption',
'OptimizerMethod',
'CalibrationStatus',
'ParameterOptimizer',
# added
'DrmClient', 'DrmServer', 'RegionModelType',
'SnowTilesParameter', 'SnowTilesState', 'SnowTilesCalculator', 'SnowTilesResponse',
'shyftdata_dir', 'percentiles',
'viz', 'repository', 'orchestration'
]
import os
from os import path
# we need timeseries, start with it early
from shyft.time_series import (IntVector, DoubleVector, TsVector, GeoPoint, GeoPointVector, TimeAxis)
# this is backward compatible
from shyft.hydrology._api import *
from shyft.hydrology import r_pm_gs_k
from shyft.hydrology import r_pt_gs_k
from shyft.hydrology import pt_gs_k
from shyft.hydrology import pt_hps_k
from shyft.hydrology import pt_hs_k
from shyft.hydrology import pt_st_k
from shyft.hydrology import pt_st_hbv
from shyft.hydrology import pt_ss_k
from shyft.hydrology import viz
from shyft.hydrology import repository
from shyft.hydrology import orchestration
from math import sqrt
import numpy as np
if "SHYFT_DATA" in os.environ:
shyftdata_dir:str = os.environ["SHYFT_DATA"]
else:
# If SHYFT_DATA environment variable is not here, then use a decent guess
shyftdata_dir:str = path.join(path.dirname(__file__),path.pardir, path.pardir, path.pardir, path.pardir, "shyft-data")
shyftdata_dir:str = path.normpath(shyftdata_dir)
[docs]
def percentiles(tsv: TsVector, time_axis: TimeAxis, percentile_list: IntVector) -> TsVector:
return tsv.percentiles(time_axis, percentile_list)
TargetSpecificationVector.size = lambda self: len(self)
# Fix up LandTypeFractions
LandTypeFractions.__str__ = lambda \
self: "LandTypeFractions(glacier={0},lake={1},reservoir={2},forest={3},unspecified={4})".format(self.glacier(),
self.lake(),
self.reservoir(),
self.forest(),
self.unspecified())
# Fix up GeoCellData
[docs]
def StrGeoCellData(gcd):
return "GeoCellData(mid_point={0},catchment_id={1},area={2},ltf={3})".format(str(gcd.mid_point()),
gcd.catchment_id(), gcd.area(),
str(gcd.land_type_fractions_info()))
GeoCellData.__str__ = lambda self: StrGeoCellData(self)
GeoCellData.__repr__ = GeoCellData.__str__
# Fix up ARegionEnvironment
TemperatureSource.vector_t = TemperatureSourceVector
PrecipitationSource.vector_t = PrecipitationSourceVector
RadiationSource.vector_t = RadiationSourceVector
RelHumSource.vector_t = RelHumSourceVector
WindSpeedSource.vector_t = WindSpeedSourceVector
Int64Vector = IntVector
[docs]
def np_array(dv: DoubleVector):
"""
convert flattened double-vector to numpy array
Parameters
----------
dv
Returns
-------
numpy array.
"""
f = dv.to_numpy()
n = int(sqrt(dv.size()))
m = f.reshape(n, n)
return m
# fixup kalman state
KalmanState.x = property(lambda self: KalmanState.get_x(self).to_numpy(),
doc="represents the current bias estimate, kalman.state.x")
KalmanState.k = property(lambda self: KalmanState.get_k(self).to_numpy(),
doc="represents the current kalman gain factors, kalman.state.k")
KalmanState.P = property(lambda self: np_array(KalmanState.get_P(self)),
doc="returns numpy array of kalman.state.P, the nxn covariance matrix")
KalmanState.W = property(lambda self: np_array(KalmanState.get_W(self)),
doc="returns numpy array of kalman.state.W, the nxn noise matrix")
# fixup KalmanBiasPredictor
def KalmanBiasPredictor_update_with_forecast(bp, fc_set, obs, time_axis):
"""
Parameters
----------
bp
fc_set : TemperatureSourceVector or TsVector
obs : TimeSeries
time_axis : TimeAxis
Returns
-------
nothing
"""
if isinstance(fc_set, TemperatureSourceVector):
KalmanBiasPredictor.update_with_geo_forecast(bp, fc_set, obs, time_axis)
else:
KalmanBiasPredictor.update_with_forecast_vector(bp, fc_set, obs, time_axis)
def KalmanBiasPredictor_compute_running_bias(bp, fc_ts, obs_ts, time_axis):
"""
compute the running bias timeseries,
using one 'merged' - forecasts and one observation time - series.
Before each day - period, the bias - values are copied out to form
a continuous bias prediction time-series.
Parameters
----------
bp : KalmanBiasPredictor
The bias predictor object it self
fc_ts : TimeSeries
a merged forecast ts
with period covering the observation_ts and time_axis supplied
obs_ts : TimeSeries
the observation time-series
time_axis : TimeAxis
covering the period/timesteps to be updated
e.g. yesterday, 3h resolution steps, according to the points in the filter
Returns
-------
bias_ts : TimeSeries(time_axis,bias_vector,POINT_AVERAGE)
computed running bias-ts
"""
return KalmanBiasPredictor.compute_running_bias_ts(bp, fc_ts, obs_ts, time_axis)
KalmanBiasPredictor.update_with_forecast = KalmanBiasPredictor_update_with_forecast
KalmanBiasPredictor.compute_running_bias = KalmanBiasPredictor_compute_running_bias
def geo_point_source_vector_values_at_time(gtsv: GeoPointSourceVector, t: int):
# if not isinstance(gtsv, GeoPointSourceVector):
# raise RuntimeError('Supplied list of timeseries must be of GeoPointSourceVector')
return compute_geo_ts_values_at_time(gtsv, t).to_numpy()
GeoPointSourceVector.values_at_time = geo_point_source_vector_values_at_time
# GeoPointSourceVector.values_at_time.__doc__ = compute_geo_ts_values_at_time.__doc__.replace('DoubleVector','ndarray')
RadiationSourceVector.values_at_time = GeoPointSourceVector.values_at_time
PrecipitationSourceVector.values_at_time = GeoPointSourceVector.values_at_time
TemperatureSourceVector.values_at_time = GeoPointSourceVector.values_at_time
RelHumSourceVector.values_at_time = GeoPointSourceVector.values_at_time
WindSpeedSourceVector.values_at_time = GeoPointSourceVector.values_at_time
ARegionEnvironment.variables = property(
lambda self: [
('temperature', self.temperature),
('precipitation', self.precipitation),
('radiation', self.radiation),
('rel_hum', self.rel_hum),
('wind_speed', self.wind_speed)
],
doc='returns the list of available forcing variables as tuples(string,reference_to_variable)'
)
#