`indi_nexus.client`¶

A reconnecting asyncio TCP client to indiserver with a typed property cache, subscriptions, and send helpers.

IndiClient¶

indi_nexus.client.client ¶

IndiClient: a reconnecting async client for indiserver.

The client is a TCP peer of the C indiserver (default port 7624). It keeps a typed :class:~indi_nexus.client.store.PropertyStore up to date from the inbound stream, lets application code watch for changes and wait on conditions, and sends updates - always as M1 typed models, never raw XML.

Concurrency is plain :mod:asyncio: a background connection loop reconnects with a fixed delay, and per connection a reader task folds inbound messages into the store (dispatching to subscribers) while a writer task drains an outbox queue. The transport is injectable (a connect coroutine returning read/write/ close callables) so tests drive the client over in-memory streams; the default opens a real TCP connection via :func:indi_nexus.transport.open_tcp. The close callable is invoked whenever a connection ends - EOF, error, or :meth:IndiClient.aclose - so the OS socket never lingers between reconnects.

IndiClient ¶

IndiClient(host: str = 'localhost', port: int = 7624, *, connect_timeout: float = 10.0, reconnect_delay: float = 2.0, connect: Connect | None = None)

A reconnecting client that mirrors indiserver state into a cache.

Parameters:

Name	Type	Description	Default
`host`	`str`	The `indiserver` host.	`'localhost'`
`port`	`int`	The `indiserver` TCP port (7624 by default).	`7624`
`connect_timeout`	`float`	Seconds to wait for each connection attempt.	`10.0`
`reconnect_delay`	`float`	Seconds to wait between a lost connection and the next attempt.	`2.0`
`connect`	`Connect`	Injectable connection factory returning `(read, write, close)` callables; used by tests. Defaults to a real TCP connection to `host`/`port`.	`None`

Source code in src/indi_nexus/client/client.py

def __init__(
    self,
    host: str = "localhost",
    port: int = 7624,
    *,
    connect_timeout: float = 10.0,
    reconnect_delay: float = 2.0,
    connect: Connect | None = None,
) -> None:
    self._host = host
    self._port = port
    self._connect_timeout = connect_timeout
    self._reconnect_delay = reconnect_delay
    self._connect = connect or self._default_connect

    self._store = PropertyStore()
    self._outbox: asyncio.Queue[IndiMessage] = asyncio.Queue()
    self._message_subs: dict[int, MessageCallback] = {}
    self._conn_subs: dict[int, ConnectionCallback] = {}
    self._sub_ids = 0

    # Replayed on every (re)connect so the server re-sends what we care about.
    self._blob_policies: dict[tuple[str, str | None], EnableBLOB] = {}

    self._loop_task: asyncio.Task[None] | None = None
    self._closing = False
    self._connected = False
    self._ready = asyncio.Event()

connected `property` ¶

connected: bool

Whether the client currently has a live connection.

store `property` ¶

store: PropertyStore

The underlying property cache.

start `async` ¶

start() -> None

Start the background connection loop and wait for the first connect.

Source code in src/indi_nexus/client/client.py

async def start(self) -> None:
    """Start the background connection loop and wait for the first connect."""
    if self._loop_task is None:
        self._loop_task = asyncio.create_task(self._connection_loop())
    await self._ready.wait()

aclose `async` ¶

aclose() -> None

Stop the connection loop and drop the connection.

Source code in src/indi_nexus/client/client.py

async def aclose(self) -> None:
    """Stop the connection loop and drop the connection."""
    self._closing = True
    if self._loop_task is not None:
        self._loop_task.cancel()
        await asyncio.gather(self._loop_task, return_exceptions=True)
        self._loop_task = None

aenter `async` ¶

__aenter__() -> IndiClient

Start the client and return it once initially connected.

Source code in src/indi_nexus/client/client.py

async def __aenter__(self) -> IndiClient:
    """Start the client and return it once initially connected."""
    await self.start()
    return self

aexit `async` ¶

__aexit__(*exc: object) -> None

Close the client on context exit.

Source code in src/indi_nexus/client/client.py

async def __aexit__(self, *exc: object) -> None:
    """Close the client on context exit."""
    await self.aclose()

get ¶

get(device: str, name: str) -> Vector | None

Return a cached vector, or None if it is not present.

Parameters:

