"""
Module permettant de gérer la configuration des navires à partir d'un fichier JSON.
Ce module contient la classe VesselConfigJsonManager qui permet de gérer la configuration des navires à partir d'un fichier JSON.
"""
from datetime import datetime
import json
from pathlib import Path
from loguru import logger
import i18n
from .exception_vessel import VesselConfigNotFoundError
from .vessel_config_manager_abc import VesselConfigManagerABC
from .vessel_config import (
VesselConfig,
get_vessel_config_from_config_dict,
VesselConfigDict,
)
from . import vessel_ids as ids
LOGGER = logger.bind(name="CSB-Processing.Vessel.VesselConfigManager.JSON")
[docs]
class VesselConfigJsonManager(VesselConfigManagerABC):
"""
Classe permettant de gérer la configuration des navires à partir d'un fichier JSON.
"""
def __init__(self, json_config_path: Path | str):
"""
Initialisation du gestionnaire de configuration des navires à partir d'un fichier JSON.
:param json_config_path: Chemin du fichier JSON.
:type json_config_path: Path | str
"""
super().__init__()
self._vessel_configs = self._load_vessel_configs_file(
json_config_path=json_config_path
)
[docs]
@staticmethod
def _load_vessel_configs_file(json_config_path: Path) -> dict[str, VesselConfig]:
"""
Méthode permettant de charger la configuration des navires depuis un fichier JSON.
:param json_config_path: Chemin du fichier JSON.
:type json_config_path: Path
:return: Les configurations des navires.
:type: dict[str, VesselConfigDict]
:return: Configurations des navires.
:rtype: dict[str, VesselConfig]
:raises FileNotFoundError: Si le fichier de configuration des navires n'existe pas.
"""
json_config_path: Path = Path(json_config_path)
if not json_config_path.is_absolute():
json_config_path = Path(__file__).parent.parent / json_config_path
LOGGER.debug(
i18n.t(
"vessel.vessel_config_json_manager.loading_config_file",
json_config_path=json_config_path,
)
)
if not json_config_path.exists():
raise FileNotFoundError(
f"Le fichier de configuration des navires n'existe pas: {json_config_path}."
)
with open(json_config_path, "r") as file:
vessel_configs: list[VesselConfigDict] = json.load(file)
return {
vessel[ids.ID]: get_vessel_config_from_config_dict(vessel)
for vessel in vessel_configs
}
[docs]
def commit_vessel_configs(self, json_config_path: Path) -> None:
"""
Méthode permettant de sauvegarder la configuration des navires dans un fichier JSON.
:param json_config_path: Chemin du fichier JSON.
:type json_config_path: Path
:raises TypeError: Si un objet n'est pas sérialisable.
"""
def default_serializer(object_):
if isinstance(object_, datetime):
return object_.strftime("%Y-%m-%dT%H:%M:%S%z")
raise TypeError(
i18n.t(
"vessel.vessel_config_json_manager.not_serializable",
class_name=object_.__class__.__name__,
)
)
LOGGER.debug(
i18n.t(
"vessel.vessel_config_json_manager.saving_config_file",
json_config_path=json_config_path,
)
)
with open(json_config_path, "w") as file:
json.dump(
[config.model_dump() for config in self._vessel_configs.values()],
file, # type: ignore
indent=2,
default=default_serializer,
)
[docs]
def get_vessel_config(self, vessel_id: str) -> VesselConfig:
"""
Méthode permettant de récupérer la configuration d'un navire.
:param vessel_id: Identifiant du navire.
:type vessel_id: str
:return: Configuration du navire.
:rtype: VesselConfig
:raises VesselConfigNotFoundError: Si la configuration du navire n'existe pas.
"""
LOGGER.debug(
i18n.t(
"vessel.vessel_config_json_manager.getting_vessel_config",
vessel_id=vessel_id,
)
)
if vessel_id not in self._vessel_configs:
raise VesselConfigNotFoundError(vessel_id=vessel_id)
return self._vessel_configs[vessel_id]
[docs]
def get_vessel_configs(self) -> list[VesselConfig]:
"""
Méthode permettant de récupérer la configuration de tous les navires.
:return: Configurations des navires.
:rtype: list[VesselConfig]
"""
LOGGER.debug(
i18n.t("vessel.vessel_config_json_manager.getting_all_vessel_configs")
)
return [config for config in self._vessel_configs.values()]
[docs]
def add_veessel_config(self, vessel_config: VesselConfig) -> None:
"""
Méthode permettant d'ajouter la configuration d'un navire.
:param vessel_config: Configuration du navire.
:type vessel_config: VesselConfig
"""
LOGGER.debug(
i18n.t(
"vessel.vessel_config_json_manager.adding_vessel_config",
vessel_id=vessel_config.id,
)
)
self._vessel_configs[vessel_config.id] = vessel_config
[docs]
def update_vessel_config(self, vessel_id: str, vessel_config: VesselConfig) -> None:
"""
Méthode permettant de mettre à jour la configuration d'un navire.
:param vessel_id: Identifiant du navire.
:type vessel_id: str
:param vessel_config: Configuration du navire.
:type vessel_config: VesselConfig
"""
LOGGER.debug(
i18n.t(
"vessel.vessel_config_json_manager.updating_vessel_config",
vessel_id=vessel_id,
)
)
self._vessel_configs[vessel_id] = vessel_config
[docs]
def delete_vessel_config(self, vessel_id: str) -> None:
"""
Méthode permettant de supprimer la configuration d'un navire.
:param vessel_id: Identifiant du navire.
:type vessel_id: str
"""
LOGGER.debug(
i18n.t(
"vessel.vessel_config_json_manager.deleting_vessel_config",
vessel_id=vessel_id,
)
)
del self._vessel_configs[vessel_id]