ahbicht.content_evaluation package

Submodules

ahbicht.content_evaluation.categorized_key_extract module

Contains the CategorizedKeyExtract and a schema for (de)serialization.

class ahbicht.content_evaluation.categorized_key_extract.CategorizedKeyExtract(hint_keys: List[str], format_constraint_keys: List[str], requirement_constraint_keys: List[str], package_keys: List[str], time_condition_keys: List[str])

Bases: object

A Categorized Key Extract contains those condition keys that are contained inside an expression. For expressions (that do not contain any unresolved package) it’s possible to pre-generate all possible outcomes of a content evaluation. CategorizedKeyExtract is also the answer to the question: ‘Which information do I need to provide in a ContentEvaluationResult in order to evaluate a given expression?’

format_constraint_keys: List[str]

list of keys for which you’ll need to provide EvaluatedFormatConstraints

generate_possible_content_evaluation_results() → List[ContentEvaluationResult]

A categorized key extract allows generating nearly all possible content evaluation results, except for hints, error messages, resolving packages.

hint_keys: List[str]

list of keys for which you’ll need to provide hint texts in a ContentEvaluationResult

package_keys: List[str]

list of packages that need to be resolved (additionally)

requirement_constraint_keys: List[str]

list of keys for which you’ll need to provide ConditionFulfilledValues

sanitize() → None

Sanitize the result (remove duplicates, sort keys)

time_condition_keys: List[str]

a list of time conditions, if present

class ahbicht.content_evaluation.categorized_key_extract.CategorizedKeyExtractSchema(*, only: Sequence[str] | AbstractSet[str] | None = None, exclude: Sequence[str] | AbstractSet[str] = (), many: bool = False, context: dict | None = None, load_only: Sequence[str] | AbstractSet[str] = (), dump_only: Sequence[str] | AbstractSet[str] = (), partial: bool | Sequence[str] | AbstractSet[str] | None = None, unknown: str | None = None)

Bases: Schema

A schema to (de)serialize CategorizedKeyExtractSchema

deserialize(data, **kwargs) → CategorizedKeyExtract

Converts the barely typed data dictionary into an actual CategorizedKeyExtractSchema

opts: SchemaOpts = <marshmallow.schema.SchemaOpts object>

ahbicht.content_evaluation.content_evaluation_result module

This module contains a class to store _all_ kinds of content evaluation results.

class ahbicht.content_evaluation.content_evaluation_result.ContentEvaluationResult(hints: Dict[str, str | None], format_constraints: Dict[str, EvaluatedFormatConstraint], requirement_constraints: Dict[str, ConditionFulfilledValue], packages: Dict[str, str] | None = None, id: UUID | None = None)

Bases: object

A class that holds the results of a full content evaluation (meaning all hints, requirement constraints and format constraints have been evaluated)

format_constraints: Dict[str, EvaluatedFormatConstraint]

maps the key of a format constraint to the respective evaluation result

hints: Dict[str, str | None]

maps the key of a hint (e.g. “501” to a hint text)

id: UUID | None

optional guid

packages: Dict[str, str] | None

