data_handlers

data_handlers ¶

Data handlers for generating values, random variables, and other data types.

CategoricalDistribution ¶

Bases: RandomVariable[T]

Categorical distribution for discrete outcomes with specified probabilities.

A categorical distribution (also called a discrete distribution) describes the probability of obtaining one of k possible outcomes. Each outcome has a probability between 0 and 1, and all probabilities must sum to 1.

If probabilities are not specified, defaults to equal probabilities for all categories (uniform discrete distribution).

Key Properties

Support is finite set of categories
PMF gives probability of each category
CDF is step function

Attributes:

Name	Type	Description
`categories`	`ndarray`	Array of possible outcomes (any type)
`probabilities`	`ndarray`	Probability for each category
`name`	`str`	Identifier for this distribution instance
`seed`	`Optional[int]`	Random seed for reproducible sampling
`replace`	`bool`	Whether or not to allow multiple draws of the same value (allowed if True)

Raises:

Type	Description
`ValueError`	If probabilities don't sum to 1
`ValueError`	If lengths of categories and probabilities don't match
`ValueError`	If any probability is negative

cdf ¶

cdf(x: ndarray, **kwargs) -> NDArray[Any, float]

Evaluate the cumulative distribution function.

For categorical distributions, this is a step function that increases at each category by that category's probability.

Parameters:

Name	Type	Description	Default
`x`	`ndarray`	Points at which to evaluate the CDF	required

Other Parameters:

Name	Type	Description
`squeeze`	`bool`	Whether to remove unnecessary dimensions. Added by RandomVariableMeta. Defaults is `True`.

Returns:

Type	Description
`NDArray[Any, float]`	NDArray[Any,float]: CDF values at the input points

Source code in src/data_handlers/random_variables.py

def cdf(self, x: np.ndarray, **kwargs) -> NDArray[Any,float]:
    """Evaluate the cumulative distribution function.

    For categorical distributions, this is a step function that
    increases at each category by that category's probability.

    Args:
        x (np.ndarray): Points at which to evaluate the CDF

    Other Parameters:
        squeeze (bool, optional): Whether to remove unnecessary dimensions.
            Added by RandomVariableMeta. Defaults is `True`.

    Returns:
        NDArray[Any,float]: CDF values at the input points
    """
    return np.array([
        np.sum(self.probabilities[self.categories <= val])
        for val in x
    ])

pdf ¶

pdf(x: ndarray, **kwargs) -> NDArray[Any, float]

Evaluate the probability mass function (PMF).

For categorical distributions, this gives the probability of each category occurring.

Parameters:

Name	Type	Description	Default
`x`	`ndarray`	Points at which to evaluate the PMF	required

Other Parameters:

Name	Type	Description
`squeeze`	`bool`	Whether to remove unnecessary dimensions. Added by RandomVariableMeta. Defaults is `True`.

Returns:

Type	Description
`NDArray[Any, float]`	NDArray[Any,float]: Probability of each input value

Source code in src/data_handlers/random_variables.py

def pdf(self, x: np.ndarray, **kwargs) -> NDArray[Any,float]:
    """Evaluate the probability mass function (PMF).

    For categorical distributions, this gives the probability of
    each category occurring.

    Args:
        x (np.ndarray): Points at which to evaluate the PMF

    Other Parameters:
        squeeze (bool, optional): Whether to remove unnecessary dimensions.
            Added by RandomVariableMeta. Defaults is `True`.

    Returns:
        NDArray[Any,float]: Probability of each input value
    """
    return np.array([
        self.probabilities[self.categories == val].item()
        if val in self.categories else 0.0
        for val in x
    ])

sample ¶

sample(size: int = 1, **kwargs) -> NDArray[Any, T]

Generate random samples from the categorical distribution.

Parameters:

Name	Type	Description	Default
`size`	`int`	Number of samples to generate. Defaults to 1.	`1`

Other Parameters:

Name	Type	Description
`squeeze`	`bool`	Whether to remove unnecessary dimensions. Added by RandomVariableMeta. Defaults is `True`.

Notes

The squeeze parameter is added automatically by the metaclass and does not appear in the function signature, but can be passed as a keyword argument.

Returns:

Type	Description
`NDArray[Any, T]`	NDArray[Any,T]: Array of samples from the categories

Source code in src/data_handlers/random_variables.py

def sample(self, size: int = 1, **kwargs) -> NDArray[Any,T]:
    """Generate random samples from the categorical distribution.

    Args:
        size (int, optional): Number of samples to generate. Defaults to 1.

    Other Parameters:
        squeeze (bool, optional): Whether to remove unnecessary dimensions.
            Added by RandomVariableMeta. Defaults is `True`.

    Notes:
        The squeeze parameter is added automatically by the metaclass and does not appear
        in the function signature, but can be passed as a keyword argument.

    Returns:
        NDArray[Any,T]: Array of samples from the categories
    """
    rng = np.random.default_rng(self.seed)
    return rng.choice(self.categories, size=size, p=self.probabilities, replace=self.replace)

validate_and_set_probabilities ¶

validate_and_set_probabilities() -> CategoricalDistribution

Validate probability values and set defaults if needed.

Source code in src/data_handlers/random_variables.py

@model_validator(mode='after')
def validate_and_set_probabilities(self) -> CategoricalDistribution:
    """Validate probability values and set defaults if needed."""
    if self.probabilities is None:
        n_categories = len(self.categories)
        self.probabilities = np.ones(n_categories) / n_categories
        return self

    if len(self.categories) != len(self.probabilities):
        raise ValueError(
            f"Number of categories ({len(self.categories)}) must match "
            f"number of probabilities ({len(self.probabilities)})"
        )
    if not np.all(self.probabilities >= 0):
        raise ValueError("All probabilities must be non-negative")
    if not np.isclose(np.sum(self.probabilities), 1.0):
        raise ValueError("Probabilities must sum to 1")
    return self

NamedObject ¶

Bases: BaseModel

Base class for named objects with serialization support.

Attributes:

Name	Type	Description
`type`	`str`	Name of object class type (computed field)
`name`	`str`	Name of the object

Configuration

model_config (ConfigDict): Pydantic model configuration

type `property` ¶

type: str

Returns the name of the object type.

__init_subclass__ ¶

__init_subclass__(*, registry_category: str = None, **kwargs)

Register subclasses in appropriate registry category.

Source code in src/data_handlers/base.py

def __init_subclass__(cls, *, registry_category: str = None, **kwargs):
    """Register subclasses in appropriate registry category."""
    super().__init_subclass__(**kwargs)
    if registry_category:
        cls._registry_category = registry_category
    ObjectRegistry.register(cls._registry_category, cls)

NamedObjectHash ¶

Bases: BaseModel

Dictionary of named objects with type checking and conflict prevention.

Attributes:

Name	Type	Description
`objects`	`dict`	Dictionary of named objects

deserialize_objects `classmethod` ¶

deserialize_objects(data: Any) -> Any

Deserialize objects during validation.

Source code in src/data_handlers/base.py

@model_validator(mode='before')
@classmethod
def deserialize_objects(cls, data: Any) -> Any:
    """Deserialize objects during validation."""
    if not isinstance(data, dict):
        return data

    objects = data.get('objects', {})
    if isinstance(objects, dict):
        for name, obj_data in objects.items():
            if isinstance(obj_data, dict):
                type_name = obj_data.get('type')
                if type_name:
                    # Remove type as it's not part of the constructor
                    obj_data = obj_data.copy()
                    obj_data.pop('type')
                    obj_type = ObjectRegistry.get(cls._registry_category, type_name)
                    data['objects'][name] = obj_type(**obj_data)
    return data

get_object ¶

get_object(name: str) -> NamedObject

Get object by name.

Source code in src/data_handlers/base.py

def get_object(self, name: str) -> NamedObject:
    """Get object by name."""
    return self.objects[name]

get_object_names ¶

get_object_names() -> Iterable[str]

Get names of all objects.

Source code in src/data_handlers/base.py

def get_object_names(self) -> Iterable[str]:
    """Get names of all objects."""
    return self.objects.keys()

get_objects ¶

get_objects() -> Iterable[NamedObject]

Get all objects.

Source code in src/data_handlers/base.py

def get_objects(self) -> Iterable[NamedObject]:
    """Get all objects."""
    return self.objects.values()

register_object ¶

register_object(obj: NamedObject) -> Self

Register a named object. Checks for naming conflicts.

Parameters:

Name	Type	Description	Default
`obj`	`NamedObject`	Object to register	required

Raises:

Type	Description
`ValueError`	If object with same name exists

Source code in src/data_handlers/base.py

def register_object(self, obj: NamedObject) -> Self:
    """
    Register a named object. Checks for naming conflicts.

    Args:
        obj (NamedObject): Object to register

    Raises:
        ValueError: If object with same name exists
    """
    if obj.name in self.objects:
        raise ValueError(
            f"Naming conflict: An object named '{obj.name}' already exists."
            f"\n\tExisting: \n{self.get_object(obj.name).model_dump_json(indent=4)}"
            f"\n\tNew: \n{obj.model_dump_json(indent=4)}"
        )
    self.objects[obj.name] = obj
    return self

NamedObjectList ¶

Bases: BaseModel

List of named objects with type checking.

Attributes:

Name	Type	Description
`objects`	`list`	List of named objects

Example

obj_list = NamedObjectList()
obj_list.append(named_object)
obj_list.extend([obj1, obj2, obj3])

getitem ¶

__getitem__(idx: int) -> NamedObject

Get object by index.

Source code in src/data_handlers/base.py

def __getitem__(self, idx: int) -> NamedObject:
    """Get object by index."""
    return self.objects[idx]

iter ¶

__iter__() -> Iterable[NamedObject]

Iterate over objects in list.

Source code in src/data_handlers/base.py

def __iter__(self) -> Iterable[NamedObject]:
    """Iterate over objects in list."""
    return iter(self.objects)

len ¶

__len__() -> int

Return number of objects in list.

Source code in src/data_handlers/base.py

def __len__(self) -> int:
    """Return number of objects in list."""
    return len(self.objects)

append ¶

append(obj: NamedObject) -> Self

Append a single object to the list.

Parameters:

Name	Type	Description	Default
`obj`	`NamedObject`	Object to append	required

Returns:

Name	Type	Description
`Self`	`Self`	Returns self for chaining

Source code in src/data_handlers/base.py

def append(self, obj: NamedObject) -> Self:
    """
    Append a single object to the list.

    Args:
        obj (NamedObject): Object to append

    Returns:
        Self: Returns self for chaining
    """
    self.objects.append(obj)
    return self

