Probes

class domprob.probes.probe.Probe(dispatcher)

Bases: object

A class representing a probes that facilitates the dispatching of observations.

dispatcher

The dispatcher responsible for handling observations.

Type:

_DispT

Example

>>> from domprob import (
...     sensor,
...     BaseObservation,
...     BasicConsumer,
...     BasicDispatcher,
...     Probe,
... )
>>>
>>> class SomeInstrument:
...
...     @staticmethod
...     def call(msg: str) -> None:
...         print(msg)
...
>>>
>>> class SampleObservation(BaseObservation):
...
...     @staticmethod
...     @sensor(SomeInstrument)
...     def sense_msg(some_instrument: SomeInstrument) -> None:
...         some_instrument.call("Sensed!")
...
>>> consumer = BasicConsumer(SomeInstrument())
>>> dispatcher = BasicDispatcher(consumer)
>>> probes = Probe(dispatcher)
>>>
>>> probes.observe(SampleObservation())
Sensed!

__eq__(other)

Determines if two Probe instances are equal.

Probes are considered equal if they are of the same type and have the same dispatcher.

Parameters:

other (Any) – The object to compare with.

Returns:

True if the probes are equal, False otherwise.

Return type:

bool

Example

>>> from domprob import BasicDispatcher
>>>
>>> probe1 = Probe(BasicDispatcher())
>>> probe2 = Probe(BasicDispatcher())
>>>
>>> assert probe1 == probe2

__repr__()

Returns a string representation of the Probe instance.

Returns:

A formatted string representing the instance.

Return type:

str

Example

>>> from domprob import BasicDispatcher
>>>
>>> probes = Probe(BasicDispatcher())
>>> repr(probes)
'Probe(dispatcher=BasicDispatcher(consumers=()))'

observe(observation)

Dispatches an observation using the associated dispatcher.

Parameters:

observation (ObservationProtocol) – The observation to be dispatched.

Return type:

None

domprob.probes.probe.get_probe(*instruments)

Creates a Probe instance with the provided instruments.

If no instruments are provided, it defaults to using a default logging.Logger instance.

Parameters:

*instruments (_InstrumT) – The instrument implementations to be used to fulfil the observations.

Returns:

A new Probe instance initialized with a

BasicDispatcher.

Return type:

Probe

Example

>>> from domprob import get_probe
>>>
>>> # Create a probes with a custom instrument
>>> custom_probe = get_probe(logging.getLogger("custom"))
>>> custom_probe
Probe(dispatcher=BasicDispatcher(consumers=(BasicConsumer(instruments=('<Logger custom (WARNING)>',)),)))
>>>
>>> # Create a probes with default instruments
>>> default_probe = get_probe()
>>> default_probe
Probe(dispatcher=BasicDispatcher(consumers=(BasicConsumer(instruments=('<Logger default (DEBUG)>',)),)))

domprob.probes.probe.probe = Probe(dispatcher=BasicDispatcher(consumers=(BasicConsumer(instruments=('<Logger default (DEBUG)>',)),)))

The default probe.

Example

>>> from domprob.probes.probe import probe
>>>
>>> probe
Probe(dispatcher=BasicDispatcher(instruments=('<RootLogger root (WARNING)>',)))