modifiers

ByAnimalModifier

ByAnimalModifier(
    subject: str,
    store: Store,
    kind: KindLike[Any],
    model_path: str,
)

Bases: ABC, Generic[TRig]

Injects per-animal state into a rig, and saves it back, through a :class:~clabe.stores.Store.

The store decides where per-animal state lives, so the same modifier works against the config library, Dataverse or an in-memory fake. The store is narrowed to subject on construction.

Example

class ManipulatorModifier(ByAnimalModifier[MyRig]):
    def __init__(self, subject: str, store: Store):
        super().__init__(subject, store, Kind(ManipulatorPosition), "manipulator.position")

    def _process_before_update(self) -> ManipulatorPosition:
        return read_position_from_hardware()

modifier = ManipulatorModifier(session.subject, store)
rig = modifier.inject(rig)
...
modifier.update()

Parameters:

Name	Type	Description	Default
subject	str	The animal this modifier reads and writes state for.	required
store	Store	The store holding this record. Narrowed to subject here.	required
kind	KindLike[Any]	The record kind to read and write.	required
model_path	str	Dot-separated path to the target attribute in the rig model.	required

Raises:

Type	Description
ValueError	If subject is empty.

Source code in src/clabe/modifiers.py

def __init__(self, subject: str, store: Store, kind: KindLike[Any], model_path: str) -> None:
    """
    Args:
        subject: The animal this modifier reads and writes state for.
        store: The store holding this record. Narrowed to ``subject`` here.
        kind: The record kind to read and write.
        model_path: Dot-separated path to the target attribute in the rig model.

    Raises:
        ValueError: If ``subject`` is empty.
    """
    if not subject:
        raise ValueError("ByAnimalModifier requires a subject.")
    self._subject = subject
    self._store = store.scoped(subject=subject)
    self._kind: Kind[Any] = as_kind(kind)
    self._model_path = model_path

inject

inject(rig: TRig) -> TRig

Injects the stored record into the rig, leaving the rig untouched if there is none.

Parameters:

Name	Type	Description	Default
rig	TRig	The rig model to modify.	required

Returns:

Type	Description
TRig	The rig, modified in place.

Source code in src/clabe/modifiers.py

def inject(self, rig: TRig) -> TRig:
    """
    Injects the stored record into the rig, leaving the rig untouched if there is none.

    Args:
        rig: The rig model to modify.

    Returns:
        The rig, modified in place.
    """
    records = self._store.list(self._kind)
    if not records:
        logger.warning(
            "No %s found for subject %s in %s. Using default.", self._kind.name, self._subject, self._store
        )
        return rig
    logger.info("Loading %s for subject %s. Deserialized: %s", self._kind.name, self._subject, records[0])
    recursive_setattr(rig, self._model_path, self._process_before_inject(records[0]))
    return rig

update

update() -> None

Persists the record produced by :meth:_process_before_update.

Raises:

Type	Description
Exception	Whatever :meth:_process_before_update or the store raises.

Source code in src/clabe/modifiers.py

def update(self) -> None:
    """
    Persists the record produced by :meth:`_process_before_update`.

    Raises:
        Exception: Whatever :meth:`_process_before_update` or the store raises.
    """
    try:
        to_update = self._process_before_update()
        logger.info("Saving %s for subject %s. Serialized: %s", self._kind.name, self._subject, to_update)
        self._store.write(self._kind, to_update)
    except Exception as e:
        logger.error("Failed to update modifier: %s", e)
        raise

recursive_setattr

recursive_setattr(obj, attr, val)

Sets an attribute value using a dot-separated path.

Parameters:

Name	Type	Description	Default
obj		The object to modify	required
attr		Dot-separated attribute path (e.g., "nested.field.value")	required
val		The value to set	required

Returns:

Type	Description
	The result of setattr on the final attribute

Example

class Inner:
    value = 1

class Outer:
    inner = Inner()

obj = Outer()
recursive_setattr(obj, "inner.value", 42)
assert obj.inner.value == 42

Source code in src/clabe/modifiers.py

def recursive_setattr(obj, attr, val):
    """
    Sets an attribute value using a dot-separated path.

    Args:
        obj: The object to modify
        attr: Dot-separated attribute path (e.g., "nested.field.value")
        val: The value to set

    Returns:
        The result of setattr on the final attribute

    Example:
        ```python
        class Inner:
            value = 1

        class Outer:
            inner = Inner()

        obj = Outer()
        recursive_setattr(obj, "inner.value", 42)
        assert obj.inner.value == 42
        ```
    """
    pre, _, post = attr.rpartition(".")
    return setattr(rgetattr(obj, pre) if pre else obj, post, val)

rgetattr

rgetattr(obj, attr, *args)

Gets an attribute value using a dot-separated path.

Parameters:

Name	Type	Description	Default
obj		The object to query	required
attr		Dot-separated attribute path (e.g., "nested.field.value")	required
*args		Optional default value if attribute doesn't exist	()

Returns:

Type	Description
	The attribute value at the specified path

Example

class Inner:
    value = 42

class Outer:
    inner = Inner()

obj = Outer()
result = rgetattr(obj, "inner.value")
assert result == 42

default = rgetattr(obj, "nonexistent.path", "default")
assert default == "default"

Source code in src/clabe/modifiers.py

def rgetattr(obj, attr, *args):
    """
    Gets an attribute value using a dot-separated path.

    Args:
        obj: The object to query
        attr: Dot-separated attribute path (e.g., "nested.field.value")
        *args: Optional default value if attribute doesn't exist

    Returns:
        The attribute value at the specified path

    Example:
        ```python
        class Inner:
            value = 42

        class Outer:
            inner = Inner()

        obj = Outer()
        result = rgetattr(obj, "inner.value")
        assert result == 42

        default = rgetattr(obj, "nonexistent.path", "default")
        assert default == "default"
        ```
    """

    def _getattr(obj, attr):
        """Helper function to get attribute with optional default."""
        return getattr(obj, attr, *args)

    return functools.reduce(_getattr, [obj] + attr.split("."))