extend ¶

extend(objects: Iterable[NamedObject]) -> Self

Extend list with multiple objects.

Parameters:

Name	Type	Description	Default
`objects`	`Iterable[NamedObject]`	Objects to add	required

Returns:

Name	Type	Description
`Self`	`Self`	Returns self for chaining

Source code in src/data_handlers/base.py

def extend(self, objects: Iterable[NamedObject]) -> Self:
    """
    Extend list with multiple objects.

    Args:
        objects (Iterable[NamedObject]): Objects to add

    Returns:
        Self: Returns self for chaining
    """
    self.objects.extend(objects)
    return self

from_iterable `classmethod` ¶

from_iterable(iterable: Iterable[NamedObject]) -> Self

Create instance from an iterable of named objects.

Parameters:

Name	Type	Description	Default
`iterable`	`Iterable[NamedObject]`	Objects to add to list	required

Returns:

Name	Type	Description
`Self`	`Self`	New instance containing the objects

Source code in src/data_handlers/base.py

@classmethod
def from_iterable(cls, iterable: Iterable[NamedObject]) -> Self:
    """
    Create instance from an iterable of named objects.

    Args:
        iterable (Iterable[NamedObject]): Objects to add to list

    Returns:
        Self: New instance containing the objects
    """
    return cls(objects=list(iterable))

get_object ¶

get_object(name: str) -> NamedObject

Get a registered object by name.

Parameters:

Name	Type	Description	Default
`name`	`str`	Name of the object to retrieve	required

Returns:

Name	Type	Description
`NamedObject`	`NamedObject`	The requested object

Raises:

Type	Description
`KeyError`	If no object exists with the given name

Example

obj_list = NamedObjectList()
obj_list.append(NamedObject("x"))
x = obj_list.get_object("x")

Source code in src/data_handlers/base.py

def get_object(self, name: str) -> NamedObject:
    """
    Get a registered object by name.

    Args:
        name (str): Name of the object to retrieve

    Returns:
        NamedObject: The requested object

    Raises:
        KeyError: If no object exists with the given name

    Example:
        ```python
        obj_list = NamedObjectList()
        obj_list.append(NamedObject("x"))
        x = obj_list.get_object("x")
        ```
    """
    for obj in self.objects:
        if obj.name == name:
            return obj
    raise KeyError(f"No object found with name '{name}'")

get_objects ¶

get_objects() -> Iterable[NamedObject]

Get all registered objects.

Returns:

Type	Description
`Iterable[NamedObject]`	Iterable[NamedObject]: Iterator over all stored objects

Example

obj_list = NamedObjectList()
obj_list.extend([NamedObject("x"), NamedObject("y")])
for obj in obj_list.get_objects():
    print(obj.name)

Source code in src/data_handlers/base.py

def get_objects(self) -> Iterable[NamedObject]:
    """
    Get all registered objects.

    Returns:
        Iterable[NamedObject]: Iterator over all stored objects

    Example:
        ```python
        obj_list = NamedObjectList()
        obj_list.extend([NamedObject("x"), NamedObject("y")])
        for obj in obj_list.get_objects():
            print(obj.name)
        ```
    """
    return iter(self.objects)

register_object ¶

register_object(obj: NamedObject) -> Self

Register a named object to the list with duplicate name checking.

Parameters:

Name	Type	Description	Default
`obj`	`NamedObject`	Object to register	required

Returns:

Name	Type	Description
`Self`	`Self`	Returns self for method chaining

Raises:

Type	Description
`ValueError`	If an object with the same name already exists

Example

obj_list = NamedObjectList()
obj_list.register_object(NamedObject("x"))

Source code in src/data_handlers/base.py

def register_object(self, obj: NamedObject) -> Self:
    """
    Register a named object to the list with duplicate name checking.

    Args:
        obj (NamedObject): Object to register

    Returns:
        Self: Returns self for method chaining

    Raises:
        ValueError: If an object with the same name already exists

    Example:
        ```python
        obj_list = NamedObjectList()
        obj_list.register_object(NamedObject("x"))
        ```
    """
    # Check for duplicates
    for existing_obj in self.objects:
        if existing_obj.name == obj.name:
            raise ValueError(
                f"Naming conflict: An object named '{obj.name}' already exists.\n"
                f"\tExisting: \n{existing_obj.model_dump_json(indent=4)}\n"
                f"\tNew: \n{obj.model_dump_json(indent=4)}"
            )
    self.objects.append(obj)
    return self

NamedValue ¶

NamedValue(name: str, value: T | None = None, **data)

Bases: NamedObject, Generic[T]

A named value container with type safety and state management.

NamedValue provides a type-safe way to store and manage values with built-in state tracking, serialization, and validation. Values can be frozen after initial setting to prevent accidental modification.

Type Parameters

T: The type of value to store, must be a SerializableValue

Attributes:

Name	Type	Description
`name`	`str`	Unique identifier for the value
`_stored_value`	`T \| NamedValueState`	The actual stored value or UNSET state
`_state`	`NamedValueState`	Current state of the value
`_type`	`type`	Runtime type information for validation

Properties

value (T): Access or modify the stored value

Example

# Create a named integer value
count = NamedValue[int]("item_count")
count.value = 42  # Sets and freezes the value
print(count.value)  # Outputs: 42
count.value = 50  # Raises ValueError - value is frozen
count.force_set_value(50)  # Allows value change

Source code in src/data_handlers/values.py

def __init__(self, name: str, value: T | None = None, **data):
    # print(f"Initializing NamedValue with class: {self.__class__}")
    # print(f"Class __args__: {getattr(self.__class__, '__args__', None)}")
    # print(f"Bases: {self.__class__.__bases__}")
    # for base in self.__class__.__bases__:
    #     print(f"Base __args__: {getattr(base, '__args__', None)}")

    data.pop('stored_value', None)
    data.pop('_stored_value', None)

    super().__init__(name=name, **data)
    self._type = self._extract_value_type()
    # print(f"Extracted type: {self._type}")
    object.__setattr__(self, '_stored_value', NamedValueState.UNSET)

    if value is not None:
        self.value = value

value `property` `writable` ¶

value: T

Get the stored value.

Returns:

Name	Type	Description
`T`	`T`	The currently stored value

Raises:

Type	Description
`ValueError`	If attempting to access before a value has been set

Note

This property provides read access to the stored value. Once set, the value is frozen and can only be changed using force_set_value().

setattr ¶

__setattr__(name: str, value: Any) -> None

Prevent direct modification of protected attributes.

Overrides attribute setting to prevent direct modification of internal state attributes. These attributes should only be modified through appropriate methods.

Parameters:

Name	Type	Description	Default
`name`	`str`	Name of the attribute to set	required
`value`	`Any`	Value to set	required

Raises:

Type	Description
`AttributeError`	If attempting to modify protected attributes directly

Example

value = NamedValue("example")
value._stored_value = 42  # Raises AttributeError

Source code in src/data_handlers/values.py

def __setattr__(self, name: str, value: Any) -> None:
    """
    Prevent direct modification of protected attributes.

    Overrides attribute setting to prevent direct modification of internal
    state attributes. These attributes should only be modified through
    appropriate methods.

    Args:
        name (str): Name of the attribute to set
        value (Any): Value to set

    Raises:
        AttributeError: If attempting to modify protected attributes directly

    Example:
        ```python
        value = NamedValue("example")
        value._stored_value = 42  # Raises AttributeError
        ```
    """
    if name in ('_stored_value', '_state'):
        raise AttributeError(f"Cannot modify {name} directly. Use appropriate methods instead.")
    super().__setattr__(name, value)

append_to_value_list ¶

append_to_value_list(l: NamedValueList) -> Self

Appends this value instance to a NamedValueList.

Convenience method for adding this value to a list while enabling method chaining.

Parameters:

Name	Type	Description	Default
`l`	`NamedValueList`	The list to append this value to	required

Returns:

Name	Type	Description
`Self`	`Self`	This instance for method chaining

Example

value_list = NamedValueList()
value = NamedValue("example", 42)
value.append_to_value_list(value_list).force_set_value(43)

Source code in src/data_handlers/values.py

def append_to_value_list(self, l: NamedValueList) -> Self:
    """
    Appends this value instance to a NamedValueList.

    Convenience method for adding this value to a list while enabling
    method chaining.

    Args:
        l (NamedValueList): The list to append this value to

    Returns:
        Self: This instance for method chaining

    Example:
        ```python
        value_list = NamedValueList()
        value = NamedValue("example", 42)
        value.append_to_value_list(value_list).force_set_value(43)
        ```
    """
    l.append(self)
    return self

force_set_value ¶

force_set_value(new_value: T) -> None

Force set the value regardless of its current state.

This method bypasses the normal freezing mechanism and allows changing an already-set value.

Parameters:

Name	Type	Description	Default
`new_value`	`T`	New value to store	required

Raises:

Type	Description
`TypeError`	If value doesn't match the expected type T

Source code in src/data_handlers/values.py

def force_set_value(self, new_value: T) -> None:
    """
    Force set the value regardless of its current state.

    This method bypasses the normal freezing mechanism and allows
    changing an already-set value.

    Args:
        new_value (T): New value to store

    Raises:
        TypeError: If value doesn't match the expected type T
    """
    object.__setattr__(self, '_stored_value', NamedValueState.UNSET)
    object.__setattr__(self, '_state', NamedValueState.UNSET)

    # if new_value == 'not a series':
    #     breakpoint()
    self.value = new_value

model_dump ¶

model_dump(**kwargs) -> dict[str, Any]

Custom serialization to include value state and stored value.

Extends the parent class serialization to include the value's state and stored value (if set) in the serialized data.

Parameters:

Name	Type	Description	Default
`**kwargs`		Additional arguments passed to parent serialization	`{}`

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Dictionary containing serialized state

Example

value = NamedValue("example", 42)
data = value.model_dump()
print(data)  # Contains 'state' and 'stored_value'

Source code in src/data_handlers/values.py

def model_dump(self, **kwargs) -> dict[str, Any]:
    """
    Custom serialization to include value state and stored value.

    Extends the parent class serialization to include the value's state
    and stored value (if set) in the serialized data.

    Args:
        **kwargs: Additional arguments passed to parent serialization

    Returns:
        dict[str, Any]: Dictionary containing serialized state

    Example:
        ```python
        value = NamedValue("example", 42)
        data = value.model_dump()
        print(data)  # Contains 'state' and 'stored_value'
        ```
    """
    data = super().model_dump(**kwargs)
    data['state'] = self._state
    if self._state == NamedValueState.SET:
        data['stored_value'] = self._stored_value
    return data