Name	Type	Description	Default
`device`	`str`	The device name.	required
`name`	`str`	The property name.	required

Returns:

Name	Type	Description
`vector`	`Vector or None`	The cached vector, or `None`.

Source code in src/indi_nexus/client/client.py

def get(self, device: str, name: str) -> Vector | None:
    """Return a cached vector, or `None` if it is not present.

    Parameters
    ----------
    device : str
        The device name.
    name : str
        The property name.

    Returns
    -------
    vector : Vector or None
        The cached vector, or `None`.
    """
    return self._store.get(device, name)

getitem ¶

__getitem__(device: str) -> Any

Return the cached properties of one device.

Source code in src/indi_nexus/client/client.py

def __getitem__(self, device: str) -> Any:
    """Return the cached properties of one device."""
    return self._store[device]

subscribe(callback: Subscriber, *, device: str | None = None, name: str | None = None) -> Callable[[], None]

Register a property-event callback (see :meth:PropertyStore.subscribe).

Parameters:

Name	Type	Description	Default
`callback`	`Subscriber`	Called with each matching :class:`PropertyEvent`; may be sync or async.	required
`device`	`str`	Restrict to one device; `None` matches every device.	`None`
`name`	`str`	Restrict to one property; `None` matches every property.	`None`

Returns:

Name	Type	Description
`unsubscribe`	`Callable`	Call with no arguments to remove the subscription.

Source code in src/indi_nexus/client/client.py

def subscribe(
    self, callback: Subscriber, *, device: str | None = None, name: str | None = None
) -> Callable[[], None]:
    """Register a property-event callback (see :meth:`PropertyStore.subscribe`).

    Parameters
    ----------
    callback : Subscriber
        Called with each matching :class:`PropertyEvent`; may be sync or async.
    device : str, optional
        Restrict to one device; `None` matches every device.
    name : str, optional
        Restrict to one property; `None` matches every property.

    Returns
    -------
    unsubscribe : Callable
        Call with no arguments to remove the subscription.
    """
    return self._store.subscribe(callback, device=device, name=name)

on_message ¶

on_message(callback: MessageCallback) -> Callable[[], None]

Register a callback for inbound message notifications.

Parameters:

Name	Type	Description	Default
`callback`	`Callable`	Called with each inbound :class:`Message`; may be sync or async.	required

Returns:

Name	Type	Description
`unsubscribe`	`Callable`	Call with no arguments to remove the subscription.

Source code in src/indi_nexus/client/client.py

def on_message(self, callback: MessageCallback) -> Callable[[], None]:
    """Register a callback for inbound ``message`` notifications.

    Parameters
    ----------
    callback : Callable
        Called with each inbound :class:`Message`; may be sync or async.

    Returns
    -------
    unsubscribe : Callable
        Call with no arguments to remove the subscription.
    """
    return self._register(self._message_subs, callback)

on_connection ¶

on_connection(callback: ConnectionCallback) -> Callable[[], None]

Register a callback for connect/disconnect transitions.

Parameters:

Name	Type	Description	Default
`callback`	`Callable`	Called with `True` on connect and `False` on disconnect; may be async.	required

Returns:

Name	Type	Description
`unsubscribe`	`Callable`	Call with no arguments to remove the subscription.

Source code in src/indi_nexus/client/client.py

def on_connection(self, callback: ConnectionCallback) -> Callable[[], None]:
    """Register a callback for connect/disconnect transitions.

    Parameters
    ----------
    callback : Callable
        Called with `True` on connect and `False` on disconnect; may be async.

    Returns
    -------
    unsubscribe : Callable
        Call with no arguments to remove the subscription.
    """
    return self._register(self._conn_subs, callback)

wait_for `async` ¶

wait_for(device: str, name: str, predicate: Predicate | None = None, *, timeout: float | None = None) -> Vector

Wait until a property exists (and satisfies predicate).

Resolves immediately if the cached property already matches.

Parameters:

Name	Type	Description	Default
`device`	`str`	The device name.	required
`name`	`str`	The property name.	required
`predicate`	`Predicate`	Called with the vector; the wait resolves when it returns `True`. Defaults to "exists".	`None`
`timeout`	`float`	Seconds to wait before raising `TimeoutError`.	`None`

Returns:

Name	Type	Description
`vector`	`Vector`	The matching vector.

Raises:

