Observations

class domprob.observations.base.BaseObservation(*args, **kwargs)[source]

Bases: ABC, ObservationProtocol

Base class for observations.

__slots__

Prevents the creation of instance __dict__ to keep memory footprint low.

Type:

tuple

Example

>>> from domprob import sensor, BaseObservation
>>>
>>> class SomeInstrument:
...     pass
...
>>> class MyObservation(BaseObservation):
...     @sensor(SomeInstrument)
...     def my_method(self, instrument: SomeInstrument) -> str:
...         pass
...
>>> observation = MyObservation()
>>> observation
MyObservation(sensors=1)
__len__()[source]

Return the number of sensors.

Returns:

Count of sensors in the class.

Return type:

int

Example

>>> from domprob import sensor, BaseObservation
>>>
>>> class SomeInstrument:
...     pass
...
>>> class MyObservation(BaseObservation):
...     @sensor(SomeInstrument)
...     def my_method(self, instrument: SomeInstrument) -> str:
...         pass
...
>>> observation = MyObservation()
>>> len(observation)
1
__repr__()[source]

Return repr(self).

Return type:

str

_abc_impl = <_abc._abc_data object>
_is_protocol = False
_is_runtime_protocol = True
_sensors: SensorSet | None = None
classmethod sensors()[source]

Yield sensors methods defined in the class.

Uses lazy evaluation to avoid unnecessary memory consumption.

Yields:

_SensorSig – Sensor method instances.

Return type:

SensorSet

Example

>>> from domprob import sensor, BaseObservation
>>>
>>> class SomeInstrument:
...     pass
...
>>> class MyObservation(BaseObservation):
...     @sensor(SomeInstrument)
...     def event_occurred(self, instrument: SomeInstrument) -> None:
...         pass
...
>>> gen = MyObservation.sensors()
>>> list(gen)
[SensorMethod(meth=<function MyObservation.event_occurred at 0x...>)]
class domprob.observations.base.SensorSet(*sensor_methods)[source]

Bases: Set[SensorMethod[_P, _R]]

A custom set-like collection for storing SensorMethod instances.

This class ensures unique sensors methods and provides set-like behavior for iteration, containment checks, and length retrieval.

Parameters:

*sensor_methods (_SensorSig) – One or more sensors method instances.

Example

>>> from domprob import sensor
>>>
>>> class MyObservation:
...
...     @sensor(...)
...     def sense_hello(self, _):
...         pass
...
...     def normal_method(self, _):
...         pass
...
>>> meth = SensorMethod(MyObservation.sense_hello)
>>> sensor_set = SensorSet(meth, meth)
>>> len(sensor_set)
1
__contains__(item)[source]

Checks if a given sensors method exists in the set.

Parameters:

item (Any) – The item to check.

Returns:

True if item is an instance of SensorMethod and

exists in the set, False otherwise.

Return type:

bool

__eq__(other)

Return self==value.

__iter__()[source]

Returns an iterator over the sensors methods in the set.

Yields:

_SensorSig – Each sensors method stored in the set.

Return type:

Generator[SensorMethod[ParamSpec(_P, bound= None), TypeVar(_R)], None, None]

__len__()[source]

Returns the number of sensors methods in the set.

Returns:

The count of stored sensors methods.

Return type:

int

__repr__()[source]

Returns a string representation of the SensorSet.

Returns:

A string describing the number of stored

sensors.

Return type:

str

_abc_impl = <_abc._abc_data object>
classmethod _from_iterable(it)

Construct an instance of the class from any iterable input.

Must override this method if the class constructor signature does not accept an iterable for an input.

_hash()

Compute the hash value of a set.

Note that we don’t define __hash__: not all sets are hashable. But if you define a hashable set type, its __hash__ should call this function.

This must be compatible __eq__.

All sets ought to compare equal if they contain the same elements, regardless of how they are implemented, and regardless of the order of the elements; so there’s not much freedom for __eq__ or __hash__. We match the algorithm used by the built-in frozenset type.

classmethod from_observation(observation_cls)[source]

Creates an SensorSet by extracting sensors methods from a given class.

This method inspects the provided class, identifies methods that qualify as sensors methods using SensorMethod.from_callable, and includes them in the returned SensorSet.

Parameters:

observation_cls (Any) – The class to inspect for sensors methods.

Returns:

A set of extracted sensors methods.

Return type:

SensorSet

Example

>>> from domprob import sensor
>>>
>>> class MyObservation:
...
...     @sensor(...)
...     def sense_hello(self, _):
...         pass
...
...     def normal_method(self, _):
...         pass
...
>>> sensor_set = SensorSet.from_observation(MyObservation)
>>> len(sensor_set)
1
isdisjoint(other)

Return True if two sets have a null intersection.

class domprob.observations.observation.ObservationProtocol(*args, **kwargs)[source]

Bases: Protocol

Protocol defining the structure of domain observations that provide sensors.

Classes implementing this protocol must define a @classmethod named sensors that returns a Generator of SensorMethod instances.

This protocol is @runtime_checkable, meaning isinstance(obj, ObservationProtocol) can be used to verify implementation at runtime.

Type Parameters:
_P (ParamSpec): Represents the parameters accepted by the

sensors’ method.

_R_co (TypeVar): Represents the return type of the sensors’

method.

Example

>>> from domprob.sensors.meth import SensorMethod
>>> from domprob.observations.observation import ObservationProtocol
>>>
>>> class ConcreteObservation:
...     @classmethod
...     def sensors(cls) -> Iterable[SensorMethod]:
...         yield SensorMethod(lambda x: x)
...
>>> assert isinstance(ConcreteObservation, ObservationProtocol)
_abc_impl = <_abc._abc_data object>
_is_protocol = True
_is_runtime_protocol = True
classmethod sensors()[source]

Retrieve all sensors methods defined in the class.

Returns:

A generator yielding

SensorMethod instances.

Return type:

Generator[_SensorSig, None, None]