indikit.client

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

IndiClient

indikit.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:~indikit.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:indikit.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.

Sending is deliberately not buffered across connections: a send with no live connection raises :class:~indikit.exceptions.NotConnectedError and the outbox is emptied whenever a connection ends, so nothing a caller issued while indiserver was away can be delivered to an instrument minutes later. See :meth:IndiClient.send.

:attr:IndiClient.stats is the operational read of all of that - how long this connection has been up, how many reconnects it took to get here, and what the parser has made of the peer - and it is what /health reports.

ClientStats dataclass

ClientStats(connected: bool, uptime_seconds: float | None, reconnects: int, last_message_age_seconds: float | None, dropped: int, resets: int, bytes_since_last_message: int, dropped_total: int, resets_total: int)

A point-in-time read of one client's upstream link and its parser.

Taken as a snapshot rather than exposed as live attributes, so a caller that reports several of these fields - /health does - reports them all as of one instant.

Attributes:

Name	Type	Description
connected	bool	Whether there is a live connection right now.
uptime_seconds	float or None	How long the current connection has been up, and None while there is none. It is deliberately not the process's uptime: a container runtime already reports that, and everything else here is about the upstream link. It resets on every reconnect.
reconnects	int	How many times a connection has been successfully re-established since the client started; the first connection is not a reconnect. Failed attempts are not counted, so a rising number means the link is genuinely flapping, while a bridge that has never reached indiserver at all reports 0 with connected false - which already tells that story.
last_message_age_seconds	float or None	Seconds since an INDI message was last parsed, and None if none ever has been. It measures parsed messages rather than received bytes, because :attr:bytes_since_last_message already answers the byte question and the two must stay distinct: a peer dribbling malformed bytes must not read as healthy here. It is not reset by a reconnect - it is the age of the last thing this client understood, whichever connection carried it.
dropped	int	Top-level elements this connection's parser discarded because a value would not parse. Before the first connection, and while the client has never had one, this is 0.
resets	int	Times this connection's parser had to reopen its synthetic document. A framing signal, not a loss count. 0 before the first connection.
bytes_since_last_message	int	Bytes fed to this connection's parser since a message last came out of it. 0 before the first connection.
dropped_total	int	:attr:dropped summed over every connection since the client started, including the current one. This is the field that answers "has this ever happened", which the per-connection counters discard on every reconnect.
resets_total	int	:attr:resets summed over every connection since the client started, including the current one.

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/indikit/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(maxsize=_OUTBOX_MAXSIZE)
    self._message_subs: dict[int, MessageCallback] = {}
    self._conn_subs: dict[int, ConnectionCallback] = {}
    self._sub_ids = 0
    # Every future wait_for() is parked on, and what it is waiting for, so
    # aclose() can tell them the answer is never coming.
    self._waiters: dict[asyncio.Future[Vector], str] = {}

    # 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()

    # This connection's parser, `None` until the first one is established.
    # Reassigned per connection by _new_parser(); never reused across one.
    self._parser: XMLStreamParser | None = None
    # Every *earlier* parser's counters, folded in as each new one is made.
    # `self._parser`'s own counters are added at read time (see `stats`), so
    # the totals include the connection that is happening right now and stay
    # right after aclose(), when there is no next fold.
    self._dropped_total = 0
    self._resets_total = 0
    # Connections established, so reconnects = this - 1. Counting
    # establishments rather than reconnects keeps the increment
    # unconditional at the one site a connection comes up.
    self._connections = 0
    self._connected_at: float | None = None
    self._last_message_at: float | None = None

connected property

connected: bool

Whether the client currently has a live connection.

stats property

stats: ClientStats

A snapshot of the upstream link and this connection's parser.

Cheap: it reads counters, so an endpoint may call it per request.

Returns:

Name	Type	Description
stats	ClientStats	The current statistics; see that class for what each field means and what it reports while disconnected.

store property

store: PropertyStore

The underlying property cache.

start async

start(*, wait: bool = True) -> None

Start the background connection loop.

Parameters:

Name	Type	Description	Default
wait	bool	Whether to block until the first connection succeeds. Scripts and monitors want that. A long-running server that must stay responsive while indiserver is down passes False and watches :meth:on_connection instead; the loop keeps retrying either way.	True

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