Type	Description
`TimeoutError`	Raised if the timeout elapses first.

Source code in src/indi_nexus/client/client.py

async def wait_for(
    self,
    device: str,
    name: str,
    predicate: Predicate | None = None,
    *,
    timeout: float | None = None,  # noqa: ASYNC109 - public API mirrors asyncio.wait_for
) -> Vector:
    """Wait until a property exists (and satisfies ``predicate``).

    Resolves immediately if the cached property already matches.

    Parameters
    ----------
    device : str
        The device name.
    name : str
        The property name.
    predicate : Predicate, optional
        Called with the vector; the wait resolves when it returns `True`.
        Defaults to "exists".
    timeout : float, optional
        Seconds to wait before raising ``TimeoutError``.

    Returns
    -------
    vector : Vector
        The matching vector.

    Raises
    ------
    TimeoutError
        Raised if the timeout elapses first.
    """
    current = self._store.get(device, name)
    if current is not None and (predicate is None or predicate(current)):
        return current

    loop = asyncio.get_running_loop()
    future: asyncio.Future[Vector] = loop.create_future()

    def on_event(event: PropertyEvent) -> None:
        """Resolve the future when a matching vector arrives."""
        vec = event.vector
        if vec is not None and not future.done() and (predicate is None or predicate(vec)):
            future.set_result(vec)

    unsubscribe = self._store.subscribe(on_event, device=device, name=name)
    try:
        if timeout is not None:
            async with asyncio.timeout(timeout):
                return await future
        return await future
    finally:
        unsubscribe()

send `async` ¶

send(msg: IndiMessage) -> None

Queue an arbitrary message to send upstream.

The typed helpers (:meth:set_number, :meth:get_properties, ...) cover the common cases; this forwards any already-built message - used by the web bridge to relay a browser-authored new*/getProperties/ enableBLOB frame verbatim.

Parameters:

Name	Type	Description	Default
`msg`	`IndiMessage`	The message to send.	required

Source code in src/indi_nexus/client/client.py

async def send(self, msg: IndiMessage) -> None:
    """Queue an arbitrary message to send upstream.

    The typed helpers (:meth:`set_number`, :meth:`get_properties`, ...) cover
    the common cases; this forwards any already-built message - used by the
    web bridge to relay a browser-authored ``new*``/``getProperties``/
    ``enableBLOB`` frame verbatim.

    Parameters
    ----------
    msg : IndiMessage
        The message to send.
    """
    self._outbox.put_nowait(msg)

get_properties `async` ¶

get_properties(device: str | None = None, name: str | None = None) -> None

Ask the server to (re-)send property definitions.

Parameters:

Name	Type	Description	Default
`device`	`str`	Restrict to one device; `None` requests every device.	`None`
`name`	`str`	Restrict to one property; `None` requests every property.	`None`

Source code in src/indi_nexus/client/client.py

async def get_properties(self, device: str | None = None, name: str | None = None) -> None:
    """Ask the server to (re-)send property definitions.

    Parameters
    ----------
    device : str, optional
        Restrict to one device; `None` requests every device.
    name : str, optional
        Restrict to one property; `None` requests every property.
    """
    await self._send(GetProperties(device=device, name=name))

enable_blob `async` ¶

enable_blob(device: str, name: str | None = None, policy: BLOBPolicy = BLOBPolicy.ALSO) -> None

Set the BLOB delivery policy for a device (or one property).

The request is remembered and replayed on every reconnect.

Parameters:

Name	Type	Description	Default
`device`	`str`	The device to set the policy for.	required
`name`	`str`	Restrict to one property; `None` applies to the whole device.	`None`
`policy`	`BLOBPolicy`	Whether BLOBs are never sent, sent alongside other updates, or sent exclusively.	`ALSO`

Source code in src/indi_nexus/client/client.py

async def enable_blob(
    self, device: str, name: str | None = None, policy: BLOBPolicy = BLOBPolicy.ALSO
) -> None:
    """Set the BLOB delivery policy for a device (or one property).

    The request is remembered and replayed on every reconnect.

    Parameters
    ----------
    device : str
        The device to set the policy for.
    name : str, optional
        Restrict to one property; `None` applies to the whole device.
    policy : BLOBPolicy, optional
        Whether BLOBs are never sent, sent alongside other updates, or sent
        exclusively.
    """
    msg = EnableBLOB(device=device, name=name, policy=policy)
    self._blob_policies[(device, name)] = msg
    await self._send(msg)