model_dump_json ¶

model_dump_json(**kwargs) -> str

Custom JSON serialization of the named value.

Serializes the named value instance to a JSON string, including all state information and stored value.

Parameters:

Name	Type	Description	Default
`**kwargs`		JSON serialization options like indent, ensure_ascii, etc.	`{}`

Returns:

Name	Type	Description
`str`	`str`	JSON string representation

Example

value = NamedValue("example", 42)
json_str = value.model_dump_json(indent=2)
print(json_str)  # Pretty-printed JSON

Source code in src/data_handlers/values.py

def model_dump_json(self, **kwargs) -> str:
    """
    Custom JSON serialization of the named value.

    Serializes the named value instance to a JSON string, including
    all state information and stored value.

    Args:
        **kwargs: JSON serialization options like indent, ensure_ascii, etc.

    Returns:
        str: JSON string representation

    Example:
        ```python
        value = NamedValue("example", 42)
        json_str = value.model_dump_json(indent=2)
        print(json_str)  # Pretty-printed JSON
        ```
    """
    # Separate JSON-specific kwargs from model_dump kwargs
    json_kwargs = {k: v for k, v in kwargs.items() if k in {'indent', 'ensure_ascii', 'separators'}}
    dump_kwargs = {k: v for k, v in kwargs.items() if k not in json_kwargs}

    # Get model data with stored value
    data = self.model_dump(**dump_kwargs)
    return json.dumps(data, **json_kwargs)

model_validate `classmethod` ¶

model_validate(data: Any) -> NamedValue

Custom deserialization to restore value state and stored value.

Reconstructs a NamedValue instance from serialized data, properly restoring both the value state and any stored value.

Parameters:

Name	Type	Description	Default
`data`	`Any`	Serialized data to deserialize	required

Returns:

Name	Type	Description
`NamedValue`	`NamedValue`	New instance with restored state

Example

data = {'name': 'example', 'state': 'set', 'stored_value': 42}
value = NamedValue.model_validate(data)
print(value.value)  # Outputs: 42

Source code in src/data_handlers/values.py

@classmethod
def model_validate(cls, data: Any) -> NamedValue:
    """
    Custom deserialization to restore value state and stored value.

    Reconstructs a NamedValue instance from serialized data, properly
    restoring both the value state and any stored value.

    Args:
        data (Any): Serialized data to deserialize

    Returns:
        NamedValue: New instance with restored state

    Example:
        ```python
        data = {'name': 'example', 'state': 'set', 'stored_value': 42}
        value = NamedValue.model_validate(data)
        print(value.value)  # Outputs: 42
        ```
    """
    if not isinstance(data, dict):
        return super().model_validate(data)

    data_copy = data.copy()
    state = NamedValueState(data_copy.pop('state', NamedValueState.UNSET))
    stored_value = data_copy.pop('stored_value', None)

    instance = super().model_validate(data_copy)

    # Only set the value if state was SET
    if state == NamedValueState.SET and stored_value is not None:
        instance.force_set_value(stored_value)

    return instance

model_validate_json `classmethod` ¶

model_validate_json(json_data: str, **kwargs) -> NamedValue

Custom JSON deserialization to NamedValue instance.

Reconstructs a NamedValue instance from a JSON string representation, restoring all state and stored value information.

Parameters:

Name	Type	Description	Default
`json_data`	`str`	JSON string to deserialize	required
`**kwargs`		Additional validation options	`{}`

Returns:

Name	Type	Description
`NamedValue`	`NamedValue`	New instance with restored state

Example

json_str = '{"name": "example", "state": "set", "stored_value": 42}'
value = NamedValue.model_validate_json(json_str)
print(value.value)  # Outputs: 42

Source code in src/data_handlers/values.py

@classmethod
def model_validate_json(cls, json_data: str, **kwargs) -> NamedValue:
    """
    Custom JSON deserialization to NamedValue instance.

    Reconstructs a NamedValue instance from a JSON string representation,
    restoring all state and stored value information.

    Args:
        json_data (str): JSON string to deserialize
        **kwargs: Additional validation options

    Returns:
        NamedValue: New instance with restored state

    Example:
        ```python
        json_str = '{"name": "example", "state": "set", "stored_value": 42}'
        value = NamedValue.model_validate_json(json_str)
        print(value.value)  # Outputs: 42
        ```
    """
    data = json.loads(json_data)
    return cls.model_validate(data, **kwargs)

register_to_value_hash ¶

register_to_value_hash(h: NamedValueHash) -> Self

Registers this value instance in a NamedValueHash.

Registers this value in the provided hash container. If the hash contains value overrides, this value's current value may be overridden during registration.

Parameters:

Name	Type	Description	Default
`h`	`NamedValueHash`	The hash to register this value in	required

Returns:

Name	Type	Description
`Self`	`Self`	This instance for method chaining

Example

value_hash = NamedValueHash()
value = NamedValue("example", 42)
value.register_to_value_hash(value_hash).force_set_value(43)

Source code in src/data_handlers/values.py

def register_to_value_hash(self, h: NamedValueHash) -> Self:
    """
    Registers this value instance in a NamedValueHash.

    Registers this value in the provided hash container. If the hash contains
    value overrides, this value's current value may be overridden during
    registration.

    Args:
        h (NamedValueHash): The hash to register this value in

    Returns:
        Self: This instance for method chaining

    Example:
        ```python
        value_hash = NamedValueHash()
        value = NamedValue("example", 42)
        value.register_to_value_hash(value_hash).force_set_value(43)
        ```
    """
    h.register_value(self)
    return self

NamedValueHash ¶

Bases: NamedObjectHash

A type-safe dictionary for storing and managing NamedValue objects.

NamedValueHash provides a dictionary-like interface for managing a collection of NamedValue instances, using their names as keys. It ensures type safety and provides convenient methods for accessing and managing the stored values.

The hash maintains unique naming across all stored values and supports serialization/deserialization of the entire collection.

Attributes:

Name	Type	Description
`_registry_category`	`str`	Category identifier for object registration
`model_config`	`ConfigDict`	Pydantic configuration for model behavior

Example

value_hash = NamedValueHash()
value_hash.register_value(NamedValue("count", 42))
print(value_hash.get_raw_value("count"))  # Outputs: 42

get_raw_value ¶

get_raw_value(name: str) -> Any

Get the underlying value by name.

Parameters:

Name	Type	Description	Default
`name`	`str`	Name of the value to retrieve	required

Returns:

Name	Type	Description
`Any`	`Any`	The actual value stored in the named value

Raises:

Type	Description
`KeyError`	If no value exists with the given name
`ValueError`	If the value hasn't been set yet

Example

hash = NamedValueHash()
hash.register_value(NamedValue("price", 10.99))
print(hash.get_raw_value("price"))  # Outputs: 10.99

Source code in src/data_handlers/values.py

def get_raw_value(self, name: str) -> Any:
    """
    Get the underlying value by name.

    Args:
        name (str): Name of the value to retrieve

    Returns:
        Any: The actual value stored in the named value

    Raises:
        KeyError: If no value exists with the given name
        ValueError: If the value hasn't been set yet

    Example:
        ```python
        hash = NamedValueHash()
        hash.register_value(NamedValue("price", 10.99))
        print(hash.get_raw_value("price"))  # Outputs: 10.99
        ```
    """
    return self.get_value(name).value

get_raw_values ¶

get_raw_values() -> Iterable[Any]

Get the underlying values of all named values.

Returns:

Type	Description
`Iterable[Any]`	Iterable[Any]: Iterator over the actual values stored in each NamedValue

Example

hash = NamedValueHash()
hash.register_value(NamedValue("x", 1))
hash.register_value(NamedValue("y", 2))
print(list(hash.get_raw_values()))  # Outputs: [1, 2]

Source code in src/data_handlers/values.py

def get_raw_values(self) -> Iterable[Any]:
    """
    Get the underlying values of all named values.

    Returns:
        Iterable[Any]: Iterator over the actual values stored in each NamedValue

    Example:
        ```python
        hash = NamedValueHash()
        hash.register_value(NamedValue("x", 1))
        hash.register_value(NamedValue("y", 2))
        print(list(hash.get_raw_values()))  # Outputs: [1, 2]
        ```
    """
    return (val.value for val in self.get_values())

get_value ¶

get_value(name: str) -> NamedValue

Retrieve a named value by its name.

Parameters:

Name	Type	Description	Default
`name`	`str`	Name of the value to retrieve	required

Returns:

Name	Type	Description
`NamedValue`	`NamedValue`	The requested named value

Raises:

Type	Description
`KeyError`	If no value exists with the given name

Example

hash = NamedValueHash()
hash.register_value(NamedValue("price", 10.99))
price = hash.get_value("price")
print(price.value)  # Outputs: 10.99

Source code in src/data_handlers/values.py

def get_value(self, name: str) -> NamedValue:
    """
    Retrieve a named value by its name.

    Args:
        name (str): Name of the value to retrieve

    Returns:
        NamedValue: The requested named value

    Raises:
        KeyError: If no value exists with the given name

    Example:
        ```python
        hash = NamedValueHash()
        hash.register_value(NamedValue("price", 10.99))
        price = hash.get_value("price")
        print(price.value)  # Outputs: 10.99
        ```
    """
    return self.get_object(name)

get_value_by_type ¶

get_value_by_type(value_type: Type) -> Iterable[NamedValue]

Get all values of a specific type.

Parameters:

Name	Type	Description	Default
`value_type`	`Type`	Type to filter values by	required

Returns:

Type	Description
`Iterable[NamedValue]`	Iterable[NamedValue]: Values matching the specified type

Example

hash = NamedValueHash()
hash.register_value(NamedValue("x", 1))
hash.register_value(NamedValue("name", "test"))
integers = list(hash.get_value_by_type(int))

Source code in src/data_handlers/values.py

def get_value_by_type(self, value_type: Type) -> Iterable[NamedValue]:
    """
    Get all values of a specific type.

    Args:
        value_type (Type): Type to filter values by

    Returns:
        Iterable[NamedValue]: Values matching the specified type

    Example:
        ```python
        hash = NamedValueHash()
        hash.register_value(NamedValue("x", 1))
        hash.register_value(NamedValue("name", "test"))
        integers = list(hash.get_value_by_type(int))
        ```
    """
    return [val for val in self.get_values() if isinstance(val, value_type)]

get_value_names ¶

get_value_names() -> Iterable[str]

Get names of all registered values.

Returns:

Type	Description
`Iterable[str]`	Iterable[str]: An iterator over all value names

Example