maps the key of a package (e.g. ‘123’) to the respective expression (e.g. ‘[1] U ([2] O [3])’

requirement_constraints: Dict[str, ConditionFulfilledValue]

maps the key of a requirement_constraint to the respective evaluation result

class ahbicht.content_evaluation.content_evaluation_result.ContentEvaluationResultSchema(*, only: Sequence[str] | AbstractSet[str] | None = None, exclude: Sequence[str] | AbstractSet[str] = (), many: bool = False, context: dict | None = None, load_only: Sequence[str] | AbstractSet[str] = (), dump_only: Sequence[str] | AbstractSet[str] = (), partial: bool | Sequence[str] | AbstractSet[str] | None = None, unknown: str | None = None)

Bases: Schema

A schema to (de)serialize ContentEvaluationResults

deserialize(data, **kwargs) → ContentEvaluationResult

Converts the barely typed data dictionary into an actual ContentEvaluationResult

opts: SchemaOpts = <marshmallow.schema.SchemaOpts object>

ahbicht.content_evaluation.evaluationdatatypes module

Dataclasses that are relevant in the context of the content_evaluation.

class ahbicht.content_evaluation.evaluationdatatypes.EvaluatableData(body: _BodyT, edifact_format: EdifactFormat, edifact_format_version: EdifactFormatVersion)

Bases: Generic[_BodyT]

Data that can be processed/evaluated by an evaluator. They must not change during a content_evaluation run. The evaluatable data act as a flexible container to pass information that might be necessary for the content_evaluation. As of now (because our content_evaluation capabilities are quite limited) the only information provided is the meta seed of the message itself. But in the future the data provided might grow.

body: _BodyT

the body of the message that is being validated in the respective format (e.g. an edifact_seed)

edifact_format: EdifactFormat

the format of the evaluatable message (e.g. UTILMD)

edifact_format_version: EdifactFormatVersion

the format version of the evaluable data (e.g. FV2210)

class ahbicht.content_evaluation.evaluationdatatypes.EvaluatableDataProvider

Bases: Generic[_BodyT]

This is just a dummy class that is used for dependency injection. Use it to call binder.bind_to_provider(EvaluatableDataProvider, func_that_returns_evaluatable_data_goes_here) during dependency injection. See https://github.com/ivankorobkov/python-inject#why-no-scopes

class ahbicht.content_evaluation.evaluationdatatypes.EvaluationContext(scope: str | None)

Bases: object

A content_evaluation context describes the setting in which a condition shall be evaluated. The content_evaluation context might have different values for the same condition in one content_evaluation run. E.g. if the purpose of the condition is to make sure that every Zähler with zähler type “EHZ” has some properties the context of the content_evaluation is one zähler entry although there might be multiple zählers present in the message.

scope: str | None

ahbicht.content_evaluation.evaluationdatatypes.copy_evaluation_context(context: EvaluationContext) → EvaluationContext

Returns a deep copy of the provided context. This allows you to create a copy of a context instead of modifying the original context (as EvaluationContexts are “pass by reference”) :param context: :return: a deep copy of the context

ahbicht.content_evaluation.evaluator_factory module

A module to generate ready to use evaluators.

AHBicht, in the first place, is designed to do the content evaluation itself with actual code that checks if conditions are fulfilled, format constraints are met and gets hints from somewhere. However, AHBicht can also be used to do a “fake” content evaluation where the single truth values of conditions are already known. Then the approach with implementing single methods whose outcomes are already known/hardcoded at compile time seems like a big overhead. The code representation of “all outcomes are already known” is an instance of the ContentEvaluationResult. Now the methods below are useful. Simply provide a content evaluation result (the data) and the evaluators are created based on the already known outcomes. You do not have to actually touch any evaluator code.

ahbicht.content_evaluation.evaluator_factory.create_and_inject_hardcoded_evaluators(content_evaluation_result: ContentEvaluationResult, evaluatable_data_provider: Callable[[], EvaluatableData] | None = None, edifact_format: EdifactFormat | None = None, edifact_format_version: EdifactFormatVersion | None = None) → None

Creates evaluators from hardcoded content_evaluation result and injects them

Parameters:

content_evaluation_result –

Returns:

ahbicht.content_evaluation.evaluator_factory.create_content_evaluation_result_based_evaluators(edifact_format: EdifactFormat | None = None, edifact_format_version: EdifactFormatVersion | None = None) → Tuple[RcEvaluator, FcEvaluator, HintsProvider, PackageResolver]

Creates evaluators that expect the content evaluation result to be present in the evaluatble data

ahbicht.content_evaluation.evaluator_factory.create_hardcoded_evaluators(content_evaluation_result: ContentEvaluationResult, edifact_format: EdifactFormat | None = None, edifact_format_version: EdifactFormatVersion | None = None) → Tuple[RcEvaluator, FcEvaluator, HintsProvider, PackageResolver]

Creates evaluators based on the given content_evaluation_result

Parameters:

content_evaluation_result –

Returns:

ahbicht.content_evaluation.evaluators module

Evaluators are classes that evaluate AHB conditions, meaning: Based on a condition key and a message they return either FULFILLED (true), UNFULFILLED (false) or NEUTRAL (None). Their results are used as input for the condition validation of the entire message.

class ahbicht.content_evaluation.evaluators.Evaluator

Bases: ABC

Base of all evaluators. To implement a content_evaluation create or update an inheriting class that has edifact_format and edifact_format_version set accordingly. Then create a method named “evaluate_123” where “123” is the condition key of the condition it evaluates.

edifact_format: EdifactFormat = NotImplementedError('The inheriting class needs to define a format to which it is applicable.')

edifact_format_version: EdifactFormatVersion = NotImplementedError('The inheriting class needs to define a format version.')

get_evaluation_method(condition_key: str) → Callable | None

Returns the method that evaluates the condition with key condition_key :param condition_key: unique key of the condition, e.g. “59” :return: The method that can be used for content_evaluation; None if no such method is implemented.

ahbicht.content_evaluation.fc_evaluators module

A format constraint (FC) evaluator is an evaluator for format constraints. Think of stuff like: * strings that should match an OBIS regex * MarktlokationsIDs having correct check sums * pre/post decimal values having specific ranges Other than requirement constraints format constraints do not affect if data are required at all, but instead only validate already required data.

class ahbicht.content_evaluation.fc_evaluators.ContentEvaluationResultBasedFcEvaluator

Bases: FcEvaluator

A format constraint evaluator that expects the evaluatable data to contain a ContentEvalutionResult as edifact seed. Other than the DictBasedFcEvaluator the outcome is not dependent on the initialization but on the evaluatable data.

async evaluate_single_format_constraint(condition_key: str) → EvaluatedFormatConstraint

Evaluates the format constraint with the given key. :param condition_key: key of the condition, e.g. “950” :return: If the format constraint is fulfilled and an optional error_message.

get_evaluation_method(condition_key: str) → Callable | None

Returns the method that evaluates the condition with key condition_key :param condition_key: unique key of the condition, e.g. “59” :return: The method that can be used for content_evaluation; None if no such method is implemented.

class ahbicht.content_evaluation.fc_evaluators.DictBasedFcEvaluator(results: Dict[str, EvaluatedFormatConstraint])

Bases: FcEvaluator

A format constraint evaluator that is initialized with a prefilled dictionary on time of creation. Once initialized the outcome of the evaluation won’t change anymore.

async evaluate_single_format_constraint(condition_key: str) → EvaluatedFormatConstraint

Evaluates the format constraint with the given key. :param condition_key: key of the condition, e.g. “950” :return: If the format constraint is fulfilled and an optional error_message.

get_evaluation_method(condition_key: str) → Callable | None

Returns the method that evaluates the condition with key condition_key :param condition_key: unique key of the condition, e.g. “59” :return: The method that can be used for content_evaluation; None if no such method is implemented.

class ahbicht.content_evaluation.fc_evaluators.FcEvaluator

Bases: Evaluator, ABC

Base of all Format Constraint (FC) evaluators. To implement a content_evaluation create or update an inheriting class that has edifact_format and edifact_format_version set accordingly. Then create a method named “evaluate_123” where “123” is the condition key of the condition it evaluates.

evaluate_931(entered_input: str) → EvaluatedFormatConstraint

Assert that the entered input is parsable as datetime with explicit UTC offset. Then assert that the UTC offset is exactly +00:00. Be aware of the fact that asserting on a fixed offset when both datetime + offset are given will not lead to any truly meaningful results. We implement it for compatability but don’t encourage you to actively write any conditions that use it.

evaluate_932(entered_input: str) → EvaluatedFormatConstraint

Assert that the entered input is the start/end of a german “Stromtag” (during central european daylight saving time). We ship this predefined method to evaluate the format constraints which are being introduced by expanding “time conditions” (UB1/UB3) in the expression_resolver.TimeConditionTransformer.

evaluate_933(entered_input: str) → EvaluatedFormatConstraint

Assert that the entered input is the start/end of a german “Stromtag” (during central european standard time). We ship this predefined method to evaluate the format constraints which are being introduced by expanding “time conditions” (UB1/UB3) in the expression_resolver.TimeConditionTransformer.

evaluate_934(entered_input: str) → EvaluatedFormatConstraint

Assert that the entered input is the start/end of a german “Gastag” (during central european daylight saving time). We ship this predefined method to evaluate the format constraints which are being introduced by expanding “time conditions” (UB2/UB3) in the expression_resolver.TimeConditionTransformer.

evaluate_935(entered_input: str) → EvaluatedFormatConstraint

Assert that the entered input is the start/end of a german “Gastag” (during central european standard time). We ship this predefined method to evaluate the format constraints which are being introduced by expanding “time conditions” (UB2/UB3) in the expression_resolver.TimeConditionTransformer.

async evaluate_format_constraints(condition_keys: List[str]) → Dict[str, EvaluatedFormatConstraint]

Evaluate the entered_input in regard to all the formats provided in condition_keys.

async evaluate_single_format_constraint(condition_key: str) → EvaluatedFormatConstraint

Evaluates the format constraint with the given key. :param condition_key: key of the condition, e.g. “950” :return: If the format constraint is fulfilled and an optional error_message.

ahbicht.content_evaluation.fc_evaluators.text_to_be_evaluated_by_format_constraint: ContextVar[str | None] = <ContextVar name='text_to_be_evaluated_by_format_constraint' default=None>

This context variable holds the text that is to be analysed/evaluated by the format constraint evaluator. It will always return the “correct” value in your context. You only have to manually set this context variable if you’re evaluating an expression outside the validation framework. The conceptual difference to the EvaluatableData which are dependency injected using the EvaluatableDataProvider is, that the data evaluated in a format constraint (via the context variable) vary over the time span of one validation run. The EvaluatableData are stable in that regard.

ahbicht.content_evaluation.german_strom_and_gas_tag module

A module to evaluate datetimes and whether they are “on the edge” of a German “Stromtag” or “Gastag” respectively

ahbicht.content_evaluation.german_strom_and_gas_tag.has_no_utc_offset(entered_input: str) → EvaluatedFormatConstraint

returns true if and only if entered_input is parsable as date time with explicit offset which has no UTC offset. This means the UTC offset is exactly “+00:00”.

ahbicht.content_evaluation.german_strom_and_gas_tag.is_gastag_limit(date_time: datetime) → bool

Returns true if and only if the given date_time is the inclusive start/exclusive end of a german “Gastag”. The “Gastag” is the balancing relevant day in the German gas market. It starts and ends at 6am in German local time which can be either 4AM or 5AM in UTC (depending on the daylight saving time in Germany).

ahbicht.content_evaluation.german_strom_and_gas_tag.is_stromtag_limit(date_time: datetime) → bool

Returns true if and only if the given date_time is the inclusive start/exclusive end of a german “Stromtag”. The “Stromtag” is the balancing relevant day in the German electricity market. It starts and ends at midnight in German local time which can be either 23PM or 22PM in UTC (depending on the daylight saving time in Germany).

ahbicht.content_evaluation.german_strom_and_gas_tag.is_xtag_limit(entered_input: str, division: Literal['Strom'] | Literal['Gas']) → EvaluatedFormatConstraint

Tries to parse the entered_input as datetime and checks if it is the start/end of a Strom- or Gastag

ahbicht.content_evaluation.german_strom_and_gas_tag.parse_as_datetime(entered_input: str) → Tuple[datetime | None, EvaluatedFormatConstraint | None]

Try to parse the given entered_input as datetime :param entered_input: a string :return: A tuple with the first item being a datetime if parsing was successful or the second item being an (erroneous) EvaluatedFormatConstraint if the parsing was not successful.

ahbicht.content_evaluation.rc_evaluators module

Requirement Constraint (RC) Evaluators are evaluators that check if data are required under given circumstances. Typical use-cases are for example * you must only provide a Gerätenummer if the Transaktionsgrund is e.g. ‘E08’ * you must only provide an Ausbaudatum if the meter is being removed e.g. ‘Z02’

class ahbicht.content_evaluation.rc_evaluators.ContentEvaluationResultBasedRcEvaluator

Bases: RcEvaluator

A requirement constraint evaluator that expects the evaluatable data to contain a (dumped) ContentEvalutionResult. Other than the DictBasedRcEvaluator the outcome is not dependent on the initialization but on the evaluatable data.

async evaluate_single_condition(condition_key: str, evaluatable_data: EvaluatableData, context: EvaluationContext | None = None) → ConditionFulfilledValue

Evaluates the condition with the given key. :param evaluatable_data: data of the edifact message/transaction :param context: optional content_evaluation context, if not the default context will be used :param condition_key: key of the condition, e.g. “78” :return:

class ahbicht.content_evaluation.rc_evaluators.DictBasedRcEvaluator(results: Dict[str, ConditionFulfilledValue])

Bases: RcEvaluator

A requirement constraint evaluator that is initialized with a prefilled dictionary. The outcome of the evaluation does not change anymore after the initialization.

async evaluate_single_condition(condition_key: str, evaluatable_data: EvaluatableData, context: EvaluationContext | None = None) → ConditionFulfilledValue

Evaluates the condition with the given key. :param evaluatable_data: data of the edifact message/transaction :param context: optional content_evaluation context, if not the default context will be used :param condition_key: key of the condition, e.g. “78” :return:

class ahbicht.content_evaluation.rc_evaluators.RcEvaluator

Bases: Evaluator, ABC

Base class of all Requirement Constraint (RC) evaluators. To implement a content_evaluation create or update an inheriting class that has edifact_format and edifact_format_version set accordingly. Then create a method named “evaluate_123” where “123” is the condition key of the condition it evaluates.

async evaluate_conditions(condition_keys: List[str], evaluatable_data: EvaluatableData, condition_keys_with_context: Dict[str, EvaluationContext] | None = None) → Dict[str, ConditionFulfilledValue]

Validate all the conditions provided in condition_keys in their respective context.

async evaluate_single_condition(condition_key: str, evaluatable_data: EvaluatableData, context: EvaluationContext | None = None) → ConditionFulfilledValue

Evaluates the condition with the given key. :param evaluatable_data: data of the edifact message/transaction :param context: optional content_evaluation context, if not the default context will be used :param condition_key: key of the condition, e.g. “78” :return:

ahbicht.content_evaluation.token_logic_provider module

Contains a class that is able to provide any RC/FC Evaluator, HintsProvider or Package Resolver.

class ahbicht.content_evaluation.token_logic_provider.SingletonTokenLogicProvider(inputs: List[Evaluator | PackageResolver | HintsProvider])

Bases: TokenLogicProvider

An TokenLogicProvider that is instantiated with a list of evaluators/providers/resolvers of which the same instances will be used during the entire uptime of the application (singleton style).

get_fc_evaluator(edifact_format: EdifactFormat | None = None, format_version: EdifactFormatVersion | None = None) → FcEvaluator

returns an appropriate FC Evaluator for the given edifact_format and edifact_format_version. The implementing class shall raise a NotImplementedError and not return None.

get_hints_provider(edifact_format: EdifactFormat | None = None, format_version: EdifactFormatVersion | None = None) → HintsProvider

returns an appropriate HintsProvider for the given edifact_format and edifact_format_version. The implementing class shall raise a NotImplementedError and not return None.

get_package_resolver(edifact_format: EdifactFormat | None = None, format_version: EdifactFormatVersion | None = None) → PackageResolver

Returns an appropriate PackageResolver for the given edifact_format and edifact_format_version. You can provide None as format(version) if there is only 1 package resolver available The implementing class shall raise a NotImplementedError and not return None.

get_rc_evaluator(edifact_format: EdifactFormat | None = None, format_version: EdifactFormatVersion | None = None) → RcEvaluator

returns an appropriate RC Evaluator for the given edifact_format and edifact_format_version. The implementing class shall raise a NotImplementedError and not return None.

class ahbicht.content_evaluation.token_logic_provider.TokenLogicProvider

Bases: ABC

A TokenLogicProvider is a class that can provide you with the correct evaluator/hints provider/package resolver for your use case/for any token in the parsed expression tree.

abstract get_fc_evaluator(edifact_format: EdifactFormat, format_version: EdifactFormatVersion) → FcEvaluator

returns an appropriate FC Evaluator for the given edifact_format and edifact_format_version. The implementing class shall raise a NotImplementedError and not return None.

abstract get_hints_provider(edifact_format: EdifactFormat | None = None, format_version: EdifactFormatVersion | None = None) → HintsProvider

returns an appropriate HintsProvider for the given edifact_format and edifact_format_version. The implementing class shall raise a NotImplementedError and not return None.

abstract get_package_resolver(edifact_format: EdifactFormat | None = None, format_version: EdifactFormatVersion | None = None) → PackageResolver

Returns an appropriate PackageResolver for the given edifact_format and edifact_format_version. You can provide None as format(version) if there is only 1 package resolver available The implementing class shall raise a NotImplementedError and not return None.

abstract get_rc_evaluator(edifact_format: EdifactFormat, format_version: EdifactFormatVersion) → RcEvaluator

returns an appropriate RC Evaluator for the given edifact_format and edifact_format_version. The implementing class shall raise a NotImplementedError and not return None.

Module contents

functions related to content evaluation

async ahbicht.content_evaluation.is_valid_expression(expression_or_tree: str | Tree[Token], content_evaluation_result_setter: Callable[[ContentEvaluationResult], Any]) → Tuple[bool, str | None]

Returns true iff the given expression is both well-formed and valid. An expression is valid if and only if all possible content evaluations lead to a meaningful results. ⚠ This only works if the provided content_evaluation_result_setter writes the EvaluatableData in such a way, that the injected Evaluators (FC/RC/Hints/Package) can work with it. This is easiest for the ContentEvaluationResultBased FC/RC/Hints/Package token logic providers. :param content_evaluation_result_setter: a threadsafe method that writes the given Content Evaluation Result into the evaluatable data :param expression_or_tree: “Muss [1] U [2]” (returns True) “Muss ([61]u [584]) o[583]” (returns False) :return: (True,None) iff the expression is valid; (False, error message) otherwise