set_number `async` ¶

set_number(device: str, name: str, values: dict[str, float]) -> None

Send new number values for a property.

Parameters:

Name	Type	Description	Default
`device`	`str`	The device name.	required
`name`	`str`	The property name.	required
`values`	`dict`	Mapping of element name to numeric value.	required

Source code in src/indi_nexus/client/client.py

async def set_number(self, device: str, name: str, values: dict[str, float]) -> None:
    """Send new number values for a property.

    Parameters
    ----------
    device : str
        The device name.
    name : str
        The property name.
    values : dict
        Mapping of element name to numeric value.
    """
    elements = [Number(name=k, value=v) for k, v in values.items()]
    vector = NumberVector(device=device, name=name, elements=elements)
    await self._send(NewVector(vector=vector))

set_text `async` ¶

set_text(device: str, name: str, values: dict[str, str]) -> None

Send new text values for a property.

Parameters:

Name	Type	Description	Default
`device`	`str`	The device name.	required
`name`	`str`	The property name.	required
`values`	`dict`	Mapping of element name to string value.	required

Source code in src/indi_nexus/client/client.py

async def set_text(self, device: str, name: str, values: dict[str, str]) -> None:
    """Send new text values for a property.

    Parameters
    ----------
    device : str
        The device name.
    name : str
        The property name.
    values : dict
        Mapping of element name to string value.
    """
    elements = [Text(name=k, value=v) for k, v in values.items()]
    vector = TextVector(device=device, name=name, elements=elements)
    await self._send(NewVector(vector=vector))

set_switch `async` ¶

set_switch(device: str, name: str, values: dict[str, Any]) -> None

Send new switch states for a property.

Parameters:

Name	Type	Description	Default
`device`	`str`	The device name.	required
`name`	`str`	The property name.	required
`values`	`dict`	Mapping of element name to state (`ISState`, `bool`, or `"On"` / `"Off"`).	required

Source code in src/indi_nexus/client/client.py

async def set_switch(self, device: str, name: str, values: dict[str, Any]) -> None:
    """Send new switch states for a property.

    Parameters
    ----------
    device : str
        The device name.
    name : str
        The property name.
    values : dict
        Mapping of element name to state (``ISState``, ``bool``, or ``"On"`` /
        ``"Off"``).
    """
    elements = [Switch(name=k, value=_coerce_switch(v)) for k, v in values.items()]
    vector = SwitchVector(device=device, name=name, elements=elements)
    await self._send(NewVector(vector=vector))

set_blob `async` ¶

set_blob(device: str, name: str, values: dict[str, bytes]) -> None

Send new BLOB payloads for a property.

Parameters:

Name	Type	Description	Default
`device`	`str`	The device name.	required
`name`	`str`	The property name.	required
`values`	`dict`	Mapping of element name to raw `bytes` payload.	required

Source code in src/indi_nexus/client/client.py

async def set_blob(self, device: str, name: str, values: dict[str, bytes]) -> None:
    """Send new BLOB payloads for a property.

    Parameters
    ----------
    device : str
        The device name.
    name : str
        The property name.
    values : dict
        Mapping of element name to raw ``bytes`` payload.
    """
    elements = [BLOB(name=k, data=v, size=len(v)) for k, v in values.items()]
    await self._send(NewVector(vector=BLOBVector(device=device, name=name, elements=elements)))

run ¶

run() -> None

Connect and process the stream until interrupted (blocking).

A convenience entrypoint for scripts and monitors: register subscriptions first, then call this. Returns on KeyboardInterrupt.

Source code in src/indi_nexus/client/client.py

def run(self) -> None:
    """Connect and process the stream until interrupted (blocking).

    A convenience entrypoint for scripts and monitors: register subscriptions
    first, then call this. Returns on ``KeyboardInterrupt``.
    """

    async def _serve() -> None:
        """Start the client and block until cancelled."""
        await self.start()
        try:
            await asyncio.Event().wait()
        finally:
            await self.aclose()

    with contextlib.suppress(KeyboardInterrupt):
        asyncio.run(_serve())

PropertyStore¶

indi_nexus.client.store ¶

PropertyStore: the client's typed cache of INDI properties.

