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 announcement, BaseObservation
>>>
>>> class SomeInstrument:
...     pass
...
>>> class MyObservation(BaseObservation):
...     @announcement(SomeInstrument)
...     def my_method(self, instrument: SomeInstrument) -> str:
...         pass
...
>>> observation = MyObservation()
>>> observation
MyObservation(announcements=1)
__len__()[source]

Return the number of announcements.

Returns:

Count of announcements in the class.

Return type:

int

Example

>>> from domprob import announcement, BaseObservation
>>>
>>> class SomeInstrument:
...     pass
...
>>> class MyObservation(BaseObservation):
...     @announcement(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>
_announcements: list[AnnouncementMethod] | None = None
_is_protocol = False
_is_runtime_protocol = True
classmethod announcements()[source]

Yield announcement methods defined in the class.

Uses lazy evaluation to avoid unnecessary memory consumption.

Yields:

_AnnounceSig – Announcement method instances.

Return type:

Generator[AnnouncementMethod, None, None]

Example

>>> from domprob import announcement, BaseObservation
>>>
>>> class SomeInstrument:
...     pass
...
>>> class MyObservation(BaseObservation):
...     @announcement(SomeInstrument)
...     def event_occurred(self, instrument: SomeInstrument) -> None:
...         pass
...
>>> gen = MyObservation.announcements()
>>> list(gen)
[AnnouncementMethod(meth=<function MyObservation.event_occurred at 0x...>)]
domprob.observations.base._is_function(obj)[source]

Check if an object is a function that is not a property or dunder method.

Parameters:

obj (object) – The object to check.

Returns:

True if the object is a regular function, False

otherwise.

Return type:

bool

Example

>>> def example_func(): pass
>>> _is_function(example_func)
True
>>> class Example:
...     @property
...     def prop(self): return 42
...
>>> _is_function(Example.prop)
False
class domprob.observations.observation.ObservationProtocol(*args, **kwargs)[source]

Bases: Protocol

Protocol defining the structure of domain observations that provide announcements.

Classes implementing this protocol must define a @classmethod named announcements that returns a Generator of AnnouncementMethod 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

announcement method.

_R_co (TypeVar): Represents the return type of the announcement

method.

Example

>>> class ConcreteObservation:
...     @classmethod
...     def announcements(cls) -> Generator[AnnouncementMethod, None, None]:
...         yield AnnouncementMethod(lambda x: x)
...
>>> assert isinstance(ConcreteObservation, ObservationProtocol)
_abc_impl = <_abc._abc_data object>
_is_protocol = True
_is_runtime_protocol = True
classmethod announcements()[source]

Retrieve all announcement methods defined in the class.

Returns:

A generator yielding

AnnouncementMethod instances.

Return type:

Generator[_AnnounceSig, None, None]