Observable¶

Reactive primitives — Observable, Subject, BehaviorSubject, ReplaySubject — plus the pipe operators used throughout the SDK.

camera_ui_sdk.observable ¶

OperatorFn `module-attribute` ¶

OperatorFn = Callable[['Observable[Any]'], 'Observable[Any]']

Function that transforms one Observable into another. Used as a building block for pipe() operator chains.

Disposable ¶

Disposable(teardown: Callable[[], None])

Subscription handle returned by subscribe().

Call :meth:dispose (or its alias :meth:unsubscribe) to detach the listener and run any teardown logic registered by the producer. Disposing twice is a no-op.

closed `property` ¶

closed: bool

Whether the subscription has already been disposed.

dispose ¶

dispose() -> None

Detach the listener and run the producer teardown. Disposing twice is a no-op.

Example

sub = sensor.onPropertyChanged.subscribe(handle)
sub.dispose()

unsubscribe ¶

unsubscribe() -> None

Alias for :meth:dispose, for rxjs-shaped call sites.

Observable ¶

Observable(subscribe_fn: Callable[[Callable[[T], None]], Disposable])

Bases: Generic[T]

Cold producer of a push-based value stream.

The subscribe_fn passed to the constructor is executed once per :meth:subscribe call, so each subscriber gets its own independent run. :meth:subscribe returns a :class:Disposable that stops the stream and triggers any teardown registered by the producer.

subscribe(callback: Callable[[T], None]) -> Disposable

Start the producer for this subscriber and route emitted values to callback.

Parameters:

Name	Type	Description	Default
`callback`	`Callable[[T], None]`	Receiver invoked once per emitted value.	required

Returns:

Type	Description
`Disposable`	Disposable that stops the stream and runs the producer teardown.

Example

sub = sensor.onPropertyChanged.subscribe(handle)

pipe ¶

pipe(*operators: OperatorFn) -> Observable[Any]

Chain operators into a new Observable, each one fed by the previous.

Parameters:

Name	Type	Description	Default
`operators`	`OperatorFn`	Operators applied left to right.	`()`

Returns:

Type	Description
`Observable[Any]`	Observable emitting the transformed values.

Example

sensor.onPropertyChanged.pipe(
    filter_op(lambda e: e["property"] == "detected"),
    map_op(lambda e: e["value"]),
).subscribe(handle)

asubscribe ¶

asubscribe(on_next: Callable[[T], Awaitable[Any]] | None = None, on_error: Callable[[Exception], Awaitable[Any]] | None = None) -> Disposable

Subscribe with coroutine handlers. Each value spawns a task running on_next.

A failing on_next disposes the subscription and hands the exception to on_error.

Parameters:

Name	Type	Description	Default
`on_next`	`Callable[[T], Awaitable[Any]] \| None`	Coroutine invoked per emitted value.	`None`
`on_error`	`Callable[[Exception], Awaitable[Any]] \| None`	Coroutine invoked once with the first exception raised.	`None`

Returns:

Type	Description
`Disposable`	Disposable that cancels the pending tasks and the subscription.

Example

sub = camera.onDetection.asubscribe(handle_detection)

Subject ¶

Subject()

Bases: Generic[T]

Multicast value source.

Calls to :meth:next are dispatched synchronously to every active subscriber. :meth:complete releases all subscribers and locks the Subject so further :meth:next calls become no-ops. :meth:subscribe returns a :class:Disposable for individual cleanup.

closed `property` ¶

closed: bool

Whether :meth:complete has already run.

next ¶

next(value: T) -> None

Dispatch a value to every current subscriber, synchronously. No-op once completed.

Parameters:

Name	Type	Description	Default
`value`	`T`	Value to multicast.	required

Example

subject.next("armed")

complete ¶

complete() -> None

Release every subscriber and lock the Subject. Later :meth:next calls are ignored.

subscribe(callback: Callable[[T], None]) -> Disposable

Register callback for every following value.

Parameters:

Name	Type	Description	Default
`callback`	`Callable[[T], None]`	Receiver invoked once per emitted value.	required

Returns:

Type	Description
`Disposable`	Disposable that unregisters the callback.

pipe ¶

pipe(*operators: OperatorFn) -> Observable[Any]

Chain operators onto this Subject's read-only view.

Parameters:

Name	Type	Description	Default
`operators`	`OperatorFn`	Operators applied left to right.	`()`

Returns:

Type	Description
`Observable[Any]`	Observable emitting the transformed values.

as_observable ¶

as_observable() -> Observable[T]

Return a read-only :class:Observable that mirrors this Subject without exposing :meth:next or :meth:complete.

Returns:

Type	Description
`Observable[T]`	Read-only view of this Subject.

Example

def on_event(self) -> Observable[Event]:
    return self._events.as_observable()

BehaviorSubject ¶

BehaviorSubject(initial_value: T)

Bases: Subject[T]

Subject seeded with an initial value that always remembers the latest emission.

New subscribers receive the current value immediately on :meth:subscribe and then all subsequent values. The current value is also accessible synchronously via :attr:value and :meth:get_value.

value `property` ¶

value: T

The most recently emitted value.

next ¶

next(value: T) -> None

Store value as the current one and dispatch it to every subscriber.

Parameters:

Name	Type	Description	Default
`value`	`T`	New current value.	required

get_value ¶

get_value() -> T

Read the current value without subscribing.

Returns:

Type	Description
`T`	The most recently emitted value.

subscribe(callback: Callable[[T], None]) -> Disposable

Register callback and invoke it immediately with the current value.

Parameters:

Name	Type	Description	Default
`callback`	`Callable[[T], None]`	Receiver invoked once per emitted value.	required

Returns:

