"""
Module de configuration des données.
Ce module contient les classes et fonctions nécessaires pour charger et valider
les configurations de filtrage des données.
"""
from enum import StrEnum
from pathlib import Path
from pydantic import BaseModel, field_validator
import re
import pandas as pd
from typing import Optional, Any
from loguru import logger
from .helper import load_config
LOGGER = logger.bind(name="CSB-Processing.Config.ProcessingConfig")
ConfigDict = dict[str, int | float | str]
CSBconfigDict = dict[str, dict[str, dict[str, ConfigDict | dict[str, ConfigDict]]]]
MIN_LATITUDE: int | float = -90
MAX_LATITUDE: int | float = 90
MIN_LONGITUDE: int | float = -180
MAX_LONGITUDE: int | float = 180
MIN_DEPTH: int | float = 0
MAX_DEPTH: int | float | None = None
MIN_SPEED: int | float | None = None
MAX_SPEED: int | float | None = None
INFO: str = "INFO"
MAX_ITERATIONS: int = 10
DECIMAL_PRECISION: int = 1
RESOLUTION: int | float = 0.00005
WATER_LEVEL_TOLERANCE: pd.Timedelta = pd.Timedelta("15 min")
CONSTANT_TVU_WLO: float = 0.04
CONSTANT_TVU_WLP: float = 0.35
DEPTH_COEFFICIENT: float = 0.5
DEPTH_COEFFICIENT_SSP: float = 4.1584
MAX_DISTANCE_SSP: float = 30000
CONE_ANGLE_SONAR: float = 20
CONSTANT_THU: float = 3
NBIN_X: int = 35
NBIN_Y: int = 35
[docs]
class FileTypes(StrEnum):
"""
Enumération des types de fichiers de sortie.
"""
GEOJSON = "geojson"
GPKG = "gpkg"
CSAR = "csar"
PARQUET = "parquet"
FEATHER = "feather"
CSV = "csv"
GEOTIFF = "geotiff"
EXPORT_FORMAT: list[str] = [FileTypes.GPKG]
[docs]
class Filter(StrEnum):
"""
Enum for status codes.
"""
SPEED_FILTER = "SPEED_FILTER"
LATITUDE_FILTER = "LATITUDE_FILTER"
LONGITUDE_FILTER = "LONGITUDE_FILTER"
TIME_FILTER = "TIME_FILTER"
DEPTH_FILTER = "DEPTH_FILTER"
FILTER_TO_APPLY = [
Filter.LATITUDE_FILTER,
Filter.LONGITUDE_FILTER,
Filter.TIME_FILTER,
Filter.DEPTH_FILTER,
]
[docs]
class DataFilterConfig(BaseModel):
"""
Classe de configuration pour le filtrage des données.
:param min_latitude: La latitude minimale.
:type min_latitude: int | float
:param max_latitude: La latitude maximale.
:type max_latitude: int | float
:param min_longitude: La longitude minimale.
:type min_longitude: int | float
:param max_longitude: La longitude maximale.
:type max_longitude: int | float
:param min_depth: La profondeur minimale.
:type min_depth: int | float
:param max_depth: La profondeur maximale.
:type max_depth: int | float | None
"""
min_latitude: Optional[int | float] = MIN_LATITUDE
"""La latitude minimale."""
max_latitude: Optional[int | float] = MAX_LATITUDE
"""La latitude maximale."""
min_longitude: Optional[int | float] = MIN_LONGITUDE
"""La longitude minimale."""
max_longitude: Optional[int | float] = MAX_LONGITUDE
"""La longitude maximale."""
min_depth: Optional[int | float] = MIN_DEPTH
"""La profondeur minimale."""
max_depth: Optional[int | float] = MAX_DEPTH
"""La profondeur maximale."""
min_speed: Optional[int | float] = MIN_SPEED
"""La vitesse minimale."""
max_speed: Optional[int | float] = MAX_SPEED
"""La vitesse maximale."""
filter_to_apply: Optional[list[Filter]] = FILTER_TO_APPLY
"""Les filtres à appliquer."""
[docs]
@field_validator("min_latitude", "max_latitude")
def validate_latitude(cls, value: int | float) -> int | float:
"""
Valide la latitude.
:param value: La valeur de la latitude.
:type value: int | float
:return: La valeur de la latitude.
:rtype: int | float
:raises ValueError: Si la latitude n'est pas comprise entre MIN_LATITUDE et MAX_LATITUDE.
"""
if value < MIN_LATITUDE or value > MAX_LATITUDE:
raise ValueError(
f"La latitude doit être comprise entre {MIN_LATITUDE} et {MAX_LATITUDE}."
)
return value
[docs]
@field_validator("min_longitude", "max_longitude")
def validate_longitude(cls, value: int | float) -> int | float:
"""
Valide la longitude.
:param value: La valeur de la longitude.
:type value: int | float
:return: La valeur de la longitude.
:rtype: int | float
:raises ValueError: Si la longitude n'est pas comprise entre MIN_LONGITUDE et MAX_LONGITUDE.
"""
if value < MIN_LONGITUDE or value > MAX_LONGITUDE:
raise ValueError(
f"La longitude doit être comprise entre {MIN_LONGITUDE} et {MAX_LONGITUDE}."
)
return value
[docs]
@field_validator("min_depth", "max_depth")
def validate_depth(cls, value: int | float | None) -> int | float | None:
"""
Valide la profondeur.
:param value: La valeur de la profondeur.
:type value: int | float | None
:return: La valeur de la profondeur.
:rtype: int | float | None
:raises ValueError: Si la profondeur est inférieure à MIN_DEPTH.
"""
if value is not None and value < 0:
raise ValueError(
f"La profondeur doit être supérieure ou égale à {MIN_DEPTH}."
)
return value
[docs]
class GeoreferenceTideConfig(BaseModel):
"""
Classe de configuration pour le géoréférencement des données.
:param water_level_tolerance: Écart maximal en minutes entre les données et les niveaux d'eau à
récupérer pour le géoréférencement.
:type water_level_tolerance: str
"""
model_config = {"arbitrary_types_allowed": True}
water_level_tolerance: Optional[pd.Timedelta | str] = WATER_LEVEL_TOLERANCE
"""La tolérance en minutes pour les données de marée à récupérer pour le géoréférencement."""
[docs]
@field_validator("water_level_tolerance")
def validate_water_level_tolerance(
cls, value: str | pd.Timedelta | None
) -> pd.Timedelta:
"""
Valide la tolérance pour water level.
:param value: La tolérance pour water level.
:type value: str | None
:return: La tolérance pour water level.
:rtype: str
:raises ValueError: Si la tolérance pour water level n'est pas au bon format.
"""
if value == "" or value is None:
return WATER_LEVEL_TOLERANCE
if isinstance(value, pd.Timedelta):
return value
if value is not None:
pattern = re.compile(r"^\d+\s*(min|h)$")
if not pattern.match(value):
raise ValueError(
'La tolerance pour water level doit être au format "<number> <min|h>".'
)
return pd.Timedelta(value)
[docs]
class TVUConfig(BaseModel):
"""
Classe de configuration pour le calcul du TVU.
:param constant_tvu_wlo: La constante du TVU pour les niveaux d'eau WLO.
:type constant_tvu_wlo: Optional[float]
:param default_constant_tvu_wlp: La constante du TVU pour les niveaux d'eau WLP.
:type default_constant_tvu_wlp: Optional[float]
:param depth_coefficient_tvu: Le coefficient de profondeur pour le calcul du TVU.
:type depth_coefficient_tvu: Optional[float]
"""
constant_tvu_wlo: Optional[float | int] = CONSTANT_TVU_WLO
"""La constante du TVU pour les niveaux d'eau WLO."""
default_constant_tvu_wlp: Optional[float | int] = CONSTANT_TVU_WLP
"""La constante du TVU pour les niveaux d'eau WLP."""
depth_coefficient_tvu: Optional[float | int] = DEPTH_COEFFICIENT
"""Le coefficient de profondeur pour le calcul du TVU."""
default_depth_ssp_error_coefficient: Optional[float | int] = DEPTH_COEFFICIENT_SSP
"""Le coefficient d'erreur SSP par défaut."""
max_distance_ssp: Optional[float | int] = MAX_DISTANCE_SSP
"""La distance maximale pour liée une valeur de SSP."""
[docs]
@field_validator(
"depth_coefficient_tvu",
"constant_tvu_wlo",
"default_constant_tvu_wlp",
"default_depth_ssp_error_coefficient",
"max_distance_ssp",
)
def validate_positive(cls, value: Optional[float]) -> Optional[float]:
if value is not None and value < 0:
raise ValueError("La valeur doit être positive.")
return value
[docs]
class THUConfig(BaseModel):
"""
Classe de configuration pour le calcul du THU.
:param cone_angle_sonar: L'angle de cône du sonar pour le calcul du THU.
:type cone_angle_sonar: Optional[float]
:param constant_thu: La constante du THU.
:type constant_thu: Optional[float]
"""
cone_angle_sonar: Optional[float | int] = CONE_ANGLE_SONAR
"""L'angle de cône du sonar pour le calcul du THU."""
constant_thu: Optional[float | int] = CONSTANT_THU
"""La constante du THU."""
[docs]
@field_validator("cone_angle_sonar", "constant_thu")
def validate_positive(cls, value: Optional[float]) -> Optional[float]:
if value is not None and value < 0:
raise ValueError("La valeur doit être positive.")
return value
[docs]
class UncertaintyConfig(BaseModel):
"""
Classe de configuration pour le calcul de l'incertitude.
:param tvu: Configuration pour le calcul du TVU.
:type tvu: Optional[TVUConfig]
:param thu: Configuration pour le calcul du THU.
:type thu: Optional[THUConfig]
"""
tvu: Optional[TVUConfig] = TVUConfig()
"""Configuration pour le calcul du TVU."""
thu: Optional[THUConfig] = THUConfig()
"""Configuration pour le calcul du THU."""
[docs]
class DataGeoreferenceConfig(BaseModel):
"""
Classe de configuration pour le géoréférencement des données.
:param tide: Configuration pour le géoréférencement des données avec les niveaux d'eau.
:type tide: GeoreferenceTideConfig
:param uncertainty: Configuration pour le calcul de l'incertitude.
:type uncertainty: Optional[UncertaintyConfig]
"""
tide: GeoreferenceTideConfig = GeoreferenceTideConfig()
"""Configuration pour le géoréférencement des données avec les niveaux d'eau."""
uncertainty: Optional[UncertaintyConfig] = UncertaintyConfig()
"""Configuration pour le calcul de l'incertitude."""
[docs]
class VesselConfigManagerType(StrEnum):
"""
Enumération des types de gestionnaire de configuration de navires.
"""
VesselConfigJsonManager = "VesselConfigJsonManager"
"""Gestionnaire de configuration de navires en JSON."""
VesselConfigSQLiteManager = "VesselConfigSQLiteManager"
"""Gestionnaire de configuration de navires en SQLite."""
[docs]
class VesselManagerConfig(BaseModel):
"""
Classe de configuration pour le gestionnaire de navires.
:param manager_type: Le type de gestionnaire de configuration de navires.
:type manager_type: VesselConfigManagerType
:param kwargs: Les arguments pour le gestionnaire de configuration de navires.
:type kwargs: dict[str, Any]
"""
manager_type: Optional[VesselConfigManagerType]
"""Le type de gestionnaire de configuration de navires."""
kwargs: Optional[dict[str, Any]] = None
"""Les arguments pour le gestionnaire de configuration de navires."""
[docs]
class ExportConfig(BaseModel):
"""
Classe de configuration pour l'exportation des données.
"""
export_format: list[FileTypes] = EXPORT_FORMAT
"""Les formats de fichiers pour l'exportation."""
resolution: Optional[int | float] = RESOLUTION
"""La résolution pour l'exportation en GeoTIFF."""
group_by_iho_order: bool = False
"""Grouper les données par ordre IHO pour l'exportation."""
[docs]
@field_validator("resolution")
def validate_resolution(cls, value: Optional[int | float]) -> Optional[int | float]:
if value is None:
return RESOLUTION
if value <= 0:
raise ValueError("La résolution doit être positive.")
return value
[docs]
class OptionsConfig(BaseModel):
"""
Classe de configuration pour les options de traitement.
"""
log_level: str = INFO
"""Le niveau de log."""
max_iterations: int = MAX_ITERATIONS
"""Le nombre maximal d'itérations pour le traitement."""
decimal_precision: int = DECIMAL_PRECISION
"""La précision décimale pour les calculs."""
[docs]
@field_validator("max_iterations")
def validate_max_iterations(cls, value: int) -> int:
"""
Valide que max_iterations est plus grand que 0.
:param value: La valeur de max_iterations.
:type value: int
:return: La valeur de max_iterations.
:rtype: int
:raises ValueError: Si max_iterations est inférieur ou égal à 0.
"""
if value <= 0:
raise ValueError("Le paramètre max_iterations doit être supérieur à 0.")
return value
[docs]
@field_validator("decimal_precision")
def validate_decimal_precision(cls, value: int) -> int:
"""
Valide que decimal_precision est plus grand ou égale à 0.
:param value: La valeur de decimal_precision.
:type value: int
:return: La valeur de decimal_precision.
:rtype: int
:raises ValueError: Si decimal_precision est inférieur ou égal à 0.
"""
if value < 0:
raise ValueError(
"Le paramètre decimal_precision doit être supérieur ou égale à 0."
)
return value
[docs]
class PlotConfig(BaseModel):
"""
Classe de configuration pour les options de visualisation.
:param nbin_x: Le nombre de bins en X pour les heatmanps.
:type nbin_x: int
:param nbin_y: Le nombre de bins en Y pour les heatmanps.
:type nbin_y: int
"""
nbin_x: int = NBIN_X
"""Le nombre de bins en X pour les heatmanps."""
nbin_y: int = NBIN_Y
"""Le nombre de bins en Y pour les heatmanps."""
[docs]
@field_validator("nbin_x", "nbin_y")
def validate_nbin(cls, value: int) -> int:
"""
Valide que nbin_x et nbin_y sont plus grand que 0.
:param value: La valeur de nbin_x ou nbin_y.
:type value: int
:return: La valeur de nbin_x ou nbin_y.
:rtype: int
:raises ValueError: Si nbin_x ou nbin_y est inférieur ou égal à 0.
"""
if value <= 0:
raise ValueError("Le paramètre nbin_x et nbin_y doit être supérieur à 0.")
return value
[docs]
class CSBprocessingConfig(BaseModel):
"""
Classe de configuration pour la transformation des données et le géoréférencement.
:param filter: Configuration pour le filtrage des données.
:type filter: DataFilterConfig
:param georeference: Configuration pour le géoréférencement des données.
:type georeference: DataGeoreferenceConfig
:param vessel_manager: Configuration pour le gestionnaire de navires.
:type vessel_manager: Optional[VesselManagerConfig]
:param export: Configuration pour l'exportation des données.
:type export: ExportConfig
:param options: Configuration pour les options de traitement.
:type options: OptionsConfig
"""
filter: DataFilterConfig
"""Configuration pour le filtrage des données."""
georeference: DataGeoreferenceConfig
"""Configuration pour le géoréférencement des données."""
vessel_manager: Optional[VesselManagerConfig]
"""Configuration pour le gestionnaire de navires."""
export: ExportConfig = ExportConfig()
"""Configuration pour l'exportation des données."""
plot: PlotConfig = PlotConfig()
"""Configuration pour les options de visualisation."""
options: OptionsConfig = OptionsConfig()
"""Configuration pour les options de traitement."""
[docs]
def get_data_config(
config_file: Path,
) -> CSBprocessingConfig:
"""
Retournes la configuration pour la transformation des données et le géoréférencement.
:param config_file: Le chemin du fichier de configuration.
:type config_file: Path
:return: La configuration pour la transformation des données et le géoréférencement.
:rtype: tuple[DataFilterConfig, DataGeoreferenceConfig]
"""
config_data: CSBconfigDict = load_config(config_file=config_file)
LOGGER.debug(
f"Initialisation de la configuration de pour la transformation des données."
)
data_filter: ConfigDict = (
config_data.get("DATA", {}).get("Transformation", {}).get("filter")
)
data_georef_tide: ConfigDict = (
config_data.get("DATA", {}).get("Georeference", {}).get("water_level")
)
data_georef_tvu: ConfigDict = (
config_data.get("DATA", {})
.get("Georeference", {})
.get("uncertainty", {})
.get("tvu", {})
)
data_georef_thu: ConfigDict = (
config_data.get("DATA", {})
.get("Georeference", {})
.get("uncertainty", {})
.get("thu", {})
)
vessel_config: ConfigDict = (
config_data.get("CSB", {}).get("Processing", {}).get("vessel")
)
export_config: ConfigDict = (
config_data.get("CSB", {}).get("Processing", {}).get("export")
)
plot_config: ConfigDict = (
config_data.get("CSB", {}).get("Processing", {}).get("plot")
)
options_config: ConfigDict = (
config_data.get("CSB", {}).get("Processing", {}).get("options")
)
return CSBprocessingConfig(
filter=(
DataFilterConfig(
min_latitude=(data_filter.get("min_latitude") or MIN_LATITUDE),
max_latitude=(data_filter.get("max_latitude") or MAX_LATITUDE),
min_longitude=(data_filter.get("min_longitude") or MIN_LONGITUDE),
max_longitude=(data_filter.get("max_longitude") or MAX_LONGITUDE),
min_depth=(data_filter.get("min_depth") or MIN_DEPTH),
max_depth=(data_filter.get("max_depth") or MAX_DEPTH),
min_speed=(
data_filter.get("min_speed")
if data_filter.get("min_speed") is not None
else MIN_SPEED
),
max_speed=(
data_filter.get("max_speed", MAX_SPEED)
if data_filter.get("max_speed") is not None
else MAX_SPEED
),
filter_to_apply=data_filter.get("filter_to_apply", FILTER_TO_APPLY),
)
if data_filter
else DataFilterConfig()
),
georeference=(
DataGeoreferenceConfig(
tide=GeoreferenceTideConfig(
**data_georef_tide if data_georef_tide else GeoreferenceTideConfig()
),
uncertainty=UncertaintyConfig(
tvu=(
TVUConfig(**data_georef_tvu) if data_georef_tvu else TVUConfig()
),
thu=(
THUConfig(**data_georef_thu) if data_georef_thu else THUConfig()
),
),
)
),
vessel_manager=(
VesselManagerConfig(
manager_type=(
VesselConfigManagerType(vessel_config["manager_type"])
if "manager_type" in vessel_config
else None
),
kwargs={
key: value
for key, value in vessel_config.items()
if key != "manager_type"
},
)
if vessel_config
else None
),
export=(ExportConfig(**export_config) if export_config else ExportConfig()),
plot=(PlotConfig(**plot_config) if plot_config else PlotConfig()),
options=(
OptionsConfig(**options_config) if options_config else OptionsConfig()
),
)