# Copyright 2026 Daitum
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
Module for defining and managing model objects like calculations and parameters.
This module contains the `Calculation` and `Parameter` classes, which represent two of the core
components in a model. These objects are used to define calculations and parameters, including
their data types, associated formulas, and additional properties such as whether they are
model-level objects or required by output.
Key Classes:
- `Calculation`: Represents a calculation with a unique `id`, a `Formula`, and
various flags indicating the calculation's dependencies and scope.
- `Parameter`: Represents a parameter in the model, consisting of a `id`,
`data_type`, and a `value`, along with flags indicating model-level attributes.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any
from typeguard import typechecked
from ._buildable import Buildable
from ._helpers import _validate_name
from .data_types import BaseDataType, DataType, MapDataType, ObjectDataType
from .formula import Formula, Operand
if TYPE_CHECKING:
from .model import ModelBuilder
from .validator import Severity, Validator
@dataclass
@typechecked
class ValidationValuesContainer:
"""
Container for the calculations generated by a validator.
Attributes:
severity: The severity level of the associated validator.
invalid_value: The calculation that evaluates to True when the named value is invalid.
message_value: The calculation that contains the error message when invalid.
summary_message: Summary error message.
"""
severity: Severity
invalid_value: NamedValue
message_value: NamedValue
summary_message: str
# pylint: disable=too-many-instance-attributes
[docs]
@typechecked
class Calculation(Buildable, Operand):
"""
A calculation is one of the three basic objects that make up a V3 Model. It consists of a name,
data_type, and a formula.
Attributes:
id (str): The id or ID of the calculation.
formula (model_generator.structures.Formula): The formula defining the calculation.
depends_on_decision (bool): Indicates if the calculation depends on a decision.
model_level (bool): Indicates if the calculation is at the model level.
required_by_output (bool): Indicates if the calculation is required by the output.
"""
[docs]
def __init__(
self,
id: str,
formula: Formula,
model_level: bool = False,
model: ModelBuilder | None = None,
):
"""
Args:
id: Unique identifier for this calculation. Must be a valid Python-style identifier.
formula: The formula that defines the calculated value.
model_level: If ``True``, the calculation is model-level; otherwise scenario-level.
model: The ``ModelBuilder`` this calculation belongs to. Can be set later via
``ModelBuilder.add_calculation``.
"""
_validate_name(id, "named value id")
self._id = id
self.formula = formula
self.data_type = formula.data_type
self.depends_on_decision: bool = False
self.model_level = model_level
self.required_by_output: bool = False
self._tracking_group: str | None = None
self._model = model
self._validators: list[Validator] = []
self._combined_message_value: NamedValue | None = None
@property
def id(self) -> str:
"""The unique identifier for this calculation."""
return self._id
@property
def tracking_group(self) -> str | None:
"""The tracking group identifier, or ``None`` if change tracking is not enabled."""
return self._tracking_group
@property
def tracking_id(self) -> str:
"""The ID of the corresponding tracking calculation, or an empty string if not tracked."""
return "" if self._tracking_group is None else self._tracking_group + "_TRACKING_" + self.id
[docs]
def set_tracking_group(self, group: str | None) -> Calculation:
"""Sets the tracking group for this calculation. Returns self."""
self._tracking_group = group
return self
[docs]
def set_depends_on_decision(self, depends_on_decision: bool) -> Calculation:
"""Sets whether this calculation depends on a decision variable."""
self.depends_on_decision = depends_on_decision
return self
[docs]
def set_required_by_output(self, required_by_output: bool) -> Calculation:
"""Sets whether this calculation is required by the output."""
self.required_by_output = required_by_output
return self
[docs]
def to_string(self) -> str:
return self.id
[docs]
def to_data_type(self) -> BaseDataType:
return self.data_type
[docs]
def add_validator(self, validator: Validator) -> None:
"""
Attach a validator to this named value.
Each named value can only have one validator per severity level.
Attempting to add another validator with the same severity
will raise a ValueError.
Args:
validator (Validator): The validator instance to attach.
Raises:
ValueError: If a validator with the same severity has
already been added to this field.
"""
if not self._model:
raise NotImplementedError("Model is not yet defined.")
if validator.severity in [val.severity for val in self._validators]:
raise ValueError(f"Duplicate severity validator detected for calculation: {self.id}")
self._validators.append(validator)
validator._attach_to_named_value(self, self._model) # pylint: disable=protected-access
[docs]
def get_validation_values(
self, severity: Severity | None = None
) -> list[ValidationValuesContainer] | ValidationValuesContainer | None:
"""
Retrieve the validation values associated with this named value's validators.
Args:
severity: If provided, returns only the validation values for that severity level.
If None, returns validation values for all attached validators.
Returns:
- A list of ``ValidationValuesContainer`` if ``severity`` is None and validators exist.
- A single ``ValidationValuesContainer`` if a validator matching ``severity`` is found.
- ``None`` if no validators are attached, or if no validator matches the given severity.
"""
from .validator import SEVERITY_RANK # pylint: disable=import-outside-toplevel
if not self._validators:
return None
if not self._model:
raise NotImplementedError("Model is not yet defined.")
# return all validations' values in a list
if not severity:
validation_values = []
for validator in self._validators:
validation_values.append(
ValidationValuesContainer(
validator.severity,
self._model.get_named_value(
f"{self.id}__invalid__{validator.severity.value}"
),
self._model.get_named_value(
f"{self.id}__message__{validator.severity.value}"
),
validator.summary_message,
)
)
return sorted(validation_values, key=lambda v: SEVERITY_RANK.get(v.severity, 0))
# return the corresponding validation fields
for validator in self._validators:
if validator.severity == severity:
return ValidationValuesContainer(
validator.severity,
self._model.get_named_value(f"{self.id}__invalid__{severity.value}"),
self._model.get_named_value(f"{self.id}__message__{severity.value}"),
validator.summary_message,
)
return None
[docs]
def get_combined_message_value(self) -> NamedValue | None:
"""
Returns a single calculated named value that combines all per-severity message values
into one array, ordered from highest to lowest severity (CRITICAL → ERROR → WARNING → INFO).
Blank messages are excluded from the array.
The resulting value is registered on the model with the ID
``{id}__message__combined`` and cached so subsequent calls return the same
value without rebuilding it.
Returns:
NamedValue | None: The combined message calculated value, or ``None`` if this named
value has no validators.
Raises:
NotImplementedError: If this named value has not yet been attached to a model.
"""
return _create_combined_message_value(self)
# pylint: disable=too-many-instance-attributes
[docs]
@typechecked
class Parameter(Buildable, Operand):
"""
A parameter is one of the three basic objects that make up a model. It is the simplest of
these three objects and consists only of a data_type, a id, as well as optional
import_format and model_level.
Attributes:
id (str): The id or ID of the parameter.
data_type (DataType): The type of data that the parameter takes.
model_level (bool): Indicates if the parameter is at the model level.
"""
[docs]
def __init__(
self,
id: str,
data_type: BaseDataType,
value: Any | None = None,
model_level: bool = False,
):
"""
Args:
id: Unique identifier for this parameter. Must be a valid Python-style identifier.
data_type: The data type of the parameter value.
value: The initial value of the parameter, or ``None`` if not yet set.
model_level: If ``True``, the parameter is model-level; otherwise scenario-level.
"""
_validate_name(id, "named value id")
self.data_type = data_type
self._id = id
self._value = value
self.model_level = model_level
self.import_format: str | None = None
self._tracking_group: str | None = None
self._model: ModelBuilder | None = None
self._validators: list[Validator] = []
self._combined_message_value: NamedValue | None = None
@property
def id(self) -> str:
"""The unique identifier for this parameter."""
return self._id
@property
def tracking_group(self) -> str | None:
"""The tracking group identifier, or ``None`` if change tracking is not enabled."""
return self._tracking_group
@property
def tracking_id(self) -> str:
"""The ID of the corresponding tracking parameter, or an empty string if not tracked."""
return "" if self._tracking_group is None else self._tracking_group + "_TRACKING_" + self.id
[docs]
def set_tracking_group(self, group: str | None) -> Parameter:
"""Sets the tracking group for this parameter. Returns self."""
self._tracking_group = group
return self
[docs]
def set_model(self, model: ModelBuilder) -> Parameter:
"""Attach this parameter to a ``ModelBuilder``. Returns self."""
self._model = model
return self
[docs]
def to_string(self) -> str:
return self.id
[docs]
def to_data_type(self) -> BaseDataType:
return self.data_type
[docs]
def to_named_value_dict(self) -> dict[str, Any]:
"""
Converts the `Parameter` object to a dictionary representation for JSON serialisation.
Returns:
dict[str, Any]: A dictionary representation of the `Parameter` object.
"""
if isinstance(self.data_type, (ObjectDataType, MapDataType)):
data_type_built = self.data_type.build()
type_str = data_type_built["type"]
table_id = data_type_built.get("tableId")
else:
assert isinstance(self.data_type, DataType)
type_str = self.data_type.value
table_id = None
return {
"@type": type_str,
"value": self._value,
"error": False,
"tableId": table_id,
}
[docs]
def add_validator(self, validator: Validator) -> None:
"""
Attach a validator to this named value.
Each named value can only have one validator per severity level.
Attempting to add another validator with the same severity
will raise a ValueError.
Args:
validator (Validator): The validator instance to attach.
Raises:
ValueError: If a validator with the same severity has
already been added to this field.
"""
if not self._model:
raise NotImplementedError("Model is not yet defined.")
if validator.severity in [val.severity for val in self._validators]:
raise ValueError(f"Duplicate severity validator detected for parameter: {self.id}")
self._validators.append(validator)
validator._attach_to_named_value(self, self._model) # pylint: disable=protected-access
[docs]
def get_validation_values(
self, severity: Severity | None = None
) -> list[ValidationValuesContainer] | ValidationValuesContainer | None:
"""
Retrieve the validation values associated with this named value's validators.
Args:
severity: If provided, returns only the validation values for that severity level.
If None, returns validation values for all attached validators.
Returns:
- A list of ``ValidationValuesContainer`` if ``severity`` is None and validators exist.
- A single ``ValidationValuesContainer`` if a validator matching ``severity`` is found.
- ``None`` if no validators are attached, or if no validator matches the given severity.
"""
from .validator import SEVERITY_RANK # pylint: disable=import-outside-toplevel
if not self._validators:
return None
if not self._model:
raise NotImplementedError("Model is not yet defined.")
# return all validations' values in a list
if not severity:
validation_values = []
for validator in self._validators:
validation_values.append(
ValidationValuesContainer(
validator.severity,
self._model.get_named_value(
f"{self.id}__invalid__{validator.severity.value}"
),
self._model.get_named_value(
f"{self.id}__message__{validator.severity.value}"
),
validator.summary_message,
)
)
return sorted(validation_values, key=lambda v: SEVERITY_RANK.get(v.severity, 0))
# return the corresponding validation fields
for validator in self._validators:
if validator.severity == severity:
return ValidationValuesContainer(
validator.severity,
self._model.get_named_value(f"{self.id}__invalid__{severity.value}"),
self._model.get_named_value(f"{self.id}__message__{severity.value}"),
validator.summary_message,
)
return None
[docs]
def get_combined_message_value(self) -> NamedValue | None:
"""
Returns a single calculated named value that combines all per-severity message values
into one array, ordered from highest to lowest severity (CRITICAL → ERROR → WARNING → INFO).
Blank messages are excluded from the array.
The resulting value is registered on the model with the ID
``{id}__message__combined`` and cached so subsequent calls return the same
value without rebuilding it.
Returns:
NamedValue | None: The combined message calculated value, or ``None`` if this named
value has no validators.
Raises:
NotImplementedError: If this named value has not yet been attached to a model.
"""
return _create_combined_message_value(self)
# pylint: disable=protected-access
def _create_combined_message_value(named_value: NamedValue) -> NamedValue | None:
"""
Build (or return the cached) combined-message calculation for *named_value*.
Combines all per-severity message values into a single ``STRING_ARRAY`` calculation,
ordered from highest to lowest severity. Blank entries are excluded.
Args:
named_value: The ``Calculation`` or ``Parameter`` whose validators' message values
should be combined.
Returns:
The combined-message ``Calculation``, or ``None`` if *named_value* has no validators.
Raises:
NotImplementedError: If *named_value* has not yet been attached to a model.
"""
# to avoid circular import
from daitum_model import formulas # pylint: disable=import-outside-toplevel
from .validator import SEVERITY_RANK # pylint: disable=import-outside-toplevel
if not named_value._validators:
return None
if not named_value._model:
raise NotImplementedError("Model is not yet defined.")
if named_value._combined_message_value is not None:
return named_value._combined_message_value
# Sort highest severity first so the combined array is ordered
sorted_validators = sorted(
named_value._validators, key=lambda v: SEVERITY_RANK.get(v.severity, 0), reverse=True
)
msg_values = [
named_value._model.get_named_value(f"{named_value.id}__message__{v.severity.value}")
for v in sorted_validators
]
# ignore_null=True so blank messages are excluded from the array
formula = formulas.ARRAY(True, *msg_values)
combined_value_id = f"{named_value.id}__message__combined"
named_value._combined_message_value = named_value._model.add_calculation(
combined_value_id, formula
)
return named_value._combined_message_value
NamedValue = Calculation | Parameter