async def start(self, *, wait: bool = True) -> None:
    """Start the background connection loop.

    Parameters
    ----------
    wait : bool, optional
        Whether to block until the first connection succeeds. Scripts and
        monitors want that. A long-running server that must stay responsive
        while ``indiserver`` is down passes `False` and watches
        :meth:`on_connection` instead; the loop keeps retrying either way.
    """
    if self._loop_task is None:
        self._loop_task = asyncio.create_task(self._connection_loop())
    if wait:
        await self._ready.wait()

aclose async

aclose() -> None

Stop the connection loop, drop the connection, and fail the waiters.

Every :meth:wait_for still parked on its future is resolved with :class:~indikit.exceptions.NotConnectedError, because nothing will ever read the socket again: without that, a wait with no timeout hangs for good and a wait with one sits out its full timeout to learn what the client already knows.

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

async def aclose(self) -> None:
    """Stop the connection loop, drop the connection, and fail the waiters.

    Every :meth:`wait_for` still parked on its future is resolved with
    :class:`~indikit.exceptions.NotConnectedError`, because nothing will
    ever read the socket again: without that, a wait with no timeout hangs
    for good and a wait with one sits out its full timeout to learn what the
    client already knows.
    """
    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
    self._fail_waiters()

__aenter__ async

__aenter__() -> IndiClient

Start the client and return it once initially connected.

Source code in src/indikit/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/indikit/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/indikit/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/indikit/client/client.py

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

subscribe

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/indikit/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/indikit/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/indikit/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.

What comes back is a snapshot, detached from the cache: the vector as it was at the instant the predicate held. The cached vector is mutated in place by every later set, and a whole TCP chunk's worth of messages is folded in before the reader yields, so a property that goes Busy, Ok, Busy inside one chunk would satisfy a state == OK wait and then read back Busy to the coroutine that was waiting on it. Read the live vector through :meth:get when that is what you want.

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	A detached copy of the matching vector, as it was when it matched.

Raises:

Type	Description
TimeoutError	Raised if the timeout elapses first.
NotConnectedError	Raised if :meth:aclose is called while the wait is still parked.

