Observable

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

camera_ui_sdk.observable

Lightweight reactive primitives for camera.ui.

Provides cold Observables, multicast Subjects (Subject, BehaviorSubject, ReplaySubject) and a small set of composable operators for building property-change notifications and event streams throughout the SDK.

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.

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

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

Start the producer for this subscriber and route emitted values to callback. Returns a :class:Disposable for stopping the stream.

asubscribe

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

Subscribe asynchronously to the observable sequence.

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.

as_observable

as_observable() -> Observable[T]

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

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.

ReplaySubject

ReplaySubject(buffer_size: int = 1)

Bases: Subject[T]

Subject that buffers up to the last buffer_size values (configurable, defaults to 1).

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

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

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 a 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)