Type	Description
`Disposable`	Disposable that unregisters the callback.

ReplaySubject ¶

ReplaySubject(buffer_size: int = 1)

Bases: Subject[T]

Subject that buffers the last buffer_size values, 1 by default.

New subscribers immediately receive every buffered value in order before continuing with live emissions.

next ¶

next(value: T) -> None

Append value to the buffer, dropping the oldest entry past buffer_size, then dispatch it.

Parameters:

Name	Type	Description	Default
`value`	`T`	Value to buffer and multicast.	required

subscribe(callback: Callable[[T], None]) -> Disposable

Replay the buffered values in order, then register callback for live emissions.

Parameters:

Name	Type	Description	Default
`callback`	`Callable[[T], None]`	Receiver invoked once per emitted value.	required

Returns:

Type	Description
`Disposable`	Disposable that unregisters the callback.

distinct_until_changed ¶

distinct_until_changed(comparator: _Comparator | None = None) -> OperatorFn

Emit a value only when it differs from the previous one. Uses == by default, or an optional custom comparator (e.g. for deep equality).

Parameters:

Name	Type	Description	Default
`comparator`	`_Comparator \| None`	Equality function; return True to suppress duplicates.	`None`

Returns:

Type	Description
`OperatorFn`	Operator that drops consecutive equal values.

Example

from camera_ui_sdk import distinct_until_changed

sensor.onPropertyChanged.pipe(
    distinct_until_changed(),
).subscribe(handle)

share(connector: Callable[[], Subject[Any]] | None = None) -> OperatorFn

Multicast a cold Observable through a Subject, sharing a single upstream subscription among all subscribers (reference-counted). Supply a custom connector (e.g. lambda: ReplaySubject(1)) to change buffering.

Parameters:

Name	Type	Description	Default
`connector`	`Callable[[], Subject[Any]] \| None`	Factory returning the multicast Subject to use.	`None`

Returns:

Type	Description
`OperatorFn`	Operator that multicasts the source.

Example

from camera_ui_sdk import ReplaySubject, share

events = source.pipe(share(lambda: ReplaySubject(1)))
events.subscribe(lambda v: print("a", v))
events.subscribe(lambda v: print("b", v))

filter_op ¶

filter_op(predicate: Callable[[Any], bool]) -> OperatorFn

Emit only the values for which predicate returns True.

Parameters:

Name	Type	Description	Default
`predicate`	`Callable[[Any], bool]`	Predicate evaluated for each upstream value.	required

Returns:

Type	Description
`OperatorFn`	Operator that drops values failing the predicate.

Example

from camera_ui_sdk import filter_op

sensor.onPropertyChanged.pipe(
    filter_op(lambda e: e["property"] == "detected"),
).subscribe(handle)

map_op ¶

map_op(transform: Callable[[Any], Any]) -> OperatorFn

Apply transform to each emitted value and emit the result.

Parameters:

Name	Type	Description	Default
`transform`	`Callable[[Any], Any]`	Projection invoked for each upstream value.	required

Returns:

Type	Description
`OperatorFn`	Operator that maps every value into a new shape.

Example

from camera_ui_sdk import map_op

sensor.onPropertyChanged.pipe(
    map_op(lambda e: e["value"]),
).subscribe(handle)

pairwise ¶

pairwise() -> OperatorFn

Emit (previous, current) pairs for every value after the first.

Returns:

Type	Description
`OperatorFn`	Operator that yields adjacent value pairs.

Example

from camera_ui_sdk import pairwise

source.pipe(pairwise()).subscribe(lambda pair: print(pair))

merge_map ¶

merge_map(project: Callable[[Any, int], list[Any]]) -> OperatorFn

Project each source value to a list and flatten the results into the output stream.

Parameters:

Name	Type	Description	Default
`project`	`Callable[[Any, int], list[Any]]`	Function returning a list of values for each input (receives the value and a zero-based index).	required

Returns:

Type	Description
`OperatorFn`	Operator that flattens projected lists into the output stream.

Example

from camera_ui_sdk import merge_map

source.pipe(merge_map(lambda v, i: [v, v * 2])).subscribe(handle)

first_value_from `async` ¶

first_value_from(observable: Observable[T] | Subject[T]) -> T

Subscribe to the source and return its first emitted value as a coroutine, then dispose the subscription.

Raises RuntimeError if the source completes before emitting (Subject, BehaviorSubject, ReplaySubject). A bare :class:Observable has no completion signal, so the coroutine stays pending until it emits; guard such calls with :func:asyncio.wait_for or another timeout.

Parameters:

Name	Type	Description	Default
`observable`	`Observable[T] \| Subject[T]`	Source observable or subject to read once.	required

Returns:

Type	Description
`T`	The first value emitted by the source.

Raises:

Type	Description
`RuntimeError`	If the source completes without emitting a value.

Example

from camera_ui_sdk import first_value_from

value = await first_value_from(behavior_subject)

Observable¶

camera_ui_sdk.observable ¶

OperatorFn module-attribute ¶

Disposable ¶

closed property ¶

dispose ¶

unsubscribe ¶

Observable ¶

subscribe ¶

pipe ¶

asubscribe ¶

Subject ¶

closed property ¶

next ¶

complete ¶

subscribe ¶

pipe ¶

as_observable ¶

BehaviorSubject ¶

value property ¶

next ¶

get_value ¶

subscribe ¶

ReplaySubject ¶

next ¶

subscribe ¶

distinct_until_changed ¶

share ¶

filter_op ¶

map_op ¶

pairwise ¶

merge_map ¶

first_value_from async ¶

OperatorFn `module-attribute` ¶

closed `property` ¶

closed `property` ¶

value `property` ¶

first_value_from `async` ¶