Source code in src/indikit/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.

    What comes back is a **snapshot**, detached from the cache: the vector
    as it was at the instant the predicate held. The cached vector is
    mutated in place by every later ``set``, and a whole TCP chunk's worth
    of messages is folded in before the reader yields, so a property that
    goes ``Busy``, ``Ok``, ``Busy`` inside one chunk would satisfy a
    ``state == OK`` wait and then read back ``Busy`` to the coroutine that
    was waiting on it. Read the live vector through :meth:`get` when that is
    what you want.

    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
        A detached copy of the matching vector, as it was when it matched.

    Raises
    ------
    TimeoutError
        Raised if the timeout elapses first.
    NotConnectedError
        Raised if :meth:`aclose` is called while the wait is still parked.
    """
    current = self._store.get(device, name)
    if current is not None and (predicate is None or predicate(current)):
        return current.detached()

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

    def on_event(event: PropertyEvent) -> None:
        """Resolve the future with a copy of the vector that matched."""
        vec = event.vector
        if vec is not None and not future.done() and (predicate is None or predicate(vec)):
            # Copy here, not at the await: the predicate was true of this
            # object one line ago and may not be by the time the waiting
            # coroutine is scheduled.
            future.set_result(vec.detached())

    unsubscribe = self._store.subscribe(on_event, device=device, name=name)
    self._waiters[future] = f"{device}.{name}"
    try:
        if timeout is not None:
            async with asyncio.timeout(timeout):
                return await future
        return await future
    finally:
        unsubscribe()
        self._waiters.pop(future, None)

send async

send(msg: IndiMessage) -> None

Hand one message to the live connection's writer.

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.

A send with no connection fails; it is never held. This is instrument control: a command queued while indiserver is down would be delivered whenever the hub came back, minutes or hours later, to hardware whose state has nothing to do with the one the caller was reasoning about. Every send routes through here, so every one of them either reaches a live connection or raises. Callers that want to wait for the link instead can watch :meth:on_connection.

Parameters:

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

Raises:

Type	Description
NotConnectedError	Raised if there is no live connection to indiserver. Also a ConnectionError.
SendQueueFull	Raised if the outbox is full because the connection has stopped draining it. Also a RuntimeError.

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

async def send(self, msg: IndiMessage) -> None:
    """Hand one message to the live connection's writer.

    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.

    **A send with no connection fails; it is never held.** This is
    instrument control: a command queued while ``indiserver`` is down would
    be delivered whenever the hub came back, minutes or hours later, to
    hardware whose state has nothing to do with the one the caller was
    reasoning about. Every send routes through here, so every one of them
    either reaches a live connection or raises. Callers that want to wait
    for the link instead can watch :meth:`on_connection`.

    Parameters
    ----------
    msg : IndiMessage
        The message to send.

    Raises
    ------
    NotConnectedError
        Raised if there is no live connection to ``indiserver``. Also a
        ConnectionError.
    SendQueueFull
        Raised if the outbox is full because the connection has stopped
        draining it. Also a RuntimeError.
    """
    if not self._connected:
        raise NotConnectedError(
            f"not connected to {self._host}:{self._port}; {type(msg).__name__} was not sent"
        )
    try:
        self._outbox.put_nowait(msg)
    except asyncio.QueueFull:
        raise SendQueueFull(
            f"outbox to {self._host}:{self._port} is full "
            f"({_OUTBOX_MAXSIZE} messages); {type(msg).__name__} was not sent"
        ) from None

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/indikit/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.

The policy is recorded even when the send fails. Unlike every other send here, this is not a command to an instrument: it is a standing subscription preference, idempotent, and already part of what the client replays onto each new connection. Recording it while disconnected is therefore the same statement as recording it while connected - "BLOBs from this device, please" - and the next connection honours it. The raise still happens, because nothing went out now, so a caller that wants to know the request reached the server can act on it; a caller that just wants BLOBs when the hub returns can ignore it.

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

Raises:

Type	Description
NotConnectedError	Raised if there is no live connection; the policy is remembered regardless.

Source code in src/indikit/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.

    **The policy is recorded even when the send fails.** Unlike every other
    send here, this is not a command to an instrument: it is a standing
    subscription preference, idempotent, and already part of what the
    client replays onto each new connection. Recording it while
    disconnected is therefore the same statement as recording it while
    connected - "BLOBs from this device, please" - and the next connection
    honours it. The raise still happens, because nothing went out now, so a
    caller that wants to know the request reached the server can act on it;
    a caller that just wants BLOBs when the hub returns can ignore it.

    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.

    Raises
    ------
    NotConnectedError
        Raised if there is no live connection; the policy is remembered
        regardless.
    """
    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/indikit/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/indikit/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/indikit/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/indikit/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/indikit/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

indikit.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, message: str | None = None, timestamp: datetime | None = 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.
message	str or None	The explanation a delProperty carried, if any. Only a del sets this: a def or set keeps its message on the vector, whereas a deletion has no vector to keep anything on, and the text is often the only account of why the property went away.
timestamp	datetime or None	When a delProperty said the retraction happened, if it said. Carried for the same reason as message, and None for a def or set, whose vector is already stamped.

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/indikit/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/indikit/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/indikit/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/indikit/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/indikit/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/indikit/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/indikit/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/indikit/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, state_present=msg.state_present)
        return PropertyEvent("set", cur.device, cur.name, cur)
    if isinstance(msg, DelProperty):
        return self._delete(msg)
    return None

subscribe

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/indikit/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.

A whole-device del reaches every subscriber for that device, including the name-filtered ones. Its event carries no name because the deletion names no property - it takes all of them - so matching the filter against it literally would silence exactly the subscribers with the most to lose: subscribe(cb, device="CCD", name="EXPOSURE") heard nothing when the CCD's driver died and indiserver withdrew the device, which is the one event that watcher must not miss.

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/indikit/client/store.py

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

    A whole-device ``del`` reaches **every** subscriber for that device,
    including the name-filtered ones. Its event carries no name because the
    deletion names no property - it takes all of them - so matching the
    filter against it literally would silence exactly the subscribers with
    the most to lose: ``subscribe(cb, device="CCD", name="EXPOSURE")`` heard
    nothing when the CCD's driver died and ``indiserver`` withdrew the
    device, which is the one event that watcher must not miss.

    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 event.name is not None and name != event.name:
            continue
        out.append(callback)
    return out