The store is the single source of cached truth for a client. It folds inbound messages into a device -> name -> vector cache following standard INDI semantics (def defines, set merges values onto the definition, del removes), and it holds the subscription registry.

It is deliberately free of any socket or asyncio behaviour: :meth:apply updates the cache and returns a :class:PropertyEvent, and :meth:matching returns the callbacks interested in that event. The client performs the actual (possibly asynchronous) dispatch, so the store stays pure and trivially testable.

PropertyEvent `dataclass` ¶

PropertyEvent(type: EventType, device: str, name: str | None, vector: Vector | None)

A change the store applied to its cache.

Attributes:

Name	Type	Description
`type`	`str`	`"def"`, `"set"`, or `"del"`.
`device`	`str`	The device the change applies to.
`name`	`str or None`	The property name, or `None` for a whole-device `del`.
`vector`	`Vector or None`	The affected (post-merge) vector, or `None` for a `del`.

PropertyStore ¶

PropertyStore()

A cache of INDI property vectors plus a subscription registry.

Create an empty store with no cached properties or subscribers.

Source code in src/indi_nexus/client/store.py

def __init__(self) -> None:
    """Create an empty store with no cached properties or subscribers."""
    self._by_device: dict[str, dict[str, Vector]] = {}
    self._subs: dict[int, tuple[Subscriber, str | None, str | None]] = {}
    self._ids = count()

get ¶

get(device: str, name: str) -> Vector | None

Return a cached vector, or None if it is not present.

Parameters:

Name	Type	Description	Default
`device`	`str`	The device name.	required
`name`	`str`	The property name.	required

Returns:

Name	Type	Description
`vector`	`Vector or None`	The cached vector, or `None`.

Source code in src/indi_nexus/client/store.py

def get(self, device: str, name: str) -> Vector | None:
    """Return a cached vector, or `None` if it is not present.

    Parameters
    ----------
    device : str
        The device name.
    name : str
        The property name.

    Returns
    -------
    vector : Vector or None
        The cached vector, or `None`.
    """
    return self._by_device.get(device, {}).get(name)

device ¶

device(name: str) -> Mapping[str, Vector]

Return a read-only mapping of one device's properties.

Parameters:

Name	Type	Description	Default
`name`	`str`	The device name.	required

Returns:

Name	Type	Description
`properties`	`Mapping`	The device's `property-name -> vector` mapping (empty if unknown).

Source code in src/indi_nexus/client/store.py

def device(self, name: str) -> Mapping[str, Vector]:
    """Return a read-only mapping of one device's properties.

    Parameters
    ----------
    name : str
        The device name.

    Returns
    -------
    properties : Mapping
        The device's ``property-name -> vector`` mapping (empty if unknown).
    """
    return dict(self._by_device.get(name, {}))

devices ¶

devices() -> list[str]

Return the names of all known devices.

Source code in src/indi_nexus/client/store.py

def devices(self) -> list[str]:
    """Return the names of all known devices."""
    return list(self._by_device)

getitem ¶

__getitem__(device: str) -> Mapping[str, Vector]

Return one device's properties (see :meth:device).

Source code in src/indi_nexus/client/store.py

def __getitem__(self, device: str) -> Mapping[str, Vector]:
    """Return one device's properties (see :meth:`device`)."""
    return self.device(device)

contains ¶

__contains__(device: str) -> bool

Return whether any property is cached for device.

Source code in src/indi_nexus/client/store.py

def __contains__(self, device: str) -> bool:
    """Return whether any property is cached for ``device``."""
    return device in self._by_device

iter ¶

__iter__() -> Iterator[str]

Iterate over the known device names.

Source code in src/indi_nexus/client/store.py

def __iter__(self) -> Iterator[str]:
    """Iterate over the known device names."""
    return iter(self._by_device)

apply ¶

apply(msg: IndiMessage) -> PropertyEvent | None

Fold one inbound message into the cache.

Parameters:

Name	Type	Description	Default
`msg`	`IndiMessage`	The parsed inbound message.	required

Returns:

Name	Type	Description
`event`	`PropertyEvent or None`	The change applied, or `None` if the message did not change the cache (an unknown `set`, or a non-property message).

Source code in src/indi_nexus/client/store.py