hash = NamedValueHash()
hash.register_value(NamedValue("x", 1))
hash.register_value(NamedValue("y", 2))
print(list(hash.get_value_names()))  # Outputs: ['x', 'y']

Source code in src/data_handlers/values.py

def get_value_names(self) -> Iterable[str]:
    """
    Get names of all registered values.

    Returns:
        Iterable[str]: An iterator over all value names

    Example:
        ```python
        hash = NamedValueHash()
        hash.register_value(NamedValue("x", 1))
        hash.register_value(NamedValue("y", 2))
        print(list(hash.get_value_names()))  # Outputs: ['x', 'y']
        ```
    """
    return self.get_object_names()

get_values ¶

get_values() -> Iterable[NamedValue]

Get all registered named values.

Returns:

Type	Description
`Iterable[NamedValue]`	Iterable[NamedValue]: An iterator over all stored named values

Example

hash = NamedValueHash()
hash.register_value(NamedValue("x", 1))
hash.register_value(NamedValue("y", 2))
for value in hash.get_values():
    print(f"{value.name}: {value.value}")

Source code in src/data_handlers/values.py

def get_values(self) -> Iterable[NamedValue]:
    """
    Get all registered named values.

    Returns:
        Iterable[NamedValue]: An iterator over all stored named values

    Example:
        ```python
        hash = NamedValueHash()
        hash.register_value(NamedValue("x", 1))
        hash.register_value(NamedValue("y", 2))
        for value in hash.get_values():
            print(f"{value.name}: {value.value}")
        ```
    """
    return self.get_objects()

model_dump ¶

model_dump(**kwargs) -> dict[str, Any]

Custom serialization to preserve stored values and their states.

Creates a dictionary representation of the hash that includes full serialization of all contained NamedValue objects, preserving their values and states.

Parameters:

Name	Type	Description	Default
`**kwargs`		Additional serialization options passed to all nested objects	`{}`

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Dictionary containing the complete hash state

Example

hash = NamedValueHash()
hash.register_value(NamedValue("x", 1))
data = hash.model_dump()
print(data['objects']['x']['stored_value'])  # Outputs: 1

Source code in src/data_handlers/values.py

def model_dump(self, **kwargs) -> dict[str, Any]:
    """
    Custom serialization to preserve stored values and their states.

    Creates a dictionary representation of the hash that includes full
    serialization of all contained NamedValue objects, preserving their
    values and states.

    Args:
        **kwargs: Additional serialization options passed to all nested objects

    Returns:
        dict[str, Any]: Dictionary containing the complete hash state

    Example:
        ```python
        hash = NamedValueHash()
        hash.register_value(NamedValue("x", 1))
        data = hash.model_dump()
        print(data['objects']['x']['stored_value'])  # Outputs: 1
        ```
    """
    data = super().model_dump(**kwargs)
    # Ensure each object's stored value is included
    if 'objects' in data:
        for name, obj in self.objects.items():
            if isinstance(obj, NamedValue):
                # Get the full dump including stored value
                obj_data = obj.model_dump(**kwargs)
                data['objects'][name] = obj_data
    return data

model_dump_json ¶

model_dump_json(**kwargs) -> str

Custom JSON serialization of the entire hash.

Serializes the NamedValueHash instance and all contained NamedValue objects to a JSON string representation. Handles both the hash structure and the nested value serialization.

Parameters:

Name	Type	Description	Default
`**kwargs`		JSON serialization options such as: - indent: Number of spaces for pretty printing - ensure_ascii: Escape non-ASCII characters - separators: Tuple of (item_sep, key_sep) for custom formatting	`{}`

Returns:

Name	Type	Description
`str`	`str`	JSON string representation of the hash

Example

hash = NamedValueHash()
hash.register_value(NamedValue("x", 1))
json_str = hash.model_dump_json(indent=2)
print(json_str)  # Pretty-printed JSON with nested values

Source code in src/data_handlers/values.py

def model_dump_json(self, **kwargs) -> str:
    """
    Custom JSON serialization of the entire hash.

    Serializes the NamedValueHash instance and all contained NamedValue
    objects to a JSON string representation. Handles both the hash structure
    and the nested value serialization.

    Args:
        **kwargs: JSON serialization options such as:
            - indent: Number of spaces for pretty printing
            - ensure_ascii: Escape non-ASCII characters
            - separators: Tuple of (item_sep, key_sep) for custom formatting

    Returns:
        str: JSON string representation of the hash

    Example:
        ```python
        hash = NamedValueHash()
        hash.register_value(NamedValue("x", 1))
        json_str = hash.model_dump_json(indent=2)
        print(json_str)  # Pretty-printed JSON with nested values
        ```
    """
    # Separate JSON-specific kwargs from model_dump kwargs
    json_kwargs = {k: v for k, v in kwargs.items() if k in {'indent', 'ensure_ascii', 'separators'}}
    dump_kwargs = {k: v for k, v in kwargs.items() if k not in json_kwargs}

    # Get model data
    data = self.model_dump(**dump_kwargs)
    # Serialize to JSON
    return json.dumps(data, **json_kwargs)

model_validate `classmethod` ¶

model_validate(data: Any) -> NamedValueHash

Custom validation to restore hash state from serialized data.

Reconstructs a NamedValueHash instance from serialized data, including all contained NamedValue objects with their values and states.

Parameters:

Name	Type	Description	Default
`data`	`Any`	Serialized data to deserialize. Should be a dictionary containing an 'objects' key with serialized NamedValue instances	required

Returns:

Name	Type	Description
`NamedValueHash`	`NamedValueHash`	New instance with all values restored

Example

data = {
    'objects': {
        'x': {'name': 'x', 'type': 'NamedValue', 'stored_value': 1}
    }
}
hash = NamedValueHash.model_validate(data)
print(hash.get_raw_value('x'))  # Outputs: 1

Source code in src/data_handlers/values.py

@classmethod
def model_validate(cls, data: Any) -> NamedValueHash:
    """
    Custom validation to restore hash state from serialized data.

    Reconstructs a NamedValueHash instance from serialized data, including
    all contained NamedValue objects with their values and states.

    Args:
        data (Any): Serialized data to deserialize. Should be a dictionary
            containing an 'objects' key with serialized NamedValue instances

    Returns:
        NamedValueHash: New instance with all values restored

    Example:
        ```python
        data = {
            'objects': {
                'x': {'name': 'x', 'type': 'NamedValue', 'stored_value': 1}
            }
        }
        hash = NamedValueHash.model_validate(data)
        print(hash.get_raw_value('x'))  # Outputs: 1
        ```
    """
    if not isinstance(data, dict):
        return super().model_validate(data)

    instance = cls()

    # Process each object in the data
    for name, obj_data in data.get('objects', {}).items():
        if isinstance(obj_data, dict):
            obj_type = obj_data.get('type')
            if obj_type:
                # Get the appropriate class from registry
                value_class = ObjectRegistry.get(cls._registry_category, obj_type)
                # Create and validate the object with its stored value
                value_obj = value_class.model_validate(obj_data)
                instance.register_value(value_obj)

    return instance

model_validate_json `classmethod` ¶

model_validate_json(json_data: str, **kwargs) -> NamedValueHash

Custom JSON deserialization to NamedValueHash instance.

Reconstructs a NamedValueHash instance from a JSON string representation, including all contained NamedValue objects with their complete state.

Parameters:

Name	Type	Description	Default
`json_data`	`str`	JSON string containing serialized hash data	required
`**kwargs`		Additional validation options for nested objects	`{}`

Returns:

Name	Type	Description
`NamedValueHash`	`NamedValueHash`	New instance with all values restored

Example

json_str = '''
{
    "objects": {
        "x": {"name": "x", "type": "NamedValue", "stored_value": 1}
    }
}
'''
hash = NamedValueHash.model_validate_json(json_str)
print(hash.get_raw_value('x'))  # Outputs: 1

Source code in src/data_handlers/values.py

@classmethod
def model_validate_json(cls, json_data: str, **kwargs) -> NamedValueHash:
    """
    Custom JSON deserialization to NamedValueHash instance.

    Reconstructs a NamedValueHash instance from a JSON string representation,
    including all contained NamedValue objects with their complete state.

    Args:
        json_data (str): JSON string containing serialized hash data
        **kwargs: Additional validation options for nested objects

    Returns:
        NamedValueHash: New instance with all values restored

    Example:
        ```python
        json_str = '''
        {
            "objects": {
                "x": {"name": "x", "type": "NamedValue", "stored_value": 1}
            }
        }
        '''
        hash = NamedValueHash.model_validate_json(json_str)
        print(hash.get_raw_value('x'))  # Outputs: 1
        ```
    """
    data = json.loads(json_data)
    return cls.model_validate(data, **kwargs)

register_value ¶

register_value(value: NamedValue) -> Self

Register a named value in the hash.

Parameters:

Name	Type	Description	Default
`value`	`NamedValue`	The value to register	required

Returns:

Name	Type	Description
`Self`	`Self`	Returns self for method chaining

Raises:

Type	Description
`ValueError`	If a value with the same name already exists

Example

hash = NamedValueHash()
value = NamedValue("price", 10.99)
hash.register_value(value).register_value(NamedValue("qty", 5))

Source code in src/data_handlers/values.py

def register_value(self, value: NamedValue) -> Self:
    """
    Register a named value in the hash.

    Args:
        value (NamedValue): The value to register

    Returns:
        Self: Returns self for method chaining

    Raises:
        ValueError: If a value with the same name already exists

    Example:
        ```python
        hash = NamedValueHash()
        value = NamedValue("price", 10.99)
        hash.register_value(value).register_value(NamedValue("qty", 5))
        ```
    """
    return self.register_object(value)

set_raw_value ¶

set_raw_value(name: str, value: Any) -> None

Set the underlying value for a named value.

Parameters:

Name	Type	Description	Default
`name`	`str`	Name of the value to update	required
`value`	`Any`	New value to set	required

Raises:

Type	Description
`KeyError`	If no value exists with the given name
`TypeError`	If value type doesn't match the expected type

Example

hash = NamedValueHash()
hash.register_value(NamedValue("price", 10.99))
hash.set_raw_value("price", 11.99)

Source code in src/data_handlers/values.py

def set_raw_value(self, name: str, value: Any) -> None:
    """
    Set the underlying value for a named value.

    Args:
        name (str): Name of the value to update
        value (Any): New value to set

    Raises:
        KeyError: If no value exists with the given name
        TypeError: If value type doesn't match the expected type

    Example:
        ```python
        hash = NamedValueHash()
        hash.register_value(NamedValue("price", 10.99))
        hash.set_raw_value("price", 11.99)
        ```
    """
    self.get_value(name).value = value

