"""
Author: HECE - University of Liege, Pierre Archambeau, Utashi Ciraane Docile
Date: 2024
Copyright (c) 2024 University of Liege. All rights reserved.
This script and its content are protected by copyright law. Unauthorized
copying or distribution of this file, via any medium, is strictly prohibited.
"""
# Libraries
import copy
import enum
import logging
import numpy as np
import os
import pandas as pd
import matplotlib.pyplot as plt
from datetime import datetime, timedelta
from matplotlib.ticker import MultipleLocator, FuncFormatter
from matplotlib import animation ,rcParams
from matplotlib.axes import Axes
from matplotlib.figure import Figure
from matplotlib.lines import Line2D
from typing import Literal,Union
from tqdm import tqdm
from .PyVertexvectors import Zones, zone, vector, wolfvertex
[docs]
class Constants(enum.Enum):
"""
Contain the constants used throughout
the module.
"""
[docs]
class Hydrograph(pd.Series):
'''
A pandas series containing discharges (values)
with their respective times of observation (indices).
@ The class is inherited which means
all series method are available in this
class escorted by a few other functions useful in
the interaction with wolfhece tools.
'''
def __init__(self,
data=None,
index=None,
dtype=None,
name=None,
copy=None,
file_path:str ='') -> None:
# file_path:str ='') -> None:
if file_path != '':
data =self.read_from_wolf_file(file_path)
elif isinstance(data, str):
data = self.read_from_wolf_file(data)
# In case a pandas series is passed
elif isinstance(data, pd.Series):
index = data.index
data = data.values
if name is None:
name = 'Discharge'
super().__init__(data,index,dtype, name, copy)
[docs]
def write_as_wolf_file(self,
file_path:str,
writing_method: Literal['continuous', 'stepwise'] = 'continuous',
epsilon:float = 1.
):
"""
Write the hydrograph on the infiltration format
of wolf models -> infil[n].tv file.
"""
lgth = self.size
if writing_method == 'continuous':
with open(file_path, 'w') as wolf_file:
wolf_file.write(f'{lgth}\n')
for i in range(lgth):
wolf_file.write(f'{self.index[i]}{Constants.SEPARATOR.value}{self.values[i]}\n')
elif writing_method == 'stepwise':
with open(file_path, 'w') as wolf_file:
wolf_file.write(f'{2*lgth -1}\n')
wolf_file.write(f'{self.index[0]}{Constants.SEPARATOR.value}{self.values[0]}\n')
for a in range(lgth - 1):
i = a + 1
wolf_file.write(f'{self.index[i] - epsilon}{Constants.SEPARATOR.value}{self.values[i-1]}\n')
wolf_file.write(f'{self.index[i]}{Constants.SEPARATOR.value}{self.values[i]}\n')
[docs]
def read_from_wolf_file(self,
file_path:str,
separator: Literal[',',';', '\t'] = Constants.SEPARATOR.value
) -> dict:
"""
Read a wolf file at the format infil[n].tv and
return its data as a dictionnary where:
- keys are times and,
- values are discharges.
"""
data = {}
with open(file_path,'r') as wolf_file:
for line in wolf_file.readlines()[1:]:
observation = line.splitlines()[0].split(separator)
data[float(observation[0])] = float(observation[1])
return data
# FIXME right convert to second method
[docs]
def set_zero_as_start(self):
"""
Set the first time of the hydrograph as zero.
"""
self.index = self.index - self.index[0]
[docs]
class HydrOperations:
"""
Contains the operations that can be performed on Hydrographs.
"""
def __init__(self) -> None:
pass
[docs]
def substract_hydrograph(self, hydrograph1:Hydrograph, hydrograph2:Hydrograph) -> Hydrograph:
"""Substract two hydrographs and return the difference.
The second hydrograph is substracted from the first.
.. note:: The Nan values in the difference are dropped to avoid
constraints of hydrographs index.
:param hydrograph1: The first hydrograph.
:type hydrograph1: Hydrograph
:param hydrograph2: The second hydrograph.
:type hydrograph2: Hydrograph
:return: The difference between the two hydrographs.
:rtype: Hydrograph
"""
difference = hydrograph1.subtract(hydrograph2)
difference.dropna(inplace=True)
return Hydrograph(difference)
[docs]
def add_hydrograph(self, hydrograph1:Hydrograph, hydrograph2:Hydrograph)->Hydrograph:
"""Add two hydrographs and return the sum.
The second hydrograph is added to the first.
.. note:: The Nan values in the sum are dropped to avoid
constraints of hydrographs index.
:param hydrograph1: The first hydrograph.
:type hydrograph1: Hydrograph
:param hydrograph2: The second hydrograph.
:type hydrograph2: Hydrograph
:return: The sum of the two hydrographs.
:rtype: Hydrograph
"""
sum = hydrograph1.add(hydrograph2)
sum.dropna(inplace=True)
return Hydrograph(sum)
[docs]
def add_list_of_hydrographs(self,hydrographs: list) -> Hydrograph:
"""Add a list of hydrographs and return the sum.
The hydrographs are added one after the other to the first one.
.. note:: The Nan values in the sum are dropped to avoid
constraints of hydrographs index.
:param hydrographs: The list of hydrographs.
:type hydrographs: list
:return: The sum of the hydrographs.
:rtype: Hydrograph
"""
sum = hydrographs[0]
for i in range(1, len(hydrographs)):
sum = self.add_hydrograph(sum, hydrographs[i])
return Hydrograph(sum)
[docs]
def substract_list_of_hydrographs(self, hydrographs: list) -> Hydrograph:
"""Substract a list of hydrographs and return the difference.
The hydrographs are substracted one after the other from the first one.
.. note:: The Nan values in the difference are dropped to avoid
constraints of hydrographs index.
:param hydrographs: The list of hydrographs.
:type hydrographs: list
:return: The difference between the hydrographs.
:rtype: Hydrograph
"""
difference = hydrographs[0]
for i in range(1, len(hydrographs)):
difference = self.substract_hydrograph(difference, hydrographs[i])
return Hydrograph(difference)
@staticmethod
[docs]
def as_stepwise_Hydrograph(hydrograph: Hydrograph|pd.Series, epsilon=1) -> Hydrograph:
"""
Return a stepwise hydrograph from a given hydrograph.
The steps are made by shifting uniformly the time index by an epsilon value (default 1 is subtracted)
while keeping the same discharge value.
This is useful for harmonizing a hydrograph in a stepwise manner.
:param hydrograph: The hydrograph to be converted to stepwise format.
:type hydrograph: Hydrograph or Pandas Series
:param epsilon: The value subtracted from each index of the hydrograph (each timestep), defaults to 1
:type epsilon: int, optional
:return: the stepwise hydrograph
:rtype: Hydrograph
"""
stepwised = {hydrograph.index[0]: hydrograph.values[0]} # Dictionnary to hold the stepwise values.
# Loop through the hydrograph and create stepwise values
# by shifting the index by epsilon and keeping the same discharge value.
for i in range(len(hydrograph)-1):
stepwised[hydrograph.index[i+1] - epsilon] = hydrograph.values[i+1]
stepwised[hydrograph.index[i+1]] = hydrograph.values[i+1]
# stepwised_hydrograph = Hydrograph(pd.Series(stepwised))
# Convert the obtain series to a Hydrograph object.
return Hydrograph(pd.Series(stepwised))
@staticmethod
[docs]
def find_conversion_factor_from_seconds(convert_to: Literal['weeks','days', 'hours', 'minutes', 'seconds'] = 'seconds') -> int:
"""
Find the conversion factor of a given time format to seconds.
:param convert_to: The time format to convert to.
:type convert_to: Literal['days', 'hours', 'minutes', 'seconds']
:return: The conversion factor.
:rtype: int
"""
if convert_to == 'days':
return 86400
elif convert_to == 'hours':
return 3600
elif convert_to == 'minutes':
return 60
elif convert_to == 'seconds':
return 1
elif convert_to == 'weeks':
return 604800
else:
raise ValueError('The time format is not recognized')
[docs]
def create_one_matplolib_axis(self,
time:float,
convert_to: Literal['weeks','days', 'hours', 'minutes', 'seconds'] = 'seconds',
figsize:tuple = (30,5),
ylabel:str = 'Discharge ($m^3$/s)',
tick_spacing:int = 1,
x_start:int = 0,
x_end:int = None,
ax = None) -> Axes:
"""
Create a matplotlib axis with a date format
of the number of days passed as argument.
"""
coefficient = self.find_conversion_factor_from_seconds(convert_to)
assert x_start*coefficient <= coefficient*time,\
f"The origin of the X-axis can not be greater than its end."
def axis_as_time_unit(val, pos):
value = val/coefficient
return f'{value:#.1F}'
if ax is None:
fig, ax = plt.subplots(1,1, figsize=figsize)
ax.xaxis.set_major_formatter(FuncFormatter(axis_as_time_unit))
if x_end is not None:
ax.set_xlim(x_start*coefficient,coefficient*x_end)
ax.set_xticks(np.arange(x_start*coefficient, coefficient*x_end, coefficient*tick_spacing))
else:
ax.set_xticks(np.arange(x_start*coefficient, coefficient*time, coefficient*tick_spacing))
ax.set_xlim(x_start*coefficient,coefficient*time)
ax.set_xlabel(f'Time [{convert_to}]')
ax.set_ylabel(ylabel)
return ax