def apply(self, msg: IndiMessage) -> PropertyEvent | None:
    """Fold one inbound message into the cache.

    Parameters
    ----------
    msg : IndiMessage
        The parsed inbound message.

    Returns
    -------
    event : PropertyEvent or None
        The change applied, or `None` if the message did not change the cache
        (an unknown ``set``, or a non-property message).
    """
    if isinstance(msg, DefVector):
        vec = msg.vector
        self._by_device.setdefault(vec.device, {})[vec.name] = vec
        return PropertyEvent("def", vec.device, vec.name, vec)
    if isinstance(msg, SetVector):
        cur = self.get(msg.vector.device, msg.vector.name)
        if cur is None:
            return None
        _merge(cur, msg.vector)
        return PropertyEvent("set", cur.device, cur.name, cur)
    if isinstance(msg, DelProperty):
        return self._delete(msg)
    return None

subscribe(callback: Subscriber, *, device: str | None = None, name: str | None = None) -> Callable[[], None]

Register a callback for matching property events.

Parameters:

Name	Type	Description	Default
`callback`	`Subscriber`	Called with each matching :class:`PropertyEvent`. May be sync or async; the client awaits coroutine results.	required
`device`	`str`	Restrict to one device; `None` matches every device.	`None`
`name`	`str`	Restrict to one property name; `None` matches every property.	`None`

Returns:

Name	Type	Description
`unsubscribe`	`Callable`	Call with no arguments to remove the subscription.

Source code in src/indi_nexus/client/store.py

def subscribe(
    self, callback: Subscriber, *, device: str | None = None, name: str | None = None
) -> Callable[[], None]:
    """Register a callback for matching property events.

    Parameters
    ----------
    callback : Subscriber
        Called with each matching :class:`PropertyEvent`. May be sync or
        async; the client awaits coroutine results.
    device : str, optional
        Restrict to one device; `None` matches every device.
    name : str, optional
        Restrict to one property name; `None` matches every property.

    Returns
    -------
    unsubscribe : Callable
        Call with no arguments to remove the subscription.
    """
    token = next(self._ids)
    self._subs[token] = (callback, device, name)

    def unsubscribe() -> None:
        """Remove this subscription."""
        self._subs.pop(token, None)

    return unsubscribe

matching ¶

matching(event: PropertyEvent) -> list[Subscriber]

Return the callbacks subscribed to a given event.

Parameters:

Name	Type	Description	Default
`event`	`PropertyEvent`	The event to match against the registry.	required

Returns:

Name	Type	Description
`callbacks`	`list of Subscriber`	The callbacks whose device/name filters match, in registration order.

Source code in src/indi_nexus/client/store.py

def matching(self, event: PropertyEvent) -> list[Subscriber]:
    """Return the callbacks subscribed to a given event.

    Parameters
    ----------
    event : PropertyEvent
        The event to match against the registry.

    Returns
    -------
    callbacks : list of Subscriber
        The callbacks whose device/name filters match, in registration order.
    """
    out: list[Subscriber] = []
    for callback, device, name in self._subs.values():
        if device is not None and device != event.device:
            continue
        if name is not None and name != event.name:
            continue
        out.append(callback)
    return out

indi_nexus.client¶

IndiClient¶

indi_nexus.client.client ¶

IndiClient ¶

connected property ¶

store property ¶

start async ¶

aclose async ¶

__aenter__ async ¶

__aexit__ async ¶

get ¶

__getitem__ ¶

subscribe ¶

on_message ¶

on_connection ¶

wait_for async ¶

send async ¶

get_properties async ¶

enable_blob async ¶

set_number async ¶

set_text async ¶

set_switch async ¶

set_blob async ¶

run ¶

PropertyStore¶

indi_nexus.client.store ¶

PropertyEvent dataclass ¶

PropertyStore ¶

get ¶

device ¶

devices ¶

__getitem__ ¶

__contains__ ¶

__iter__ ¶

apply ¶

subscribe ¶

matching ¶

`indi_nexus.client`¶

connected `property` ¶

store `property` ¶

start `async` ¶

aclose `async` ¶

aenter `async` ¶

aexit `async` ¶

getitem ¶

wait_for `async` ¶

send `async` ¶

get_properties `async` ¶

enable_blob `async` ¶

set_number `async` ¶

set_text `async` ¶

set_switch `async` ¶

set_blob `async` ¶

PropertyEvent `dataclass` ¶

getitem ¶

contains ¶

iter ¶