NamedValueList ¶

Bases: NamedObjectList

An ordered list container for managing NamedValue objects.

NamedValueList maintains an ordered collection of NamedValue objects while providing type safety and convenient access methods. It preserves insertion order while also allowing access by name.

Attributes:

Name	Type	Description
`_registry_category`	`str`	Category identifier for object registration
`objects`	`List[SerializeAsAny[InstanceOf[NamedValue]]]`	The list of stored values

Example

value_list = NamedValueList()
value_list.append(NamedValue("first", 1))
value_list.append(NamedValue("second", 2))
print([v.value for v in value_list])  # Outputs: [1, 2]

getitem ¶

__getitem__(idx: int) -> NamedValue

Get a named value by its index in the list.

Parameters:

Name	Type	Description	Default
`idx`	`int`	Index of the named value to retrieve	required

Returns:

Name	Type	Description
`NamedValue`	`NamedValue`	The named value at the specified index

Raises:

Type	Description
`IndexError`	If the index is out of range

Example

value_list = NamedValueList()
value_list.append(NamedValue(name="price", value=10.5))
first_value = value_list[0] # Get first named value

Source code in src/data_handlers/values.py

def __getitem__(self, idx: int) -> NamedValue:
    """Get a named value by its index in the list.

    Args:
        idx (int): Index of the named value to retrieve

    Returns:
        NamedValue: The named value at the specified index

    Raises:
        IndexError: If the index is out of range

    Example:
        ```python
        value_list = NamedValueList()
        value_list.append(NamedValue(name="price", value=10.5))
        first_value = value_list[0] # Get first named value
        ```
    """
    return super().__getitem__(idx)

append ¶

append(value: NamedValue) -> Self

Append a named value to the end of the list.

Parameters:

Name	Type	Description	Default
`value`	`NamedValue`	Named value to append	required

Returns:

Name	Type	Description
`Self`	`Self`	The list instance for method chaining

Example

value_list = NamedValueList()
value_list.append(NamedValue("x", 1)).append(NamedValue("y", 2))

Source code in src/data_handlers/values.py

def append(self, value: NamedValue) -> Self:
    """
    Append a named value to the end of the list.

    Args:
        value (NamedValue): Named value to append

    Returns:
        Self: The list instance for method chaining

    Example:
        ```python
        value_list = NamedValueList()
        value_list.append(NamedValue("x", 1)).append(NamedValue("y", 2))
        ```
    """
    return super().append(value)

extend ¶

extend(values: Iterable[NamedValue]) -> Self

Extend the list with multiple named values.

Parameters:

Name	Type	Description	Default
`values`	`Iterable[NamedValue]`	Collection of named values to add	required

Returns:

Name	Type	Description
`Self`	`Self`	The list instance for method chaining

Example

value_list = NamedValueList()
new_values = [NamedValue("x", 1), NamedValue("y", 2)]
value_list.extend(new_values)

Source code in src/data_handlers/values.py

def extend(self, values: Iterable[NamedValue]) -> Self:
    """
    Extend the list with multiple named values.

    Args:
        values (Iterable[NamedValue]): Collection of named values to add

    Returns:
        Self: The list instance for method chaining

    Example:
        ```python
        value_list = NamedValueList()
        new_values = [NamedValue("x", 1), NamedValue("y", 2)]
        value_list.extend(new_values)
        ```
    """
    return super().extend(values)

get_value ¶

get_value(name: str) -> NamedValue

Get a registered named value by name.

Parameters:

Name	Type	Description	Default
`name`	`str`	Name of the value to retrieve	required

Returns:

Name	Type	Description
`NamedValue`	`NamedValue`	The requested named value

Raises:

Type	Description
`KeyError`	If no value exists with the given name

Example

value_list = NamedValueList()
value_list.append(NamedValue("x", 1))
x = value_list.get_value("x")
print(x.value)  # Outputs: 1

Source code in src/data_handlers/values.py

def get_value(self, name: str) -> NamedValue:
    """
    Get a registered named value by name.

    Args:
        name (str): Name of the value to retrieve

    Returns:
        NamedValue: The requested named value

    Raises:
        KeyError: If no value exists with the given name

    Example:
        ```python
        value_list = NamedValueList()
        value_list.append(NamedValue("x", 1))
        x = value_list.get_value("x")
        print(x.value)  # Outputs: 1
        ```
    """
    return self.get_object(name)

get_value_by_type ¶

get_value_by_type(value_type: Type) -> Iterable[NamedValue]

Get all values whose stored value is of a specific type.

Parameters:

Name	Type	Description	Default
`value_type`	`Type`	Type to filter values by	required

Returns:

Type	Description
`Iterable[NamedValue]`	Iterable[NamedValue]: Values whose stored value matches the specified type

Example

value_list = NamedValueList()
value_list.append(NamedValue("x", 1))
value_list.append(NamedValue("name", "test"))
integers = list(value_list.get_value_by_type(int))

Source code in src/data_handlers/values.py

def get_value_by_type(self, value_type: Type) -> Iterable[NamedValue]:
    """
    Get all values whose stored value is of a specific type.

    Args:
        value_type (Type): Type to filter values by

    Returns:
        Iterable[NamedValue]: Values whose stored value matches the specified type

    Example:
        ```python
        value_list = NamedValueList()
        value_list.append(NamedValue("x", 1))
        value_list.append(NamedValue("name", "test"))
        integers = list(value_list.get_value_by_type(int))
        ```
    """
    for value in self.get_values():
        try:
            if isinstance(value.value, value_type):
                yield value
        except ValueError:
            # Skip unset values
            continue

get_values ¶

get_values() -> Iterable[NamedValue]

Get all registered named values.

Returns:

Type	Description
`Iterable[NamedValue]`	Iterable[NamedValue]: Iterator over all stored named values

Example

value_list = NamedValueList()
value_list.extend([NamedValue("x", 1), NamedValue("y", 2)])
for value in value_list.get_values():
    print(f"{value.name}: {value.value}")

Source code in src/data_handlers/values.py

def get_values(self) -> Iterable[NamedValue]:
    """
    Get all registered named values.

    Returns:
        Iterable[NamedValue]: Iterator over all stored named values

    Example:
        ```python
        value_list = NamedValueList()
        value_list.extend([NamedValue("x", 1), NamedValue("y", 2)])
        for value in value_list.get_values():
            print(f"{value.name}: {value.value}")
        ```
    """
    return self.get_objects()

model_dump ¶

model_dump(**kwargs) -> dict[str, Any]

Custom serialization to preserve stored values.

Extends the parent class serialization to ensure proper serialization of all stored named values and their states.

Parameters:

Name	Type	Description	Default
`**kwargs`		Additional serialization options	`{}`

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Dictionary containing serialized state

Source code in src/data_handlers/values.py

def model_dump(self, **kwargs) -> dict[str, Any]:
    """
    Custom serialization to preserve stored values.

    Extends the parent class serialization to ensure proper serialization
    of all stored named values and their states.

    Args:
        **kwargs: Additional serialization options

    Returns:
        dict[str, Any]: Dictionary containing serialized state
    """
    data = super().model_dump(**kwargs)
    if 'objects' in data:
        # Ensure each object's stored value is included
        data['objects'] = [
            obj.model_dump(**kwargs) if isinstance(obj, NamedValue) else obj
            for obj in self.objects
        ]
    return data

model_dump_json ¶

model_dump_json(**kwargs) -> str

Custom JSON serialization of the value list.

Serializes the NamedValueList instance and all contained NamedValue objects to a JSON string representation. Preserves the order of values and their complete state.

Parameters:

Name	Type	Description	Default
`**kwargs`		JSON serialization options such as: - indent: Number of spaces for pretty printing - ensure_ascii: Escape non-ASCII characters - separators: Tuple of (item_sep, key_sep) for custom formatting	`{}`

Returns:

Name	Type	Description
`str`	`str`	JSON string representation of the list

Example

value_list = NamedValueList()
value_list.append(NamedValue("x", 1))
value_list.append(NamedValue("y", 2))
json_str = value_list.model_dump_json(indent=2)
print(json_str)  # Pretty-printed JSON with ordered values

Source code in src/data_handlers/values.py

def model_dump_json(self, **kwargs) -> str:
    """
    Custom JSON serialization of the value list.

    Serializes the NamedValueList instance and all contained NamedValue
    objects to a JSON string representation. Preserves the order of values
    and their complete state.

    Args:
        **kwargs: JSON serialization options such as:
            - indent: Number of spaces for pretty printing
            - ensure_ascii: Escape non-ASCII characters
            - separators: Tuple of (item_sep, key_sep) for custom formatting

    Returns:
        str: JSON string representation of the list

    Example:
        ```python
        value_list = NamedValueList()
        value_list.append(NamedValue("x", 1))
        value_list.append(NamedValue("y", 2))
        json_str = value_list.model_dump_json(indent=2)
        print(json_str)  # Pretty-printed JSON with ordered values
        ```
    """
    # Separate JSON-specific kwargs from model_dump kwargs
    json_kwargs = {k: v for k, v in kwargs.items() if k in {'indent', 'ensure_ascii', 'separators'}}
    dump_kwargs = {k: v for k, v in kwargs.items() if k not in json_kwargs}

    # Get model data
    data = self.model_dump(**dump_kwargs)
    # Serialize to JSON
    return json.dumps(data, **json_kwargs)

model_validate `classmethod` ¶

model_validate(data: Any) -> NamedValueList

Custom validation to restore stored values.

Reconstructs a NamedValueList instance from serialized data, properly restoring all contained named values and their states.

Parameters:

Name	Type	Description	Default
`data`	`Any`	Serialized data to deserialize	required

Returns:

Name	Type	Description
`NamedValueList`	`NamedValueList`	New instance with restored values

Source code in src/data_handlers/values.py

@classmethod
def model_validate(cls, data: Any) -> NamedValueList:
    """
    Custom validation to restore stored values.

    Reconstructs a NamedValueList instance from serialized data,
    properly restoring all contained named values and their states.

    Args:
        data (Any): Serialized data to deserialize

    Returns:
        NamedValueList: New instance with restored values
    """
    if not isinstance(data, dict):
        return super().model_validate(data)

    instance = cls()

    # Process each object in the data
    for obj_data in data.get('objects', []):
        if isinstance(obj_data, dict):
            obj_type = obj_data.get('type')
            if obj_type:
                # Get the appropriate class from registry
                value_class = ObjectRegistry.get(cls._registry_category, obj_type)
                # Create and validate the object
                value_obj = value_class.model_validate(obj_data)
                instance.append(value_obj)

    return instance

