"""
Module to provide the condition hints from their respective json file
as dictionary with the condition keys as keys and the hint texts as values.
"""
import asyncio
import inspect
import json
import logging
from abc import ABC, abstractmethod
from pathlib import Path
from typing import Any, Mapping, Optional
from efoli import EdifactFormat, EdifactFormatVersion
from ahbicht.condition_node_distinction import PACKAGE_1P_HINT_KEY
from ahbicht.content_evaluation.evaluationdatatypes import EvaluatableData
# pylint: disable = too-few-public-methods
from ahbicht.models.condition_nodes import Hint
from ahbicht.models.content_evaluation_result import ContentEvaluationResult
PACKAGE_1P_HINT_TEXT: str = (
"Hinweis: Das ist das Standardpaket, wenn keine Bedingung zum Tragen kommt, z.B. im COM-Segment."
)
"""
The hardcoded hint text for package '1P'.
Package '1P' is resolved to a synthetic hint with condition key PACKAGE_1P_HINT_KEY (9999).
This hint text is returned for that key, regardless of any configuration.
See PACKAGE_1P_HINT_KEY in condition_node_distinction.py for the full explanation of why
we use a synthetic hint key for package '1P'.
See also: https://github.com/Hochfrequenz/AHahnB/issues/715
"""
[docs]
class HintsProvider(ABC):
"""
A Hints Provider provides plain text hints for a given condition number.
This class provides the hints from the respective json file (defined by EdifactFormatVersion and EdifactFormat)
as dictionary with the condition keys as keys and the hint texts as values.
"""
edifact_format: EdifactFormat = NotImplementedError( # type: ignore[assignment]
"The inheriting class needs to define a format to which it is applicable."
)
edifact_format_version: EdifactFormatVersion = NotImplementedError( # type: ignore[assignment]
"The inheriting class needs to define a format version."
)
def __init__(self) -> None:
self.logger = logging.getLogger(self.__module__)
self.logger.setLevel(logging.DEBUG)
self.logger.info("Instantiated %s", self.__class__.__name__)
[docs]
@abstractmethod
async def get_hint_text(self, condition_key: str) -> Optional[str]:
"""
Get the hint text for the given condition key.
:param condition_key: e.g. "501"
:return: the corresponding hint text, e.g. "Data is thought to be interpreted as foo bar."
"""
raise NotImplementedError("The inheriting class has to implement this method")
[docs]
async def get_hints(self, condition_keys: list[str], raise_key_error: bool = True) -> dict[str, Hint]:
"""
Get Hints for given condition keys by asynchronously awaiting all self.get_hint_text at once
"""
results: list[Optional[str]]
if inspect.iscoroutinefunction(self.get_hint_text):
tasks = [self.get_hint_text(ck) for ck in condition_keys]
results = await asyncio.gather(*tasks)
else:
results = [self.get_hint_text(ck) for ck in condition_keys] # type: ignore[misc]
result: dict[str, Hint] = {}
for key, value in zip(condition_keys, results):
if value is None:
if raise_key_error:
raise KeyError(f"There seems to be no hint implemented with condition key '{key}'.")
else:
result[key] = Hint(hint=value, condition_key=key)
self.logger.debug("Found %i hints for %s", len(results), ", ".join(result.keys()))
return result
[docs]
class DictBasedHintsProvider(HintsProvider):
"""
A Hints Provider that is based on hardcoded values from a dictionary
"""
def __init__(self, results: Mapping[str, Optional[str]]) -> None:
"""
Initialize with a dictionary that contains all the Hinweis texts.
:param results:
"""
super().__init__()
self._all_hints: Mapping[str, Optional[str]] = results
[docs]
async def get_hint_text(self, condition_key: str) -> Optional[str]:
if not condition_key:
raise ValueError(f"The condition key must not be None/empty but was '{condition_key}'")
# Special case: Package '1P' hint key is always resolved to the hardcoded hint text.
# See PACKAGE_1P_HINT_KEY docstring for details.
if condition_key == PACKAGE_1P_HINT_KEY:
return PACKAGE_1P_HINT_TEXT
if condition_key in self._all_hints:
return self._all_hints[condition_key]
return None
[docs]
class JsonFileHintsProvider(DictBasedHintsProvider):
"""
The JsonFileHintsProvider loads hints from a JSON file.
"""
def __init__(
self, edifact_format: EdifactFormat, edifact_format_version: EdifactFormatVersion, file_path: Path
) -> None:
super().__init__(self._open_and_load_hint_json(file_path))
self.edifact_format = edifact_format
self.edifact_format_version = edifact_format_version
@staticmethod
def _open_and_load_hint_json(file_path: Path) -> dict[str, str]:
"""
Opens the hint json file and loads it into an attribute of the class.
"""
with open(file_path, encoding="utf-8") as json_infile:
return json.load(json_infile) # type: ignore[no-any-return]
[docs]
class ContentEvaluationResultBasedHintsProvider(HintsProvider):
"""
A hints provider that expects the evaluatable data to contain a ContentEvalutionResult as edifact seed.
Other than the DictBasedHintsProvider the outcome is not dependent on the initialization but on the evaluatable
data.
"""
def __init__(self, evaluatable_data: Optional[EvaluatableData[Any]] = None) -> None:
super().__init__()
self._evaluatable_data = evaluatable_data
[docs]
async def get_hint_text(self, condition_key: str) -> Optional[str]:
# Special case: Package '1P' hint key is always resolved to the hardcoded hint text.
# See PACKAGE_1P_HINT_KEY docstring for details.
if condition_key == PACKAGE_1P_HINT_KEY:
return PACKAGE_1P_HINT_TEXT
if self._evaluatable_data is None:
raise ValueError(
"ContentEvaluationResultBasedHintsProvider requires evaluatable_data. "
"Pass it in the constructor or use AhbContext."
)
content_evaluation_result = ContentEvaluationResult.model_validate(self._evaluatable_data.body)
try:
self.logger.debug("Retrieving hint '%s' from Content Evaluation Result", condition_key)
return content_evaluation_result.hints[condition_key]
except KeyError as key_error:
self.logger.debug("Hint '%s' was not contained in the CER", str(key_error))
return None