model_validate_json `classmethod` ¶

model_validate_json(json_data: str, **kwargs) -> NamedValueList

Custom JSON deserialization to NamedValueList instance.

Reconstructs a NamedValueList instance from a JSON string representation, preserving the order of values and restoring their complete state.

Parameters:

Name	Type	Description	Default
`json_data`	`str`	JSON string containing serialized list data	required
`**kwargs`		Additional validation options for nested objects	`{}`

Returns:

Name	Type	Description
`NamedValueList`	`NamedValueList`	New instance with all values restored in order

Example

json_str = '''
{
    "objects": [
        {"name": "x", "type": "NamedValue", "stored_value": 1},
        {"name": "y", "type": "NamedValue", "stored_value": 2}
    ]
}
'''
value_list = NamedValueList.model_validate_json(json_str)
print([v.value for v in value_list])  # Outputs: [1, 2]

Source code in src/data_handlers/values.py

@classmethod
def model_validate_json(cls, json_data: str, **kwargs) -> NamedValueList:
    """
    Custom JSON deserialization to NamedValueList instance.

    Reconstructs a NamedValueList instance from a JSON string representation,
    preserving the order of values and restoring their complete state.

    Args:
        json_data (str): JSON string containing serialized list data
        **kwargs: Additional validation options for nested objects

    Returns:
        NamedValueList: New instance with all values restored in order

    Example:
        ```python
        json_str = '''
        {
            "objects": [
                {"name": "x", "type": "NamedValue", "stored_value": 1},
                {"name": "y", "type": "NamedValue", "stored_value": 2}
            ]
        }
        '''
        value_list = NamedValueList.model_validate_json(json_str)
        print([v.value for v in value_list])  # Outputs: [1, 2]
        ```
    """
    data = json.loads(json_data)
    return cls.model_validate(data, **kwargs)

register_value ¶

register_value(value: NamedValue) -> Self

Register a named value to the list.

Similar to append but uses the register_object method internally, which may perform additional validation.

Parameters:

Name	Type	Description	Default
`value`	`NamedValue`	Named value to register	required

Returns:

Name	Type	Description
`Self`	`Self`	The list instance for method chaining

Example

value_list = NamedValueList()
value_list.register_value(NamedValue("x", 1))

Source code in src/data_handlers/values.py

def register_value(self, value: NamedValue) -> Self:
    """
    Register a named value to the list.

    Similar to append but uses the register_object method internally,
    which may perform additional validation.

    Args:
        value (NamedValue): Named value to register

    Returns:
        Self: The list instance for method chaining

    Example:
        ```python
        value_list = NamedValueList()
        value_list.register_value(NamedValue("x", 1))
        ```
    """
    return self.register_object(value)

NamedValueState ¶

Bases: str, Enum

State enumeration for NamedValue objects.

This enum tracks whether a named value has been set or remains unset. Once set, values are typically frozen unless explicitly forced to change.

Attributes:

Name	Type	Description
`UNSET`		Indicates no value has been set yet
`SET`		Indicates value has been set and is frozen

repr ¶

__repr__() -> str

Get string representation for debugging.

Returns:

Name	Type	Description
`str`	`str`	The state value as a string ("unset" or "set")

Source code in src/data_handlers/values.py

def __repr__(self) -> str:
    """
    Get string representation for debugging.

    Returns:
        str: The state value as a string ("unset" or "set")
    """
    return self.value  # Returns just "unset" or "set"

str ¶

__str__() -> str

Convert state to string representation.

Returns:

Name	Type	Description
`str`	`str`	The state value as a string ("unset" or "set")

Source code in src/data_handlers/values.py

def __str__(self) -> str:
    """
    Convert state to string representation.

    Returns:
        str: The state value as a string ("unset" or "set")
    """
    return self.value  # Returns just "unset" or "set"

NormalDistribution ¶

Bases: RandomVariable[float]

Normal (Gaussian) distribution with mean μ and standard deviation σ.

The normal distribution is a continuous probability distribution that is symmetric about its mean, showing the familiar bell-shaped curve.

Key Properties

Symmetric about the mean
~68% of values lie within 1σ of μ
~95% lie within 2σ of μ
~99.7% lie within 3σ of μ

Attributes:

Name	Type	Description
`mu`	`float`	Mean (μ) of the distribution
`sigma`	`float`	Standard deviation (σ) of the distribution
`name`	`str`	Identifier for this distribution instance
`seed`	`Optional[int]`	Random seed for reproducible sampling

cdf ¶

cdf(x: ndarray, **kwargs) -> NDArray[Any, float]

Evaluate the normal cumulative distribution function.

The CDF is computed using the error function: F(x) = 1/2 * (1 + erf((x-μ)/(σ√2)))

Parameters:

Name	Type	Description	Default
`x`	`ndarray`	Points at which to evaluate the CDF	required

Other Parameters:

Name	Type	Description
`squeeze`	`bool`	Whether to remove unnecessary dimensions. Added by RandomVariableMeta. Defaults is `True`.

Returns:

Type	Description
`NDArray[Any, float]`	NDArray[Any,float]: CDF values at the input points

Source code in src/data_handlers/random_variables.py

def cdf(self, x: np.ndarray, **kwargs) -> NDArray[Any,float]:
    """Evaluate the normal cumulative distribution function.

    The CDF is computed using the error function:
    F(x) = 1/2 * (1 + erf((x-μ)/(σ√2)))

    Args:
        x (np.ndarray): Points at which to evaluate the CDF

    Other Parameters:
        squeeze (bool, optional): Whether to remove unnecessary dimensions.
            Added by RandomVariableMeta. Defaults is `True`.

    Returns:
        NDArray[Any,float]: CDF values at the input points
    """
    return 0.5 * (1 + sp.erf((x - self.mu)/(self.sigma * np.sqrt(2))))

pdf ¶

pdf(x: ndarray, **kwargs) -> NDArray[Any, float]

Evaluate the normal probability density function.

The PDF is given by: f(x) = 1/(σ√(2π)) * exp(-(x-μ)²/(2σ²))

Parameters:

Name	Type	Description	Default
`x`	`ndarray`	Points at which to evaluate the PDF	required

Other Parameters:

Name	Type	Description
`squeeze`	`bool`	Whether to remove unnecessary dimensions. Added by RandomVariableMeta. Defaults is `True`.

Returns:

Type	Description
`NDArray[Any, float]`	NDArray[Any,float]: PDF values at the input points

Source code in src/data_handlers/random_variables.py

def pdf(self, x: np.ndarray, **kwargs) -> NDArray[Any,float]:
    """Evaluate the normal probability density function.

    The PDF is given by:
    f(x) = 1/(σ√(2π)) * exp(-(x-μ)²/(2σ²))

    Args:
        x (np.ndarray): Points at which to evaluate the PDF

    Other Parameters:
        squeeze (bool, optional): Whether to remove unnecessary dimensions.
            Added by RandomVariableMeta. Defaults is `True`.

    Returns:
        NDArray[Any,float]: PDF values at the input points
    """
    return 1/(self.sigma * np.sqrt(2*np.pi)) * np.exp(-(x - self.mu)**2 / (2*self.sigma**2))

sample ¶

sample(size: int | tuple[int, ...] = 1, **kwargs) -> NDArray[Any, float]

Generate random samples from the normal distribution.

Parameters:

Name	Type	Description	Default
`size`	`int \| tuple[int, ...]`	Number or shape of samples	`1`

Other Parameters:

Name	Type	Description
`squeeze`	`bool`	Whether to remove unnecessary dimensions. Added by RandomVariableMeta. Defaults is `True`.

Returns:

Type	Description
`NDArray[Any, float]`	NDArray[Any,float]: Array of samples from N(μ, σ)

Source code in src/data_handlers/random_variables.py

def sample(self, size: int | tuple[int, ...] = 1, **kwargs) -> NDArray[Any,float]:
    """Generate random samples from the normal distribution.

    Args:
        size (int | tuple[int, ...]): Number or shape of samples

    Other Parameters:
        squeeze (bool, optional): Whether to remove unnecessary dimensions.
            Added by RandomVariableMeta. Defaults is `True`.

    Returns:
        NDArray[Any,float]: Array of samples from N(μ, σ)
    """
    rng = np.random.default_rng(self.seed)
    return rng.normal(self.mu, self.sigma, size=size)

NumericDunders ¶

A mixin class containing numeric dunder methods that can be applied to any class via composition

Attributes:

Name	Type	Description
`RESERVED_NAME`	`str`	Name of the reserved attribute that holds the value

get_value `classmethod` ¶

get_value(instance)

Get the value of the instance

Parameters:

Name	Type	Description	Default
`instance`	`Any`	Instance of the class	required

Returns:

Name	Type	Description
`Any`		Value of the instance stored in the reserved attribute or else returns the instance. For example, if the reserved attribute is 'value', then it returns `instance.value` else if the instance is just a float or an int or something like that, then it returns the instance itself.

Source code in src/data_handlers/mixins.py

@classmethod
def get_value(cls, instance):
    """
    Get the value of the instance

    Args:
        instance (Any): Instance of the class

    Returns:
        Any: Value of the instance stored in the reserved attribute or else returns the instance.
            For example, if the reserved attribute is 'value', then it returns `instance.value` else
            if the instance is just a float or an int or something like that, then it returns the instance itself.
    """
    return getattr(instance, cls.RESERVED_NAME, instance)

ObjectRegistry ¶

Generic registry for named object types.

Stores object types for deserialization from JSON files. Types are stored at the class level for consistent access across modules.

Types are stored at the class level for consistent access across modules.

Methods:

Name	Description
`register`	Register an object type
`get`	Get an object type by name
`get_all`	Get all registered object types

get `classmethod` ¶

get(category: str, name: str) -> Type[NamedObject]

Get an object type by category and name.

Source code in src/data_handlers/base.py

@classmethod
def get(cls, category: str, name: str) -> Type[NamedObject]:
    """Get an object type by category and name."""
    registry = cls.get_registry(category)
    if name not in registry:
        raise ValueError(f"{name} not found in {category} registry")
    return registry[name]

get_all `classmethod` ¶

get_all(category: str) -> list[Type[NamedObject]]

Get all registered types for a category.

Source code in src/data_handlers/base.py

@classmethod
def get_all(cls, category: str) -> list[Type[NamedObject]]:
    """Get all registered types for a category."""
    return list(cls.get_registry(category).values())

get_registry `classmethod` ¶

get_registry(category: str) -> Dict[str, Type[NamedObject]]

Get or create registry for a category.

Source code in src/data_handlers/base.py

@classmethod
def get_registry(cls, category: str) -> Dict[str, Type[NamedObject]]:
    """Get or create registry for a category."""
    # Get the actual string value if it's a PrivateAttr
    if hasattr(category, 'default'):
        category = category.default
    if category not in cls._registries:
        cls._registries[category] = {}
    return cls._registries[category]

register `classmethod` ¶

register(category: str, obj_type: Type[NamedObject]) -> None

Register an object type in its category.

Source code in src/data_handlers/base.py

@classmethod
def register(cls, category: str, obj_type: Type[NamedObject]) -> None:
    """Register an object type in its category."""
    # Get the actual string value if it's a PrivateAttr
    if hasattr(category, 'default'):
        category = category.default
    registry = cls.get_registry(category)
    registry[obj_type.__name__] = obj_type

RandomVariable ¶

Bases: NamedObject, Generic[T]

Base class for random variables.

This class provides a common interface for random variable implementations. Subclasses must implement sample(), pdf(), and cdf() methods. The metaclass ensures these methods support dimension control via the squeeze parameter.

The class is generic over the type of values it produces (T), which must be a subtype of SerializableValue to ensure proper serialization behavior.

Attributes:

Name	Type	Description
`_registry_category`	`str`	Category name for object registration
`seed`	`Optional[int]`	Random seed for reproducible sampling
`name`	`str`	Identifier for this random variable instance

Type Variables

T: The type of values produced by this random variable

cdf ¶

cdf(x: ndarray, **kwargs) -> NDArray[Any, T]

Evaluate cumulative distribution function at specified points.

Parameters:

Name	Type	Description	Default
`x`	`ndarray`	Points at which to evaluate the CDF	required

Other Parameters:

Name	Type	Description
`squeeze`	`bool`	Whether to remove unnecessary dimensions. Added by RandomVariableMeta. Defaults is `True`.

Returns:

Type	Description
`NDArray[Any, T]`	CDF values at the input points

Raises:

Type	Description
`NotImplementedError`	Must be implemented by subclasses

Source code in src/data_handlers/random_variables.py

def cdf(self, x: np.ndarray, **kwargs) -> NDArray[Any, T]:
    """Evaluate cumulative distribution function at specified points.

    Args:
        x (np.ndarray): Points at which to evaluate the CDF

    Other Parameters:
        squeeze (bool, optional): Whether to remove unnecessary dimensions.
            Added by RandomVariableMeta. Defaults is `True`.

    Returns:
        (NDArray[Any, T]): CDF values at the input points

    Raises:
        NotImplementedError: Must be implemented by subclasses
    """
    raise NotImplementedError

pdf ¶

pdf(x: ndarray, **kwargs) -> NDArray[Any, T]

Evaluate probability density function at specified points.

Parameters:

Name	Type	Description	Default
`x`	`ndarray`	Points at which to evaluate the PDF	required

Other Parameters:

Name	Type	Description
`squeeze`	`bool`	Whether to remove unnecessary dimensions. Added by RandomVariableMeta. Defaults is `True`.

Returns:

Type	Description
`NDArray[Any, T]`	PDF values at the input points

Raises:

Type	Description
`NotImplementedError`	Must be implemented by subclasses

Source code in src/data_handlers/random_variables.py

def pdf(self, x: np.ndarray, **kwargs) -> NDArray[Any, T]:
    """Evaluate probability density function at specified points.

    Args:
        x (np.ndarray): Points at which to evaluate the PDF

    Other Parameters:
        squeeze (bool, optional): Whether to remove unnecessary dimensions.
            Added by RandomVariableMeta. Defaults is `True`.

    Returns:
        (NDArray[Any, T]): PDF values at the input points

    Raises:
        NotImplementedError: Must be implemented by subclasses
    """
    raise NotImplementedError

register_to_hash ¶

register_to_hash(
    var_hash: RandomVariableHash, size: int = 1, sample: bool = True, squeeze: bool = True
) -> NamedValue[T | NDArray[Any, T]]

Register this random variable to a hash and return sampled values.

This is a convenience method for adding a random variable to a collection and immediately sampling from it.

Parameters:

Name	Type	Description	Default
`var_hash`	`RandomVariableHash`	Hash object to register to	required
`size`	`int`	Number of samples to generate	`1`

Returns:

Type	Description
`NamedValue[T \| NDArray[Any, T]]`	Named value containing samples

Source code in src/data_handlers/random_variables.py

def register_to_hash(
        self, 
        var_hash: RandomVariableHash, 
        size: int = 1, 
        sample: bool = True,
        squeeze: bool = True,
    ) -> NamedValue[T|NDArray[Any,T]]:
    """Register this random variable to a hash and return sampled values.

    This is a convenience method for adding a random variable to a collection
    and immediately sampling from it.

    Args:
        var_hash (RandomVariableHash): Hash object to register to
        size (int): Number of samples to generate

    Returns:
        (NamedValue[T|NDArray[Any,T]]): Named value containing samples
    """
    return var_hash.register_variable(self, size=size, sample=sample, squeeze=squeeze)

sample ¶

sample(size: int = 1, **kwargs) -> NDArray[Any, T]

Generate random samples from the categorical distribution.

Parameters:

Name	Type	Description	Default
`size`	`int`	Number of samples to generate. Defaults to 1.	`1`

Other Parameters:

Name	Type	Description
`squeeze`	`bool`	Whether to remove unnecessary dimensions. Added by RandomVariableMeta. Defaults is `True`.

Returns:

Type	Description
`NDArray[Any, T]`	Array of samples from the categories

Raises:

Type	Description
`NotImplementedError`	Must be implemented by subclasses

Source code in src/data_handlers/random_variables.py

def sample(self, size: int = 1, **kwargs) -> NDArray[Any, T]:
    """Generate random samples from the categorical distribution.

    Args:
        size (int): Number of samples to generate. Defaults to 1.

    Other Parameters:
        squeeze (bool, optional): Whether to remove unnecessary dimensions.
            Added by RandomVariableMeta. Defaults is `True`.

    Returns:
        (NDArray[Any, T]): Array of samples from the categories

    Raises:
        NotImplementedError: Must be implemented by subclasses
    """
    raise NotImplementedError

RandomVariableHash ¶

Bases: NamedObjectHash

Collection of random variables.

This class manages a collection of RandomVariable objects, providing methods to register, retrieve and sample from multiple distributions.

Attributes:

Name	Type	Description
`_registry_category`	`str`	Category name for object registration

get_variables ¶

get_variables() -> Iterable[RandomVariable]

Get all registered random variables.

Returns:

Type	Description
`Iterable[RandomVariable]`	Iterable[RandomVariable]: Iterator over all registered variables

Source code in src/data_handlers/random_variables.py

def get_variables(self) -> Iterable[RandomVariable]:
    """Get all registered random variables.

    Returns:
        Iterable[RandomVariable]: Iterator over all registered variables
    """
    return self.get_objects()

register_variable ¶

register_variable(
    var: RandomVariable, size: int = 1, sample: bool = True, squeeze: bool = True
) -> NamedValue[SerializableValue | NDArray[Any, SerializableValue]]

Register a random variable and return its samples wrapped in a NamedValue.

Parameters:

Name	Type	Description	Default
`var`	`RandomVariable`	Random variable to register	required
`size`	`int`	Number of samples to generate	`1`

Returns:

Type	Description
`NamedValue[SerializableValue \| NDArray[Any, SerializableValue]]`	NamedValue[SerializableValue\|NDArray[Any,SerializableValue]]: Named value containing samples

Raises:

Type	Description
`ValueError`	If a variable with the same name already exists

Source code in src/data_handlers/random_variables.py

def register_variable(
        self, 
        var: RandomVariable, 
        size: int = 1,
        sample: bool = True,
        squeeze: bool = True,
    ) -> NamedValue[SerializableValue|NDArray[Any,SerializableValue]]:
    """Register a random variable and return its samples wrapped in a NamedValue.

    Args:
        var (RandomVariable): Random variable to register
        size (int): Number of samples to generate

    Returns:
        NamedValue[SerializableValue|NDArray[Any,SerializableValue]]: Named value containing samples

    Raises:
        ValueError: If a variable with the same name already exists
    """
    if var.name in self.objects:
        raise ValueError(f"A variable with name '{var.name}' already exists in the collection")

    self.register_object(var)
    if sample:
        samples = var.sample(size=size)
        if squeeze:
            return NamedValue(name=var.name, value=samples.squeeze())
        else:
            return NamedValue(name=var.name, value=samples)
    else:
        return None

sample_all ¶

sample_all(size: int = 1) -> dict[str, ndarray]

Sample from all registered distributions.

Parameters:

Name	Type	Description	Default
`size`	`int`	Number of samples to generate per distribution	`1`

Returns:

Type	Description
`dict[str, ndarray]`	dict[str, np.ndarray]: Dictionary mapping variable names to their samples

Source code in src/data_handlers/random_variables.py

def sample_all(self, size: int = 1) -> dict[str, np.ndarray]:
    """Sample from all registered distributions.

    Args:
        size (int): Number of samples to generate per distribution

    Returns:
        dict[str, np.ndarray]: Dictionary mapping variable names to their samples
    """
    return {name: var.sample(size) for name, var in self.objects.items()}

RandomVariableList ¶

Bases: NamedObjectList

List of random variables.

This class manages an ordered list of RandomVariable objects, providing methods to add, access, and sample from multiple distributions while maintaining their order.

Attributes:

Name	Type	Description
`_registry_category`	`str`	Category name for object registration
`objects`	`List[SerializeAsAny[InstanceOf[RandomVariable]]]`	List of random variable objects

getitem ¶

__getitem__(idx: int) -> RandomVariable

Get a random variable by its index in the list.

Parameters:

Name	Type	Description	Default
`idx`	`int`	Index of the random variable to retrieve	required

Returns:

Name	Type	Description
`RandomVariable`	`RandomVariable`	The random variable at the specified index

Raises:

Type	Description
`IndexError`	If the index is out of range

Source code in src/data_handlers/random_variables.py

def __getitem__(self, idx: int) -> RandomVariable:
    """Get a random variable by its index in the list.

    Args:
        idx (int): Index of the random variable to retrieve

    Returns:
        RandomVariable: The random variable at the specified index

    Raises:
        IndexError: If the index is out of range
    """
    return super().__getitem__(idx)

append ¶

append(variable: RandomVariable) -> Self

Append a random variable to the end of the list.

Parameters:

Name	Type	Description	Default
`variable`	`RandomVariable`	Random variable to append	required

Returns:

Name	Type	Description
`Self`	`Self`	The RandomVariableList instance for method chaining

Source code in src/data_handlers/random_variables.py

def append(self, variable: RandomVariable) -> Self:
    """Append a random variable to the end of the list.

    Args:
        variable (RandomVariable): Random variable to append

    Returns:
        Self: The RandomVariableList instance for method chaining
    """
    return super().append(variable)

extend ¶

extend(variables: Iterable[RandomVariable]) -> Self

Extend the list with multiple random variables.

Parameters:

Name	Type	Description	Default
`variables`	`Iterable[RandomVariable]`	Collection of random variables to add	required

Returns:

Name	Type	Description
`Self`	`Self`	The RandomVariableList instance for method chaining

Source code in src/data_handlers/random_variables.py

def extend(self, variables: Iterable[RandomVariable]) -> Self:
    """Extend the list with multiple random variables.

    Args:
        variables (Iterable[RandomVariable]): Collection of random variables to add

    Returns:
        Self: The RandomVariableList instance for method chaining
    """
    return super().extend(variables)

get_variable ¶

get_variable(name: str) -> RandomVariable

Get a registered random variable by name.

Parameters:

Name	Type	Description	Default
`name`	`str`	Name of the random variable	required

Returns:

Name	Type	Description
`RandomVariable`	`RandomVariable`	The requested random variable

Raises:

Type	Description
`KeyError`	If no variable exists with the given name

Source code in src/data_handlers/random_variables.py

def get_variable(self, name: str) -> RandomVariable:
    """Get a registered random variable by name.

    Args:
        name (str): Name of the random variable

    Returns:
        RandomVariable: The requested random variable

    Raises:
        KeyError: If no variable exists with the given name
    """
    return self.get_object(name)

get_variables ¶

get_variables() -> Iterable[RandomVariable]

Get all registered random variables.

Returns:

Type	Description
`Iterable[RandomVariable]`	Iterable[RandomVariable]: Iterator over all registered variables

Source code in src/data_handlers/random_variables.py

def get_variables(self) -> Iterable[RandomVariable]:
    """Get all registered random variables.

    Returns:
        Iterable[RandomVariable]: Iterator over all registered variables
    """
    return self.get_objects()

register_variable ¶

register_variable(var: RandomVariable) -> Self

Register a random variable to the collection.

Parameters:

Name	Type	Description	Default
`var`	`RandomVariable`	Random variable to register	required

Returns:

Name	Type	Description
`Self`	`Self`	The RandomVariableList instance

Source code in src/data_handlers/random_variables.py

def register_variable(self, var: RandomVariable) -> Self:
    """Register a random variable to the collection.

    Args:
        var (RandomVariable): Random variable to register

    Returns:
        Self: The RandomVariableList instance
    """
    return self.register_object(var)

sample_all ¶

sample_all(size: int = 1) -> dict[str, ndarray]

Sample from all variables in the list.

Parameters:

Name	Type	Description	Default
`size`	`int`	Number of samples to generate per variable	`1`

Returns:

Type	Description
`dict[str, ndarray]`	dict[str, np.ndarray]: Dictionary mapping variable names to their samples

Source code in src/data_handlers/random_variables.py

def sample_all(self, size: int = 1) -> dict[str, np.ndarray]:
    """Sample from all variables in the list.

    Args:
        size (int): Number of samples to generate per variable

    Returns:
        dict[str, np.ndarray]: Dictionary mapping variable names to their samples
    """
    return {var.name: var.sample(size) for var in self.objects}

UniformDistribution ¶

Bases: RandomVariable[float]

Continuous uniform distribution over an interval [low, high].

The uniform distribution describes equal probability over a continuous interval. Any value between low and high is equally likely to be drawn.

Key Properties

Mean = (low + high)/2
Variance = (high - low)²/12
Constant PDF over [low, high]
Linear CDF over [low, high]

Attributes:

Name	Type	Description
`low`	`float`	Lower bound of the interval
`high`	`float`	Upper bound of the interval
`name`	`str`	Identifier for this distribution instance
`seed`	`Optional[int]`	Random seed for reproducible sampling

cdf ¶

cdf(x: ndarray, **kwargs) -> NDArray[Any, float]

Evaluate the uniform cumulative distribution function.

The CDF is: - 0 for x < low - (x-low)/(high-low) for low ≤ x ≤ high - 1 for x > high

Parameters:

Name	Type	Description	Default
`x`	`ndarray`	Points at which to evaluate the CDF	required

Other Parameters:

Name	Type	Description
`squeeze`	`bool`	Whether to remove unnecessary dimensions. Added by RandomVariableMeta. Defaults is `True`.

Returns:

Type	Description
`NDArray[Any, float]`	NDArray[Any,float]: CDF values at the input points

Source code in src/data_handlers/random_variables.py

def cdf(self, x: np.ndarray, **kwargs) -> NDArray[Any,float]:
    """Evaluate the uniform cumulative distribution function.

    The CDF is:
    - 0 for x < low
    - (x-low)/(high-low) for low ≤ x ≤ high
    - 1 for x > high

    Args:
        x (np.ndarray): Points at which to evaluate the CDF

    Other Parameters:
        squeeze (bool, optional): Whether to remove unnecessary dimensions.
            Added by RandomVariableMeta. Defaults is `True`.

    Returns:
        NDArray[Any,float]: CDF values at the input points
    """
    return np.where(
        x < self.low,
        0.0,
        np.where(
            x > self.high,
            1.0,
            (x - self.low) / (self.high - self.low)
        )
    )

pdf ¶

pdf(x: ndarray, **kwargs) -> NDArray[Any, float]

Evaluate the uniform probability density function.

The PDF is 1/(high-low) for x in [low,high] and 0 elsewhere.

Parameters:

Name	Type	Description	Default
`x`	`ndarray`	Points at which to evaluate the PDF	required

Other Parameters:

Name	Type	Description
`squeeze`	`bool`	Whether to remove unnecessary dimensions. Added by RandomVariableMeta. Defaults is `True`.

Returns:

Type	Description
`NDArray[Any, float]`	NDArray[Any,float]: PDF values at the input points

Source code in src/data_handlers/random_variables.py

def pdf(self, x: np.ndarray, **kwargs) -> NDArray[Any,float]:
    """Evaluate the uniform probability density function.

    The PDF is 1/(high-low) for x in [low,high] and 0 elsewhere.

    Args:
        x (np.ndarray): Points at which to evaluate the PDF

    Other Parameters:
        squeeze (bool, optional): Whether to remove unnecessary dimensions.
            Added by RandomVariableMeta. Defaults is `True`.

    Returns:
        NDArray[Any,float]: PDF values at the input points
    """
    return np.where(
        (x >= self.low) & (x <= self.high),
        1.0 / (self.high - self.low),
        0.0
    )

sample ¶

sample(size: int = 1, **kwargs) -> NDArray[Any, float]

Generate random samples from the uniform distribution.

Parameters:

Name	Type	Description	Default
`size`	`int`	Number of samples to generate	`1`

Other Parameters:

Name	Type	Description
`squeeze`	`bool`	Whether to remove unnecessary dimensions. Added by RandomVariableMeta. Defaults is `True`.

Returns:

Type	Description
`NDArray[Any, float]`	NDArray[Any,float]: Array of samples from U(low,high)

Source code in src/data_handlers/random_variables.py

def sample(self, size: int = 1, **kwargs) -> NDArray[Any,float]:
    """Generate random samples from the uniform distribution.

    Args:
        size (int): Number of samples to generate

    Other Parameters:
        squeeze (bool, optional): Whether to remove unnecessary dimensions.
            Added by RandomVariableMeta. Defaults is `True`.

    Returns:
        NDArray[Any,float]: Array of samples from U(low,high)
    """
    rng = np.random.default_rng(self.seed)
    return rng.uniform(self.low, self.high, size=size)

validate_bounds ¶

validate_bounds() -> UniformDistribution

Validate that high > low.

Source code in src/data_handlers/random_variables.py

@model_validator(mode='after')
def validate_bounds(self) -> UniformDistribution:
    """Validate that high > low."""
    if self.high <= self.low:
        raise ValueError(f"Upper bound ({self.high}) must be greater than lower bound ({self.low})")
    return self

data_handlers

data_handlers ¶

CategoricalDistribution ¶

cdf ¶

pdf ¶

sample ¶

validate_and_set_probabilities ¶

NamedObject ¶

type property ¶

__init_subclass__ ¶

NamedObjectHash ¶

deserialize_objects classmethod ¶

get_object ¶

get_object_names ¶

get_objects ¶

register_object ¶

NamedObjectList ¶

__getitem__ ¶

__iter__ ¶

__len__ ¶

append ¶

extend ¶

from_iterable classmethod ¶

get_object ¶

get_objects ¶

register_object ¶

NamedValue ¶

value property writable ¶

__setattr__ ¶

append_to_value_list ¶

force_set_value ¶

model_dump ¶

model_dump_json ¶

model_validate classmethod ¶

model_validate_json classmethod ¶

register_to_value_hash ¶

NamedValueHash ¶

get_raw_value ¶

get_raw_values ¶

get_value ¶

get_value_by_type ¶

get_value_names ¶

get_values ¶

model_dump ¶

model_dump_json ¶

model_validate classmethod ¶

model_validate_json classmethod ¶

register_value ¶

set_raw_value ¶

NamedValueList ¶

__getitem__ ¶

append ¶

extend ¶

get_value ¶

get_value_by_type ¶

get_values ¶

model_dump ¶

model_dump_json ¶

model_validate classmethod ¶

model_validate_json classmethod ¶

register_value ¶

NamedValueState ¶

__repr__ ¶

__str__ ¶

NormalDistribution ¶

cdf ¶

pdf ¶

sample ¶

NumericDunders ¶

get_value classmethod ¶

ObjectRegistry ¶

get classmethod ¶

get_all classmethod ¶

get_registry classmethod ¶

register classmethod ¶

RandomVariable ¶

cdf ¶

pdf ¶

register_to_hash ¶

sample ¶

type `property` ¶

deserialize_objects `classmethod` ¶

getitem ¶

iter ¶

len ¶

from_iterable `classmethod` ¶

value `property` `writable` ¶

setattr ¶

model_validate `classmethod` ¶

model_validate_json `classmethod` ¶

model_validate `classmethod` ¶

model_validate_json `classmethod` ¶

getitem ¶

model_validate `classmethod` ¶

model_validate_json `classmethod` ¶

repr ¶

str ¶

get_value `classmethod` ¶

get `classmethod` ¶

get_all `classmethod` ¶

get_registry `classmethod` ¶

register `classmethod` ¶

getitem ¶