indikit.driver

The driver SDK: subclass Device, declare properties in setup(), poll with @every, handle client writes with @on_new, and serve over stdio under indiserver.

Device

indikit.driver.device

The Device base class - what a driver author subclasses.

A driver is a subclass of :class:Device that

• defines its properties in :meth:Device.setup (called once, when a client first asks what this device exposes),
• pushes updates through the :class:BoundProperty handles that define_* returns - typically from @every polling jobs,
• and handles client writes with @on_new methods.

The vocabulary is plain Python rather than the libindi C surface (IUFind, IDSetNumber, IEAddTimer).

Device

Device(name: str | None = None)

Base class for an INDI driver device.

Subclass it, set :attr:name (optional; defaults to the class name), and override :meth:setup.

Attributes:

Name	Type	Description
name	str	Class attribute; override to set the INDI device name. Empty means "use the class name".
serialize_dispatch	bool	Class attribute; whether periodic ticks and client writes are run under a per-device lock so they never interleave. On by default.

Initialise the device and discover its @on_new handlers.

Parameters:

Name	Type	Description	Default
name	str	Instance-level device name override. Falls back to the class :attr:name, then to the class name.	None

Source code in src/indikit/driver/device.py

def __init__(self, name: str | None = None) -> None:
    """Initialise the device and discover its ``@on_new`` handlers.

    Parameters
    ----------
    name : str, optional
        Instance-level device name override. Falls back to the class
        :attr:`name`, then to the class name.
    """
    self._device = name or type(self).name or type(self).__name__
    self._properties: dict[str, BoundProperty[Any]] = {}
    # iter_new_handlers walks the MRO subclass-first, so keep the *first*
    # handler per property name: a subclass @on_new shadows any base-class
    # handler for the same property (e.g. the built-in CONNECTION one).
    self._new_handlers: dict[str, NewHandler] = {}
    for prop_name, method in iter_new_handlers(self):
        self._new_handlers.setdefault(prop_name, method)
    self._emit: Emit | None = None
    # This device's configuration as it stands right now: element values
    # keyed by property name, for every property declared persist=True. It
    # is one authoritative map rather than a cache of the file, and the four
    # rules in define_config's docstring are the whole of its behaviour.
    self._config_values: dict[str, dict[str, Any]] = {}
    # Where that configuration is written. Injected by whatever is serving
    # the device (the runtime, the harness); `None` until then, and `None`
    # afterwards on a machine with no resolvable configuration directory.
    self._config_dir: Path | None = None
    self._setup_done = False
    # Set once setup() has run; periodic (@every) jobs wait on it so they
    # never touch a property before setup() defines it.
    self._setup_complete = asyncio.Event()
    # Guards ticks against handlers; see the serialize_dispatch docstring.
    self._dispatch_lock = asyncio.Lock()

device property

device: str

The resolved INDI device name.

connected property

connected: bool

Whether the device link is up.

True when the CONNECTION switch is on - or always, for a device that has no CONNECTION property (no connection semantics).

__repr__

__repr__() -> str

Return a debug representation naming the class and device.

Source code in src/indikit/driver/device.py

def __repr__(self) -> str:
    """Return a debug representation naming the class and device."""
    return f"<{type(self).__name__} device={self._device!r}>"

setup async

setup() -> None

Define the device's properties. Called once, on first getProperties.

Override and call self.define_* here. The base implementation does nothing.

Source code in src/indikit/driver/device.py

async def setup(self) -> None:
    """Define the device's properties. Called once, on first ``getProperties``.

    Override and call ``self.define_*`` here. The base implementation does
    nothing.
    """

on_new_default async

on_new_default(vector: Vector) -> None

Handle a client write to a property with no @on_new handler.

The default is to ignore it. Override for a catch-all.

Parameters:

Name	Type	Description	Default
vector	Vector	The parsed vector the client asked to change.	required

Source code in src/indikit/driver/device.py

async def on_new_default(self, vector: Vector) -> None:
    """Handle a client write to a property with no ``@on_new`` handler.

    The default is to ignore it. Override for a catch-all.

    Parameters
    ----------
    vector : Vector
        The parsed vector the client asked to change.
    """

on_connect async

on_connect() -> None

Open the device's link. Called when a client turns CONNECT on.

Override to open your serial/network connection and define any properties that only exist while connected. The base implementation does nothing. Only used with :meth:define_connection.

Source code in src/indikit/driver/device.py

async def on_connect(self) -> None:
    """Open the device's link. Called when a client turns CONNECT on.

    Override to open your serial/network connection and define any
    properties that only exist while connected. The base implementation
    does nothing. Only used with :meth:`define_connection`.
    """

on_disconnect async

on_disconnect() -> None

Close the device's link. Called when a client turns DISCONNECT on.

Override to halt motion and close your serial/network connection. The base implementation does nothing. Only used with :meth:define_connection.

Source code in src/indikit/driver/device.py

async def on_disconnect(self) -> None:
    """Close the device's link. Called when a client turns DISCONNECT on.

    Override to halt motion and close your serial/network connection. The
    base implementation does nothing. Only used with
    :meth:`define_connection`.
    """

define_connection

define_connection(*, label: str = 'Connection', group: str = 'Main Control') -> BoundProperty[SwitchVector]

Define the standard INDI CONNECTION switch (initially off).

Call this first in :meth:setup and the device gains the standard connect/disconnect lifecycle for free: the built-in handler flips the switch, calls :meth:on_connect/:meth:on_disconnect, and announces the transition; :attr:connected and :meth:require_connected read the state, and @every(..., when_connected=True) jobs pause while disconnected. (libindi's INDI::DefaultDevice provides the same property implicitly; here it is one explicit line.)

Parameters:

Name	Type	Description	Default
label	str	The property label shown by clients.	'Connection'
group	str	The property group (tab) shown by clients.	'Main Control'

Returns:

Name	Type	Description
prop	BoundProperty	The handle for the CONNECTION property.

Source code in src/indikit/driver/device.py

def define_connection(
    self, *, label: str = "Connection", group: str = "Main Control"
) -> BoundProperty[SwitchVector]:
    """Define the standard INDI ``CONNECTION`` switch (initially off).

    Call this first in :meth:`setup` and the device gains the standard
    connect/disconnect lifecycle for free: the built-in handler flips the
    switch, calls :meth:`on_connect`/:meth:`on_disconnect`, and announces
    the transition; :attr:`connected` and :meth:`require_connected` read
    the state, and ``@every(..., when_connected=True)`` jobs pause while
    disconnected. (libindi's ``INDI::DefaultDevice`` provides the same
    property implicitly; here it is one explicit line.)

    Parameters
    ----------
    label : str, optional
        The property label shown by clients.
    group : str, optional
        The property group (tab) shown by clients.

    Returns
    -------
    prop : BoundProperty
        The handle for the CONNECTION property.
    """
    return self.define_switch(
        "CONNECTION",
        [
            Switch(name="CONNECT", label="Connect", value=ISState.OFF),
            Switch(name="DISCONNECT", label="Disconnect", value=ISState.ON),
        ],
        rule=ISRule.ONE_OF_MANY,
        label=label,
        group=group,
    )

require_connected

require_connected() -> bool

Return whether commands may run, logging the standard error if not.

The one-line guard for @on_new handlers::

if not self.require_connected():
    return

Returns:

Name	Type	Description
allowed	bool	True when connected (or connection-less); otherwise False after sending the standard "not connected" error message.

Source code in src/indikit/driver/device.py

def require_connected(self) -> bool:
    """Return whether commands may run, logging the standard error if not.

    The one-line guard for ``@on_new`` handlers::

        if not self.require_connected():
            return

    Returns
    -------
    allowed : bool
        `True` when connected (or connection-less); otherwise `False`
        after sending the standard "not connected" error message.
    """
    if self.connected:
        return True
    self.log_error(f"{self._device} is not connected.")
    return False

define_config

define_config(*, label: str = 'Configuration', group: str = 'Options') -> BoundProperty[SwitchVector]

Define the standard INDI CONFIG_PROCESS switch.

Three momentary actions - CONFIG_LOAD, CONFIG_SAVE and CONFIG_PURGE - wired to :meth:load_config, :meth:save_config and :meth:purge_config by a built-in handler, which a subclass @on_new("CONFIG_PROCESS") shadows the way it shadows the CONNECTION one. Every libindi driver publishes this property, so a client already knows what the buttons do.

This defines a property and does no I/O. Restoring the saved configuration at startup is one explicit line in :meth:setup::

async def setup(self) -> None:
    self.define_connection()
    self.define_config()
    self.define_number("GEOGRAPHIC_COORD", [...], persist=True)
    with contextlib.suppress(ConfigError):
        await self.load_config()   # a first run has nothing saved

Where that line goes is not a correctness question - :meth:load_config applies to every persisted property already defined and stays in place for every one defined after it, an on_connect's included - but the two orders differ in one visible way. Load before the persisted define_* calls and each property is announced once, already holding its saved value. Load after them, as above, and each is announced with its built-in default and corrected a moment later, in exchange for :meth:on_config_loaded being handed the names while the properties are all there.

What is saved is chosen per property, at define time, with persist=True. Values only, never definitions: labels, permissions and limits belong to the code, which is the only thing that knows what this version of the driver publishes. Lights and BLOBs cannot be persisted at all.

Because that choice is declarative, the device can tell a client what Save writes, which no libindi driver can: a read-only INDIKIT_CONFIG_PERSISTED text property whose PROPERTIES element lists the persisted property names, separated by spaces. It is published once :meth:setup returns, so it names the whole set rather than growing an element at a time, and updated whenever the membership really changes - a persisted property defined on connect, or withdrawn on disconnect. A device that persists nothing publishes it empty: "this driver saves nothing" and "this driver cannot tell you" are different answers, and only the property being absent means the second.

The device keeps one authoritative map of its configuration, and four rules govern it:

• :meth:load_config merges the file into it, applies it to every persisted property currently defined, and leaves it in place for the ones defined afterwards.
• define_*(persist=True) applies it before announcing the property, so startup puts one frame on the wire and not a default followed by a correction.
• Withdrawing a persisted property captures its current values into it first, so defining the property again restores what the operator had rather than what is on disk.
• :meth:save_config refreshes it from every live persisted property and then writes the whole of it, so a Save taken while a connect-time property is withdrawn does not erase that property's values.

Two drivers sharing one configuration directory and one device name overwrite each other, last writer wins. Nothing can arbitrate that across processes, and two devices answering to one name is already unresolvable for a client; libindi has the identical property with $HOME/.indi/<device>_config.xml.

Parameters:

Name	Type	Description	Default
label	str	The property label shown by clients.	'Configuration'
group	str	The property group (tab) shown by clients.	'Options'

Returns:

Name	Type	Description
prop	BoundProperty	The handle for the CONFIG_PROCESS property.

Source code in src/indikit/driver/device.py

def define_config(
    self, *, label: str = "Configuration", group: str = "Options"
) -> BoundProperty[SwitchVector]:
    """Define the standard INDI ``CONFIG_PROCESS`` switch.

    Three momentary actions - ``CONFIG_LOAD``, ``CONFIG_SAVE`` and
    ``CONFIG_PURGE`` - wired to :meth:`load_config`, :meth:`save_config` and
    :meth:`purge_config` by a built-in handler, which a subclass
    ``@on_new("CONFIG_PROCESS")`` shadows the way it shadows the
    ``CONNECTION`` one. Every libindi driver publishes this property, so a
    client already knows what the buttons do.

    **This defines a property and does no I/O.** Restoring the saved
    configuration at startup is one explicit line in :meth:`setup`::

        async def setup(self) -> None:
            self.define_connection()
            self.define_config()
            self.define_number("GEOGRAPHIC_COORD", [...], persist=True)
            with contextlib.suppress(ConfigError):
                await self.load_config()   # a first run has nothing saved

    Where that line goes is not a correctness question - :meth:`load_config`
    applies to every persisted property already defined *and* stays in place
    for every one defined after it, an ``on_connect``'s included - but the
    two orders differ in one visible way. Load **before** the persisted
    ``define_*`` calls and each property is announced once, already holding
    its saved value. Load after them, as above, and each is announced with
    its built-in default and corrected a moment later, in exchange for
    :meth:`on_config_loaded` being handed the names while the properties are
    all there.

    What is saved is chosen per property, at define time, with
    ``persist=True``. Values only, never definitions: labels, permissions
    and limits belong to the code, which is the only thing that knows what
    this version of the driver publishes. Lights and BLOBs cannot be
    persisted at all.

    Because that choice is declarative, the device can **tell a client what
    Save writes**, which no libindi driver can: a read-only
    ``INDIKIT_CONFIG_PERSISTED`` text property whose ``PROPERTIES`` element
    lists the persisted property names, separated by spaces. It is published
    once :meth:`setup` returns, so it names the whole set rather than
    growing an element at a time, and updated whenever the membership really
    changes - a persisted property defined on connect, or withdrawn on
    disconnect. A device that persists nothing publishes it **empty**: "this
    driver saves nothing" and "this driver cannot tell you" are different
    answers, and only the property being absent means the second.

    The device keeps one authoritative map of its configuration, and four
    rules govern it:

    * :meth:`load_config` merges the file into it, applies it to every
      persisted property currently defined, and leaves it in place for the
      ones defined afterwards.
    * ``define_*(persist=True)`` applies it before announcing the property,
      so startup puts one frame on the wire and not a default followed by a
      correction.
    * Withdrawing a persisted property captures its current values into it
      first, so defining the property again restores what the operator had
      rather than what is on disk.
    * :meth:`save_config` refreshes it from every live persisted property
      and then writes the whole of it, so a Save taken while a connect-time
      property is withdrawn does not erase that property's values.

    **Two drivers sharing one configuration directory and one device name
    overwrite each other**, last writer wins. Nothing can arbitrate that
    across processes, and two devices answering to one name is already
    unresolvable for a client; libindi has the identical property with
    ``$HOME/.indi/<device>_config.xml``.

    Parameters
    ----------
    label : str, optional
        The property label shown by clients.
    group : str, optional
        The property group (tab) shown by clients.

    Returns
    -------
    prop : BoundProperty
        The handle for the CONFIG_PROCESS property.
    """
    return self.define_switch(
        CONFIG_PROCESS,
        [
            Switch(name=CONFIG_LOAD, label="Load"),
            Switch(name=CONFIG_SAVE, label="Save"),
            Switch(name=CONFIG_PURGE, label="Purge"),
        ],
        rule=ISRule.AT_MOST_ONE,
        label=label,
        group=group,
    )

on_config_loaded async

on_config_loaded(names: list[str]) -> None

React to a configuration that has just been restored.

Called by :meth:load_config after the values are in the properties and on the wire, with the properties it actually applied to. The default does nothing, which is right for a driver whose configuration is only read when it is used.

Override it when a restored value has to become true of the hardware - a focuser that must physically move to the position it was saved at, a filter wheel that must turn. The shape that works is to keep the body of the corresponding @on_new handler in a method of its own and call it from both places, so the restore does exactly what a client write would do; examples/openmeteo_device.py is the worked version.

Parameters:

Name	Type	Description	Default
names	list of str	The properties the load applied values to.	required

Source code in src/indikit/driver/device.py

async def on_config_loaded(self, names: list[str]) -> None:
    """React to a configuration that has just been restored.

    Called by :meth:`load_config` after the values are in the properties and
    on the wire, with the properties it actually applied to. The default
    does nothing, which is right for a driver whose configuration is only
    read when it is used.

    Override it when a restored value has to *become* true of the hardware -
    a focuser that must physically move to the position it was saved at, a
    filter wheel that must turn. The shape that works is to keep the body of
    the corresponding ``@on_new`` handler in a method of its own and call it
    from both places, so the restore does exactly what a client write would
    do; ``examples/openmeteo_device.py`` is the worked version.

    Parameters
    ----------
    names : list of str
        The properties the load applied values to.
    """

load_config async

load_config() -> None

Restore this device's saved configuration and apply it.

Reads the file, merges it into the device's configuration, publishes a set for every persisted property that is defined right now, and then calls :meth:on_config_loaded. Properties defined afterwards pick their values up as they are defined.

Raises:

Type	Description
ConfigError	Raised if there is nothing saved, or the configuration cannot be located or read. Also an OSError. A first run has nothing saved, so a :meth:setup that calls this handles the failure rather than letting it roll the whole attempt back.

Source code in src/indikit/driver/device.py

async def load_config(self) -> None:
    """Restore this device's saved configuration and apply it.

    Reads the file, merges it into the device's configuration, publishes a
    ``set`` for every persisted property that is defined right now, and then
    calls :meth:`on_config_loaded`. Properties defined afterwards pick their
    values up as they are defined.

    Raises
    ------
    ConfigError
        Raised if there is nothing saved, or the configuration cannot be
        located or read. Also an OSError. A first run has nothing saved, so
        a :meth:`setup` that calls this handles the failure rather than
        letting it roll the whole attempt back.
    """
    path = self._config_file()
    document = await self.off_thread(read_document, path)
    self._config_values.update(document.properties)
    applied: list[str] = []
    rejected: list[str] = []
    for name, values in document.properties.items():
        prop = self._properties.get(name)
        if prop is None or not prop.persist:
            # A property this version of the driver no longer publishes, or
            # one that is not defined yet. Its values stay in the map, so a
            # later define_* still restores them and a Save keeps them.
            continue
        refused = prop._restore(values)
        applied.append(name)
        rejected += [f"{name}.{element}" for element in refused]
    self.message(self._loaded_message(applied, rejected))
    await self.on_config_loaded(applied)

save_config async

save_config() -> None

Write this device's current configuration to disk.

Every persisted property that is defined right now is read into the device's configuration first, and then the whole configuration is written - including properties that are not defined at the moment, whose values were captured when they were withdrawn. That is what makes a Save taken while the instrument is disconnected preserve the connect-time properties instead of erasing them.

The file is replaced whole, so there is no read-modify-write to lose an update to a second process.

Raises:

Type	Description
ConfigError	Raised if the configuration cannot be located or written. Also an OSError.

Source code in src/indikit/driver/device.py

async def save_config(self) -> None:
    """Write this device's current configuration to disk.

    Every persisted property that is defined right now is read into the
    device's configuration first, and then the whole configuration is
    written - including properties that are not defined at the moment,
    whose values were captured when they were withdrawn. That is what makes
    a Save taken while the instrument is disconnected preserve the
    connect-time properties instead of erasing them.

    The file is replaced whole, so there is no read-modify-write to lose an
    update to a second process.

    Raises
    ------
    ConfigError
        Raised if the configuration cannot be located or written. Also an
        OSError.
    """
    path = self._config_file()
    for name, prop in self._properties.items():
        if prop.persist:
            self._config_values[name] = values_of(prop.vector)
    document = ConfigDocument(device=self._device, properties=dict(self._config_values))
    await self.off_thread(write_document, path, document)
    self.message("Configuration saved.")

purge_config async

purge_config() -> None

Delete this device's saved configuration file.

Purging what is not there succeeds: the operator asked for there to be no saved configuration, and afterwards there is none.

The device's live configuration is untouched, deliberately. Purge says "forget the file", not "forget the values the properties are holding", and clearing the map would throw away the last known values of any property that happens to be withdrawn right now.

Raises:

Type	Description
ConfigError	Raised if the configuration cannot be located, or a file is there and cannot be removed. Also an OSError.

Source code in src/indikit/driver/device.py

async def purge_config(self) -> None:
    """Delete this device's saved configuration file.

    Purging what is not there succeeds: the operator asked for there to be
    no saved configuration, and afterwards there is none.

    The device's *live* configuration is untouched, deliberately. Purge says
    "forget the file", not "forget the values the properties are holding",
    and clearing the map would throw away the last known values of any
    property that happens to be withdrawn right now.

    Raises
    ------
    ConfigError
        Raised if the configuration cannot be located, or a file is there
        and cannot be removed. Also an OSError.
    """
    path = self._config_file()
    await self.off_thread(remove_document, path)
    self.message("Saved configuration purged.")

define

define(vector: VectorT, *, emit: EmitPolicy = 'always', persist: bool = False) -> BoundProperty[VectorT]

Register a property vector, emit its def, and return its handle.

A persist=True property is restored from the device's saved configuration before its def goes out, so a driver that comes up with a configuration on disk announces the saved values directly rather than announcing a default and correcting it a moment later. The order is the point: two frames would leave every client briefly holding a value the operator replaced weeks ago, and a panel showing it.

Parameters:

Name	Type	Description	Default
vector	VectorT	The vector to define. If its device is unset, this device's name is filled in.	required
emit	str	When later set calls reach the wire; see ~indikit.driver.property.EmitPolicy.	'always'
persist	bool	Whether this property's element values belong in the device's saved configuration; see :meth:define_config.	False

Returns:

Name	Type	Description
prop	BoundProperty	The handle used to push later updates for this property, typed by the vector kind that was defined.

Raises:

Type	Description
ValueError	Raised if persist=True is asked for on a light or BLOB vector, or on a property whose name contains whitespace.

Source code in src/indikit/driver/device.py

def define[VectorT: Vector](
    self, vector: VectorT, *, emit: EmitPolicy = "always", persist: bool = False
) -> BoundProperty[VectorT]:
    """Register a property vector, emit its ``def``, and return its handle.

    A ``persist=True`` property is restored from the device's saved
    configuration **before** its ``def`` goes out, so a driver that comes up
    with a configuration on disk announces the saved values directly rather
    than announcing a default and correcting it a moment later. The order is
    the point: two frames would leave every client briefly holding a value
    the operator replaced weeks ago, and a panel showing it.

    Parameters
    ----------
    vector : VectorT
        The vector to define. If its ``device`` is unset, this device's name
        is filled in.
    emit : str, optional
        When later ``set`` calls reach the wire; see
        `~indikit.driver.property.EmitPolicy`.
    persist : bool, optional
        Whether this property's element values belong in the device's saved
        configuration; see :meth:`define_config`.

    Returns
    -------
    prop : BoundProperty
        The handle used to push later updates for this property, typed by
        the vector kind that was defined.

    Raises
    ------
    ValueError
        Raised if ``persist=True`` is asked for on a light or BLOB vector,
        or on a property whose name contains whitespace.
    """
    if not vector.device:
        vector.device = self._device
    if persist and isinstance(vector, _UNPERSISTABLE):
        raise ValueError(
            f"{self._device}.{vector.name} is a {type(vector).__name__} and cannot be "
            "persisted: a light is a judgement the driver recomputes and a BLOB is not "
            "configuration"
        )
    # Nothing in INDI forbids whitespace in a property name - models.py puts
    # no pattern on `name` and the 1.7 DTD types it CDATA - so this guard is
    # not a restatement of the protocol. It is what makes the space-separated
    # INDIKIT_CONFIG_PERSISTED encoding unambiguous: one name with a space in
    # it and every client reading that list sees two properties, neither of
    # which exists. It binds only persisted names, because they are the only
    # ones that list carries.
    if persist and any(character.isspace() for character in vector.name):
        raise ValueError(
            f"{self._device}.{vector.name!r} cannot be persisted: whitespace in the name "
            f"would be indistinguishable from a separator in {CONFIG_PERSISTED}"
        )
    prop = BoundProperty(vector, self._send, policy=emit, owner=self, persist=persist)
    self._properties[vector.name] = prop
    if persist:
        self._restore(prop)
    prop._announce()
    if persist:
        # After the announcement, so the list never names a property the
        # client has not been told about yet.
        self._refresh_persisted()
    return prop

define_number

define_number(name: str, elements: list[Number], *, label: str | None = None, group: str | None = None, state: IPState = IPState.IDLE, perm: IPerm = IPerm.RW, timeout: float | None = None, emit: EmitPolicy = 'always', persist: bool = False) -> BoundProperty[NumberVector]

Define a number vector property.

Parameters:

Name	Type	Description	Default
name	str	The property name.	required
elements	list of Number	The number elements the vector contains.	required
label	str	Display label.	None
group	str	GUI group the property belongs to.	None
state	IPState	Initial vector state.	IDLE
perm	IPerm	Client access permission.	RW
timeout	float	Worst-case update time, in seconds.	None
emit	str	When later set calls reach the wire; see ~indikit.driver.property.EmitPolicy.	'always'
persist	bool	Whether this property's element values belong in the device's saved configuration; see :meth:define_config.	False

Returns:

Name	Type	Description
prop	BoundProperty	The handle for the newly defined property.

Source code in src/indikit/driver/device.py

def define_number(
    self,
    name: str,
    elements: list[Number],
    *,
    label: str | None = None,
    group: str | None = None,
    state: IPState = IPState.IDLE,
    perm: IPerm = IPerm.RW,
    timeout: float | None = None,
    emit: EmitPolicy = "always",
    persist: bool = False,
) -> BoundProperty[NumberVector]:
    """Define a number vector property.

    Parameters
    ----------
    name : str
        The property name.
    elements : list of Number
        The number elements the vector contains.
    label : str, optional
        Display label.
    group : str, optional
        GUI group the property belongs to.
    state : IPState, optional
        Initial vector state.
    perm : IPerm, optional
        Client access permission.
    timeout : float, optional
        Worst-case update time, in seconds.
    emit : str, optional
        When later ``set`` calls reach the wire; see
        `~indikit.driver.property.EmitPolicy`.
    persist : bool, optional
        Whether this property's element values belong in the device's saved
        configuration; see :meth:`define_config`.

    Returns
    -------
    prop : BoundProperty
        The handle for the newly defined property.
    """
    return self.define(
        NumberVector(
            device=self._device,
            name=name,
            label=label,
            group=group,
            state=state,
            perm=perm,
            timeout=timeout,
            elements=elements,
        ),
        emit=emit,
        persist=persist,
    )

define_text

define_text(name: str, elements: list[Text], *, label: str | None = None, group: str | None = None, state: IPState = IPState.IDLE, perm: IPerm = IPerm.RW, timeout: float | None = None, emit: EmitPolicy = 'always', persist: bool = False) -> BoundProperty[TextVector]

Define a text vector property.

Parameters:

Name	Type	Description	Default
name	str	The property name.	required
elements	list of Text	The text elements the vector contains.	required
label	str	Display label.	None
group	str	GUI group the property belongs to.	None
state	IPState	Initial vector state.	IDLE
perm	IPerm	Client access permission.	RW
timeout	float	Worst-case update time, in seconds.	None
emit	str	When later set calls reach the wire; see ~indikit.driver.property.EmitPolicy.	'always'
persist	bool	Whether this property's element values belong in the device's saved configuration; see :meth:define_config.	False

Returns:

Name	Type	Description
prop	BoundProperty	The handle for the newly defined property.

Source code in src/indikit/driver/device.py

def define_text(
    self,
    name: str,
    elements: list[Text],
    *,
    label: str | None = None,
    group: str | None = None,
    state: IPState = IPState.IDLE,
    perm: IPerm = IPerm.RW,
    timeout: float | None = None,
    emit: EmitPolicy = "always",
    persist: bool = False,
) -> BoundProperty[TextVector]:
    """Define a text vector property.

    Parameters
    ----------
    name : str
        The property name.
    elements : list of Text
        The text elements the vector contains.
    label : str, optional
        Display label.
    group : str, optional
        GUI group the property belongs to.
    state : IPState, optional
        Initial vector state.
    perm : IPerm, optional
        Client access permission.
    timeout : float, optional
        Worst-case update time, in seconds.
    emit : str, optional
        When later ``set`` calls reach the wire; see
        `~indikit.driver.property.EmitPolicy`.
    persist : bool, optional
        Whether this property's element values belong in the device's saved
        configuration; see :meth:`define_config`.

    Returns
    -------
    prop : BoundProperty
        The handle for the newly defined property.
    """
    return self.define(
        TextVector(
            device=self._device,
            name=name,
            label=label,
            group=group,
            state=state,
            perm=perm,
            timeout=timeout,
            elements=elements,
        ),
        emit=emit,
        persist=persist,
    )

define_switch

define_switch(name: str, elements: list[Switch], *, rule: ISRule = ISRule.ANY_OF_MANY, label: str | None = None, group: str | None = None, state: IPState = IPState.IDLE, perm: IPerm = IPerm.RW, timeout: float | None = None, emit: EmitPolicy = 'always', persist: bool = False) -> BoundProperty[SwitchVector]

Define a switch vector property.

Parameters:

Name	Type	Description	Default
name	str	The property name.	required
elements	list of Switch	The switch elements the vector contains.	required
rule	ISRule	The switch constraint (e.g. OneOfMany).	ANY_OF_MANY
label	str	Display label.	None
group	str	GUI group the property belongs to.	None
state	IPState	Initial vector state.	IDLE
perm	IPerm	Client access permission.	RW
timeout	float	Worst-case update time, in seconds.	None
emit	str	When later set calls reach the wire; see ~indikit.driver.property.EmitPolicy.	'always'
persist	bool	Whether this property's element values belong in the device's saved configuration; see :meth:define_config.	False

Returns:

Name	Type	Description
prop	BoundProperty	The handle for the newly defined property.

Source code in src/indikit/driver/device.py

def define_switch(
    self,
    name: str,
    elements: list[Switch],
    *,
    rule: ISRule = ISRule.ANY_OF_MANY,
    label: str | None = None,
    group: str | None = None,
    state: IPState = IPState.IDLE,
    perm: IPerm = IPerm.RW,
    timeout: float | None = None,
    emit: EmitPolicy = "always",
    persist: bool = False,
) -> BoundProperty[SwitchVector]:
    """Define a switch vector property.

    Parameters
    ----------
    name : str
        The property name.
    elements : list of Switch
        The switch elements the vector contains.
    rule : ISRule, optional
        The switch constraint (e.g. ``OneOfMany``).
    label : str, optional
        Display label.
    group : str, optional
        GUI group the property belongs to.
    state : IPState, optional
        Initial vector state.
    perm : IPerm, optional
        Client access permission.
    timeout : float, optional
        Worst-case update time, in seconds.
    emit : str, optional
        When later ``set`` calls reach the wire; see
        `~indikit.driver.property.EmitPolicy`.
    persist : bool, optional
        Whether this property's element values belong in the device's saved
        configuration; see :meth:`define_config`.

    Returns
    -------
    prop : BoundProperty
        The handle for the newly defined property.
    """
    return self.define(
        SwitchVector(
            device=self._device,
            name=name,
            label=label,
            group=group,
            state=state,
            perm=perm,
            rule=rule,
            timeout=timeout,
            elements=elements,
        ),
        emit=emit,
        persist=persist,
    )

define_light

define_light(name: str, elements: list[Light], *, label: str | None = None, group: str | None = None, state: IPState = IPState.IDLE, emit: EmitPolicy = 'always', persist: bool = False) -> BoundProperty[LightVector]

Define a light vector property.

Lights are always read-only in INDI, so there is no perm argument.

Parameters:

Name	Type	Description	Default
name	str	The property name.	required
elements	list of Light	The light elements the vector contains.	required
label	str	Display label.	None
group	str	GUI group the property belongs to.	None
state	IPState	Initial vector state.	IDLE
emit	str	When later set calls reach the wire; see ~indikit.driver.property.EmitPolicy.	'always'
persist	bool	Whether this property's element values belong in the device's saved configuration; see :meth:define_config.	False

Returns:

Name	Type	Description
prop	BoundProperty	The handle for the newly defined property.

Source code in src/indikit/driver/device.py

def define_light(
    self,
    name: str,
    elements: list[Light],
    *,
    label: str | None = None,
    group: str | None = None,
    state: IPState = IPState.IDLE,
    emit: EmitPolicy = "always",
    persist: bool = False,
) -> BoundProperty[LightVector]:
    """Define a light vector property.

    Lights are always read-only in INDI, so there is no ``perm`` argument.

    Parameters
    ----------
    name : str
        The property name.
    elements : list of Light
        The light elements the vector contains.
    label : str, optional
        Display label.
    group : str, optional
        GUI group the property belongs to.
    state : IPState, optional
        Initial vector state.
    emit : str, optional
        When later ``set`` calls reach the wire; see
        `~indikit.driver.property.EmitPolicy`.
    persist : bool, optional
        Whether this property's element values belong in the device's saved
        configuration; see :meth:`define_config`.

    Returns
    -------
    prop : BoundProperty
        The handle for the newly defined property.
    """
    return self.define(
        LightVector(
            device=self._device,
            name=name,
            label=label,
            group=group,
            state=state,
            elements=elements,
        ),
        emit=emit,
        persist=persist,
    )

define_blob

define_blob(name: str, elements: list[BLOB], *, label: str | None = None, group: str | None = None, state: IPState = IPState.IDLE, perm: IPerm = IPerm.RW, timeout: float | None = None, emit: EmitPolicy = 'always', persist: bool = False) -> BoundProperty[BLOBVector]

Define a BLOB vector property.

Parameters:

Name	Type	Description	Default
name	str	The property name.	required
elements	list of BLOB	The BLOB elements the vector contains.	required
label	str	Display label.	None
group	str	GUI group the property belongs to.	None
state	IPState	Initial vector state.	IDLE
perm	IPerm	Client access permission.	RW
timeout	float	Worst-case update time, in seconds.	None
emit	str	When later set calls reach the wire; see ~indikit.driver.property.EmitPolicy.	'always'
persist	bool	Whether this property's element values belong in the device's saved configuration; see :meth:define_config.	False

Returns:

Name	Type	Description
prop	BoundProperty	The handle for the newly defined property.

Source code in src/indikit/driver/device.py

def define_blob(
    self,
    name: str,
    elements: list[BLOB],
    *,
    label: str | None = None,
    group: str | None = None,
    state: IPState = IPState.IDLE,
    perm: IPerm = IPerm.RW,
    timeout: float | None = None,
    emit: EmitPolicy = "always",
    persist: bool = False,
) -> BoundProperty[BLOBVector]:
    """Define a BLOB vector property.

    Parameters
    ----------
    name : str
        The property name.
    elements : list of BLOB
        The BLOB elements the vector contains.
    label : str, optional
        Display label.
    group : str, optional
        GUI group the property belongs to.
    state : IPState, optional
        Initial vector state.
    perm : IPerm, optional
        Client access permission.
    timeout : float, optional
        Worst-case update time, in seconds.
    emit : str, optional
        When later ``set`` calls reach the wire; see
        `~indikit.driver.property.EmitPolicy`.
    persist : bool, optional
        Whether this property's element values belong in the device's saved
        configuration; see :meth:`define_config`.

    Returns
    -------
    prop : BoundProperty
        The handle for the newly defined property.
    """
    return self.define(
        BLOBVector(
            device=self._device,
            name=name,
            label=label,
            group=group,
            state=state,
            perm=perm,
            timeout=timeout,
            elements=elements,
        ),
        emit=emit,
        persist=persist,
    )

delete_property

delete_property(name: str, message: str | None = None) -> None

Withdraw a property by name, or do nothing if there is no such property.

The counterpart to define_*, and the shape a property that only exists while the instrument is reachable wants::

async def on_connect(self) -> None:
    self.define_number("CCD_COOLER", [Number(name="TEMPERATURE")])

async def on_disconnect(self) -> None:
    self.delete_property("CCD_COOLER", "only while connected")

The property is dropped from the device and retracted with a delProperty, so a client that joins after this is not told about it; defining it again on the next connect starts the cycle over with a fresh handle. That is the whole life of an INDI property: defined, deleted and defined again, once per connection, for as long as the driver runs.

An unknown name is deliberately silent - no message on the wire, no exception. That is what makes the call above safe to run on every disconnect, including the disconnect that follows a connect which never got as far as defining anything, and it is what libindi's INDI::DefaultDevice::deleteProperty does (removeProperty fails, and the error it fills in is never read).

No property is protected, CONNECTION included; libindi guards none of them here either. A device that deletes its CONNECTION becomes a device without connection semantics, and :attr:connected reports True for it from then on.

Parameters:

Name	Type	Description	Default
name	str	The property to withdraw.	required
message	str	Optional explanation to include with the deletion.	None

Source code in src/indikit/driver/device.py

def delete_property(self, name: str, message: str | None = None) -> None:
    """Withdraw a property by name, or do nothing if there is no such property.

    The counterpart to ``define_*``, and the shape a property that only
    exists while the instrument is reachable wants::

        async def on_connect(self) -> None:
            self.define_number("CCD_COOLER", [Number(name="TEMPERATURE")])

        async def on_disconnect(self) -> None:
            self.delete_property("CCD_COOLER", "only while connected")

    The property is dropped from the device *and* retracted with a
    ``delProperty``, so a client that joins after this is not told about it;
    defining it again on the next connect starts the cycle over with a fresh
    handle. That is the whole life of an INDI property: defined, deleted and
    defined again, once per connection, for as long as the driver runs.

    An unknown name is deliberately silent - no message on the wire, no
    exception. That is what makes the call above safe to run on every
    disconnect, including the disconnect that follows a connect which never
    got as far as defining anything, and it is what libindi's
    ``INDI::DefaultDevice::deleteProperty`` does (``removeProperty`` fails,
    and the error it fills in is never read).

    No property is protected, ``CONNECTION`` included; libindi guards none
    of them here either. A device that deletes its ``CONNECTION`` becomes a
    device without connection semantics, and :attr:`connected` reports `True`
    for it from then on.

    Parameters
    ----------
    name : str
        The property to withdraw.
    message : str, optional
        Optional explanation to include with the deletion.
    """
    prop = self._properties.get(name)
    if prop is None:
        return
    prop.delete(message)

property

property(name: str) -> BoundProperty[Any]

Return the handle for a previously defined property.

A lookup by name cannot know the vector kind, so the handle it returns is untyped in its vector. When you need prop.vector to narrow - to iterate elements, say - use :meth:number, :meth:text, :meth:switch, :meth:light or :meth:blob instead, or keep the handle that define_* returned.

Parameters:

Name	Type	Description	Default
name	str	The property name passed to a define_* call.	required

Returns:

Name	Type	Description
prop	BoundProperty	The handle for that property.

Raises:

Type	Description
PropertyNotFound	Raised if no property with that name has been defined. Also a KeyError, so mapping-style handling still applies.

Source code in src/indikit/driver/device.py

def property(self, name: str) -> BoundProperty[Any]:
    """Return the handle for a previously defined property.

    A lookup by name cannot know the vector kind, so the handle it returns
    is untyped in its vector. When you need ``prop.vector`` to narrow - to
    iterate elements, say - use :meth:`number`, :meth:`text`, :meth:`switch`,
    :meth:`light` or :meth:`blob` instead, or keep the handle that
    ``define_*`` returned.

    Parameters
    ----------
    name : str
        The property name passed to a ``define_*`` call.

    Returns
    -------
    prop : BoundProperty
        The handle for that property.

    Raises
    ------
    PropertyNotFound
        Raised if no property with that name has been defined. Also a
        KeyError, so mapping-style handling still applies.
    """
    return self._lookup(name)

__getitem__

__getitem__(name: str) -> BoundProperty[Any]

Return the handle for property name (see :meth:property).

Source code in src/indikit/driver/device.py

def __getitem__(self, name: str) -> BoundProperty[Any]:
    """Return the handle for property ``name`` (see :meth:`property`)."""
    return self._lookup(name)

__contains__

__contains__(name: str) -> bool

Return whether a property named name has been defined.

Source code in src/indikit/driver/device.py

def __contains__(self, name: str) -> bool:
    """Return whether a property named ``name`` has been defined."""
    return name in self._properties

number

number(name: str) -> BoundProperty[NumberVector]

Return the handle for a number property, typed as such.

Parameters:

Name	Type	Description	Default
name	str	The property name.	required

Returns:

Name	Type	Description
prop	BoundProperty	The handle, with prop.vector.elements typed list[Number].

Source code in src/indikit/driver/device.py

def number(self, name: str) -> BoundProperty[NumberVector]:
    """Return the handle for a number property, typed as such.

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

    Returns
    -------
    prop : BoundProperty
        The handle, with ``prop.vector.elements`` typed ``list[Number]``.
    """
    return self._typed(name, NumberVector)

text

text(name: str) -> BoundProperty[TextVector]

Return the handle for a text property, typed as such.

Parameters:

Name	Type	Description	Default
name	str	The property name.	required

Returns:

Name	Type	Description
prop	BoundProperty	The handle, with prop.vector.elements typed list[Text].

Source code in src/indikit/driver/device.py

def text(self, name: str) -> BoundProperty[TextVector]:
    """Return the handle for a text property, typed as such.

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

    Returns
    -------
    prop : BoundProperty
        The handle, with ``prop.vector.elements`` typed ``list[Text]``.
    """
    return self._typed(name, TextVector)

switch

switch(name: str) -> BoundProperty[SwitchVector]

Return the handle for a switch property, typed as such.

Parameters:

Name	Type	Description	Default
name	str	The property name.	required

Returns:

Name	Type	Description
prop	BoundProperty	The handle, with prop.vector.elements typed list[Switch].

Source code in src/indikit/driver/device.py

def switch(self, name: str) -> BoundProperty[SwitchVector]:
    """Return the handle for a switch property, typed as such.

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

    Returns
    -------
    prop : BoundProperty
        The handle, with ``prop.vector.elements`` typed ``list[Switch]``.
    """
    return self._typed(name, SwitchVector)

light

light(name: str) -> BoundProperty[LightVector]

Return the handle for a light property, typed as such.

Parameters:

Name	Type	Description	Default
name	str	The property name.	required

Returns:

Name	Type	Description
prop	BoundProperty	The handle, with prop.vector.elements typed list[Light].

Source code in src/indikit/driver/device.py

def light(self, name: str) -> BoundProperty[LightVector]:
    """Return the handle for a light property, typed as such.

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

    Returns
    -------
    prop : BoundProperty
        The handle, with ``prop.vector.elements`` typed ``list[Light]``.
    """
    return self._typed(name, LightVector)

blob

blob(name: str) -> BoundProperty[BLOBVector]

Return the handle for a BLOB property, typed as such.

Parameters:

Name	Type	Description	Default
name	str	The property name.	required

Returns:

Name	Type	Description
prop	BoundProperty	The handle, with prop.vector.elements typed list[BLOB].

Source code in src/indikit/driver/device.py

def blob(self, name: str) -> BoundProperty[BLOBVector]:
    """Return the handle for a BLOB property, typed as such.

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

    Returns
    -------
    prop : BoundProperty
        The handle, with ``prop.vector.elements`` typed ``list[BLOB]``.
    """
    return self._typed(name, BLOBVector)

off_thread async staticmethod

off_thread(func: Callable[..., T], /, *args: Any, **kwargs: Any) -> T

Run a blocking call in a worker thread and await its result.

Instrument libraries are overwhelmingly synchronous - pyserial, a vendor SDK, a requests session. Calling one directly from an async def compiles, reads fine and blocks the event loop for its whole duration: the driver stops answering indiserver, every other property freezes, and nothing reports an error. Route it through here instead::

reading = await self.off_thread(self._hardware.read_all)
self["telemetry"].set(**reading, state=IPState.OK)

Only the blocking call belongs in the thread. Keep property writes on the event loop, as above: the outbox behind set is an :class:asyncio.Queue, which is not thread-safe.

Parameters:

Name	Type	Description	Default
func	Callable	The blocking callable to run.	required
*args	object	Positional arguments for func.	()
**kwargs	object	Keyword arguments for func.	{}

Returns:

Name	Type	Description
result	T	Whatever func returned.

Source code in src/indikit/driver/device.py

@staticmethod
async def off_thread[T](func: Callable[..., T], /, *args: Any, **kwargs: Any) -> T:
    """Run a blocking call in a worker thread and await its result.

    Instrument libraries are overwhelmingly synchronous - ``pyserial``, a
    vendor SDK, a ``requests`` session. Calling one directly from an
    ``async def`` compiles, reads fine and blocks the event loop for its
    whole duration: the driver stops answering ``indiserver``, every other
    property freezes, and nothing reports an error. Route it through here
    instead::

        reading = await self.off_thread(self._hardware.read_all)
        self["telemetry"].set(**reading, state=IPState.OK)

    Only the blocking call belongs in the thread. Keep property writes on
    the event loop, as above: the outbox behind ``set`` is an
    :class:`asyncio.Queue`, which is not thread-safe.

    Parameters
    ----------
    func : Callable
        The blocking callable to run.
    *args : object
        Positional arguments for ``func``.
    **kwargs : object
        Keyword arguments for ``func``.

    Returns
    -------
    result : T
        Whatever ``func`` returned.
    """
    return await asyncio.to_thread(func, *args, **kwargs)

message

message(text: str, *, level: str = 'INFO', timestamp: datetime | None = None) -> None

Send a free-form log/notification message to the client.

Parameters:

Name	Type	Description	Default
text	str	The message body.	required
level	str	A severity label prefixed to the text (e.g. INFO, ERROR).	'INFO'
timestamp	datetime	Message timestamp; defaults to now. INDI timestamps are UTC, so a naive one is read as UTC and an aware one is converted.	None

Source code in src/indikit/driver/device.py

def message(
    self, text: str, *, level: str = "INFO", timestamp: dt.datetime | None = None
) -> None:
    """Send a free-form log/notification ``message`` to the client.

    Parameters
    ----------
    text : str
        The message body.
    level : str, optional
        A severity label prefixed to the text (e.g. ``INFO``, ``ERROR``).
    timestamp : datetime, optional
        Message timestamp; defaults to now. INDI timestamps are UTC, so a
        naive one is read as UTC and an aware one is converted.
    """
    self._send(
        Message(
            device=self._device,
            timestamp=timestamp or indi_now(),
            message=f"[{level}] {text}",
        )
    )

log_error

log_error(text: str) -> None

Send an ERROR-level :meth:message.

Parameters:

Name	Type	Description	Default
text	str	The error text.	required

Source code in src/indikit/driver/device.py

def log_error(self, text: str) -> None:
    """Send an ``ERROR``-level :meth:`message`.

    Parameters
    ----------
    text : str
        The error text.
    """
    self.message(text, level="ERROR")

run classmethod

run(name: str | None = None) -> None

Run this device as an indiserver stdio driver until stdin closes.

Parameters:

Name	Type	Description	Default
name	str	Device-name override passed to the constructor.	None

Source code in src/indikit/driver/device.py

@classmethod
def run(cls, name: str | None = None) -> None:
    """Run this device as an ``indiserver`` stdio driver until stdin closes.

    Parameters
    ----------
    name : str, optional
        Device-name override passed to the constructor.
    """
    from indikit.driver.runtime import run

    run(cls(name=name))

BoundProperty

indikit.driver.property

BoundProperty: a driver-side handle over a protocol vector.

The protocol models in :mod:indikit.protocol.models are pure data - they are the shared wire contract with the frontend and must stay free of runtime behaviour. BoundProperty is the driver-side wrapper that adds the "and now tell the client" behaviour: mutate the vector's elements and emit the corresponding setXxxVector in one call.

The handle is generic in its vector, so define_switch(...) hands back a BoundProperty[SwitchVector] and prop.vector.elements is a list[Switch] rather than the whole element union - reading back what you defined type-checks without a narrowing dance.

The handle is also the only thing holding a driver's live, mutable vector, which is why the rule that an emission is a value is enforced here: every message carrying a vector out of this class carries a copy, never the live model, so nothing the driver does next can change what has already gone on the wire. See :meth:BoundProperty._detached for what that is worth and what it costs.

A driver never constructs this directly; Device.define_* returns one.

BoundProperty

BoundProperty(vector: VectorT, emit: Emit, *, policy: EmitPolicy = 'always', owner: Device | None = None, persist: bool = False)

A property vector plus the hook that pushes updates to the client.

Parameters:

Name	Type	Description	Default
vector	VectorT	The protocol vector this handle wraps and mutates in place.	required
emit	Callable	Callback that queues an outbound message on the runtime.	required
policy	str	When to put a set on the wire; see :data:EmitPolicy.	'always'
owner	Device	The device this property is registered with, so :meth:delete can remove it there rather than only announcing its removal. None for a handle built outside a device (a unit test over a bare vector), which makes :meth:delete announce-only.	None
persist	bool	Whether this property's element values belong in the device's saved configuration; see ~indikit.driver.device.Device.define_config.	False

Wrap vector with the runtime's outbound-message callback.

Source code in src/indikit/driver/property.py

def __init__(
    self,
    vector: VectorT,
    emit: Emit,
    *,
    policy: EmitPolicy = "always",
    owner: Device | None = None,
    persist: bool = False,
) -> None:
    """Wrap ``vector`` with the runtime's outbound-message callback."""
    self._vector = vector
    self._emit = emit
    self._policy: EmitPolicy = policy
    self._owner = owner
    self._persist = persist
    # Set by delete(): the client has been told this property is gone, so
    # anything published through this handle afterwards contradicts that.
    self._retracted = False

vector property

vector: VectorT

The underlying (mutable) protocol model.

name property

name: str

The property's name.

state property

state: IPState

The property's current vector state.

persist property

persist: bool

Whether this property's values belong in the saved configuration.

__getitem__

__getitem__(name: str) -> Element

Return element name (raises :class:PropertyNotFound if absent).

Source code in src/indikit/driver/property.py

def __getitem__(self, name: str) -> Element:
    """Return element ``name`` (raises :class:`PropertyNotFound` if absent)."""
    return self._vector.element(name)

__contains__

__contains__(name: str) -> bool

Return whether this property has an element called name.

The guard for driving a property from hardware that may report a value the driver has no element for::

if reported not in self["state_message"]:
    self.log_error(f"Unknown state {reported!r}")

Source code in src/indikit/driver/property.py

def __contains__(self, name: str) -> bool:
    """Return whether this property has an element called ``name``.

    The guard for driving a property from hardware that may report a value
    the driver has no element for::

        if reported not in self["state_message"]:
            self.log_error(f"Unknown state {reported!r}")
    """
    return any(el.name == name for el in self._vector.elements)

value

value(name: str) -> Any

Return the current value of an element.

Parameters:

Name	Type	Description	Default
name	str	The element name.	required

Returns:

Name	Type	Description
value	object	The element's value (or data for a BLOB element).

Source code in src/indikit/driver/property.py

def value(self, name: str) -> Any:
    """Return the current value of an element.

    Parameters
    ----------
    name : str
        The element name.

    Returns
    -------
    value : object
        The element's ``value`` (or ``data`` for a BLOB element).
    """
    el = self._vector.element(name)
    if isinstance(el, BLOB):
        return el.data
    return el.value

set

set(values: dict[str, Any] | None = None, *, state: IPState | None = None, message: str | None = None, timestamp: datetime | None = None, force: bool = False, **kwargs: Any) -> None

Assign element values, update state, and emit a set to the client.

set(RA=1.23, DEC=4.56, state=IPState.OK) writes the two elements, sets the vector state, stamps the timestamp, and sends a single setNumberVector. For a OneOfMany or AtMostOne switch vector, turning one element On automatically turns its siblings Off.

Under the "on_change" emit policy the values are still written, but nothing goes on the wire (and the timestamp is left alone) when the result is identical to what the client was last told. "Identical" means the wire representation: a number whose declared format renders it the same way has not changed anything a client can see.

Parameters:

Name	Type	Description	Default
values	dict	Element values keyed by name, for names that collide with the reserved keywords below, e.g. set({"state": "Ok"}, state=IPState.OK).	None
state	IPState	New vector state, if changing it.	None
message	str	Optional message to attach to the update.	None
timestamp	datetime	Update timestamp; defaults to now. INDI timestamps are UTC, so a naive one is read as UTC and an aware one is converted.	None
force	bool	Emit even under "on_change" when nothing differs - for the occasional deliberate re-announcement.	False
**kwargs	object	Element values by name (the common case).	{}

Raises:

Type	Description
PropertyNotFound	Raised if a named element is not part of this vector. Also a KeyError.
ProtocolError	Raised if a number element is given a non-finite value, which neither wire format can carry. Also a ValueError.
PropertyRetracted	Raised if the property has been retracted (see :meth:delete). Also a RuntimeError.

Source code in src/indikit/driver/property.py

def set(
    self,
    values: dict[str, Any] | None = None,
    *,
    state: IPState | None = None,
    message: str | None = None,
    timestamp: dt.datetime | None = None,
    force: bool = False,
    **kwargs: Any,
) -> None:
    """Assign element values, update state, and emit a ``set`` to the client.

    ``set(RA=1.23, DEC=4.56, state=IPState.OK)`` writes the two elements, sets
    the vector state, stamps the timestamp, and sends a single
    ``setNumberVector``. For a ``OneOfMany`` or ``AtMostOne`` switch vector,
    turning one element On automatically turns its siblings Off.

    Under the ``"on_change"`` emit policy the values are still written, but
    nothing goes on the wire (and the timestamp is left alone) when the
    result is identical to what the client was last told. "Identical" means
    the *wire* representation: a number whose declared ``format`` renders it
    the same way has not changed anything a client can see.

    Parameters
    ----------
    values : dict, optional
        Element values keyed by name, for names that collide with the
        reserved keywords below, e.g. ``set({"state": "Ok"}, state=IPState.OK)``.
    state : IPState, optional
        New vector state, if changing it.
    message : str, optional
        Optional message to attach to the update.
    timestamp : datetime, optional
        Update timestamp; defaults to now. INDI timestamps are UTC, so a
        naive one is read as UTC and an aware one is converted.
    force : bool, optional
        Emit even under ``"on_change"`` when nothing differs - for the
        occasional deliberate re-announcement.
    **kwargs : object
        Element values by name (the common case).

    Raises
    ------
    PropertyNotFound
        Raised if a named element is not part of this vector. Also a
        KeyError.
    ProtocolError
        Raised if a number element is given a non-finite value, which
        neither wire format can carry. Also a ValueError.
    PropertyRetracted
        Raised if the property has been retracted (see :meth:`delete`).
        Also a RuntimeError.
    """
    self._require_live()
    merged = {**(values or {}), **kwargs}
    before = self._snapshot()
    for elem_name, val in merged.items():
        self._assign(elem_name, val)
    if state is not None:
        self._vector.state = state
    if message is not None:
        self._vector.message = message
    self._publish(before, force=force, timestamp=timestamp)

set_all

set_all(value: Any, *, state: IPState | None = None, message: str | None = None, force: bool = False) -> None

Assign one value to every element and emit a single set.

The reset half of the "one of N lights is lit" idiom::

self["state_message"].set_all(IPState.IDLE)
self["state_message"].set(**{lit: IPState.BUSY}, state=IPState.BUSY)

Parameters:

Name	Type	Description	Default
value	object	The value written to every element, coerced per element kind.	required
state	IPState	New vector state, if changing it.	None
message	str	Optional message to attach to the update.	None
force	bool	Emit even under "on_change" when nothing differs.	False

Source code in src/indikit/driver/property.py

def set_all(
    self,
    value: Any,
    *,
    state: IPState | None = None,
    message: str | None = None,
    force: bool = False,
) -> None:
    """Assign one value to every element and emit a single ``set``.

    The reset half of the "one of N lights is lit" idiom::

        self["state_message"].set_all(IPState.IDLE)
        self["state_message"].set(**{lit: IPState.BUSY}, state=IPState.BUSY)

    Parameters
    ----------
    value : object
        The value written to every element, coerced per element kind.
    state : IPState, optional
        New vector state, if changing it.
    message : str, optional
        Optional message to attach to the update.
    force : bool, optional
        Emit even under ``"on_change"`` when nothing differs.
    """
    self.set(
        dict.fromkeys((el.name for el in self._vector.elements), value),
        state=state,
        message=message,
        force=force,
    )

select

select(name: str, value: Any, *, others: Any = None, state: IPState | None = None, message: str | None = None, force: bool = False) -> None

Give one element value, reset the rest, and emit once.

"Exactly one of these is the current one" is the most common shape in INDI status reporting - a bank of lights where one shows the state the instrument is in, and the vector takes that light's state::

self.light("state_message").select("domeslit_opening", IPState.BUSY)

which is the whole idiom: the named light goes Busy, every sibling goes Idle, and so does the vector. Without a state the vector follows value when that is an :class:IPState.

Parameters:

Name	Type	Description	Default
name	str	The element to select.	required
value	object	The value it takes.	required
others	object	The value every other element takes. Defaults to Idle for a light vector and Off for a switch vector.	None
state	IPState	New vector state. Defaults to value when that is an IPState, otherwise the state is left alone.	None
message	str	Optional message to attach to the update.	None
force	bool	Emit even under "on_change" when nothing differs.	False

Raises:

Type	Description
PropertyNotFound	Raised if name is not an element of this vector. Also a KeyError.
WrongPropertyKind	Raised for a vector kind with no natural "unselected" value, unless others says what it is. Also a TypeError.

Source code in src/indikit/driver/property.py

def select(
    self,
    name: str,
    value: Any,
    *,
    others: Any = None,
    state: IPState | None = None,
    message: str | None = None,
    force: bool = False,
) -> None:
    """Give one element ``value``, reset the rest, and emit once.

    "Exactly one of these is the current one" is the most common shape in
    INDI status reporting - a bank of lights where one shows the state the
    instrument is in, and the vector takes that light's state::

        self.light("state_message").select("domeslit_opening", IPState.BUSY)

    which is the whole idiom: the named light goes Busy, every sibling goes
    Idle, and so does the vector. Without a ``state`` the vector follows
    ``value`` when that is an :class:`IPState`.

    Parameters
    ----------
    name : str
        The element to select.
    value : object
        The value it takes.
    others : object, optional
        The value every other element takes. Defaults to ``Idle`` for a
        light vector and ``Off`` for a switch vector.
    state : IPState, optional
        New vector state. Defaults to ``value`` when that is an `IPState`,
        otherwise the state is left alone.
    message : str, optional
        Optional message to attach to the update.
    force : bool, optional
        Emit even under ``"on_change"`` when nothing differs.

    Raises
    ------
    PropertyNotFound
        Raised if ``name`` is not an element of this vector. Also a
        KeyError.
    WrongPropertyKind
        Raised for a vector kind with no natural "unselected" value, unless
        ``others`` says what it is. Also a TypeError.
    """
    if name not in self:
        raise PropertyNotFound(f"{name!r} not in {self._vector.device}.{self._vector.name}")
    if others is None:
        others = _UNSELECTED.get(type(self._vector))
        if others is None:
            raise WrongPropertyKind(
                f"select() needs others= for a {type(self._vector).__name__}; "
                "only light and switch vectors have a natural unselected value"
            )
    if state is None and isinstance(value, IPState):
        state = value
    self.set(
        {el.name: (value if el.name == name else others) for el in self._vector.elements},
        state=state,
        message=message,
        force=force,
    )

delete

delete(message: str | None = None) -> None

Withdraw this property: drop it from the device, then tell the client.

Deletion is a removal, not just an announcement. The property leaves the device's registry first and the delProperty follows, so a getProperties arriving afterwards - a client joining late - is not told about a property the driver has withdrawn. That order is libindi's: INDI::DefaultDevice::deleteProperty calls removeProperty and only emits if it succeeded.

The handle goes with the property: :meth:set through it afterwards raises rather than publishing an update for something the client has been told no longer exists. A property that comes back comes back through define_*, which registers it again and hands out a fresh handle.

Deleting twice is a no-op the second time - nothing is left to remove and the client has already been told - so a driver that keeps its handle can retract unconditionally::

async def on_disconnect(self) -> None:
    self._cooler.delete("only while connected")

A driver that reaches its properties by name wants ~indikit.driver.device.Device.delete_property instead: self["CCD_COOLER"] raises :class:PropertyNotFound once the property is gone, so the name-based call is the one that can be repeated.

Parameters:

Name	Type	Description	Default
message	str	Optional explanation to include with the deletion, shown by clients that surface it (libindi logs it as a device message before applying the deletion).	None

Source code in src/indikit/driver/property.py

def delete(self, message: str | None = None) -> None:
    """Withdraw this property: drop it from the device, then tell the client.

    Deletion is a *removal*, not just an announcement. The property leaves
    the device's registry first and the ``delProperty`` follows, so a
    ``getProperties`` arriving afterwards - a client joining late - is not
    told about a property the driver has withdrawn. That order is libindi's:
    ``INDI::DefaultDevice::deleteProperty`` calls ``removeProperty`` and only
    emits if it succeeded.

    The handle goes with the property: :meth:`set` through it afterwards
    raises rather than publishing an update for something the client has been
    told no longer exists. A property that comes back comes back through
    ``define_*``, which registers it again and hands out a fresh handle.

    Deleting twice is a no-op the second time - nothing is left to remove and
    the client has already been told - so a driver that keeps its handle can
    retract unconditionally::

        async def on_disconnect(self) -> None:
            self._cooler.delete("only while connected")

    A driver that reaches its properties by name wants
    `~indikit.driver.device.Device.delete_property` instead:
    ``self["CCD_COOLER"]`` raises :class:`PropertyNotFound` once the
    property is gone, so the name-based call is the one that can be
    repeated.

    Parameters
    ----------
    message : str, optional
        Optional explanation to include with the deletion, shown by clients
        that surface it (libindi logs it as a device message before applying
        the deletion).
    """
    if self._retracted:
        return
    self._retracted = True
    # A handle retracts exactly the property it owns, and says nothing when
    # it owns nothing: already retracted above, or superseded here by a
    # redefinition under the same name, whose ``def`` the client has already
    # seen and which a ``delProperty`` for that name would wrongly withdraw.
    if self._owner is not None and not self._owner._forget(self):
        return
    self._emit(
        DelProperty(
            device=self._vector.device,
            name=self._vector.name,
            # Stamped like every other emission: libindi's IDDelete always
            # dates the retraction, and a client logging the event has
            # nothing else to date it by.
            timestamp=indi_now(),
            message=message,
        )
    )

Scheduling (@every)

indikit.driver.scheduling

The @every decorator: declarative periodic jobs for a driver.

The decorator only tags a method with a small :class:PeriodicSpec. Discovery and execution are per-instance: the runtime scans the concrete device object for tagged methods (:func:iter_periodic) and supervises one asyncio task per method. No shared mutable state, so two device instances never interfere.

PeriodicSpec dataclass

PeriodicSpec(interval: float, start_immediately: bool = False, when_connected: bool = False, name: str | None = None)

The schedule attached to an @every-tagged method.

Attributes:

Name	Type	Description
interval	float	Seconds between runs.
start_immediately	bool	Whether to run once at startup before the first interval elapses.
when_connected	bool	Whether ticks are skipped while the device is not connected.
name	str or None	Optional label for the job (currently informational).

every

every(*, seconds: float = 0.0, minutes: float = 0.0, hours: float = 0.0, start_immediately: bool = False, when_connected: bool = False, name: str | None = None) -> Callable[[F], F]

Tag a device method to run on a fixed interval.

The interval is the sum of seconds + minutes + hours. The method may be sync or async. This only records a :class:PeriodicSpec on the function; :class:~indikit.driver.runtime.DriverRuntime discovers and runs it once the device is served.

Parameters:

Name	Type	Description	Default
seconds	float	Seconds component of the interval.	0.0
minutes	float	Minutes component of the interval.	0.0
hours	float	Hours component of the interval. The three components are summed and must total a positive duration.	0.0
start_immediately	bool	If True, run once right away and then every interval thereafter; otherwise the first run is one interval in.	False
when_connected	bool	If True, ticks are skipped while device.connected is false - the usual behavior for polling jobs that talk to real hardware.	False
name	str	Optional label for the job.	None

Returns:

Name	Type	Description
decorator	Callable	A decorator that tags and returns the method unchanged.

Raises:

Type	Description
ValueError	Raised if the combined interval is not positive.

Examples:

>>> class Mount(Device):
...     @every(seconds=1)
...     async def poll(self) -> None:
...         ra, dec = await self.read_mount()
...         self["EQUATORIAL_EOD_COORD"].set(RA=ra, DEC=dec)

Source code in src/indikit/driver/scheduling.py

def every(
    *,
    seconds: float = 0.0,
    minutes: float = 0.0,
    hours: float = 0.0,
    start_immediately: bool = False,
    when_connected: bool = False,
    name: str | None = None,
) -> Callable[[F], F]:
    """Tag a device method to run on a fixed interval.

    The interval is the sum of ``seconds`` + ``minutes`` + ``hours``. The method
    may be sync or async. This only records a :class:`PeriodicSpec` on the
    function; :class:`~indikit.driver.runtime.DriverRuntime` discovers and runs
    it once the device is served.

    Parameters
    ----------
    seconds : float, optional
        Seconds component of the interval.
    minutes : float, optional
        Minutes component of the interval.
    hours : float, optional
        Hours component of the interval. The three components are summed and
        must total a positive duration.
    start_immediately : bool, optional
        If `True`, run once right away and then every interval thereafter;
        otherwise the first run is one interval in.
    when_connected : bool, optional
        If `True`, ticks are skipped while ``device.connected`` is false - the
        usual behavior for polling jobs that talk to real hardware.
    name : str, optional
        Optional label for the job.

    Returns
    -------
    decorator : Callable
        A decorator that tags and returns the method unchanged.

    Raises
    ------
    ValueError
        Raised if the combined interval is not positive.

    Examples
    --------
    >>> class Mount(Device):
    ...     @every(seconds=1)
    ...     async def poll(self) -> None:
    ...         ra, dec = await self.read_mount()
    ...         self["EQUATORIAL_EOD_COORD"].set(RA=ra, DEC=dec)
    """
    interval = seconds + minutes * 60.0 + hours * 3600.0
    if interval <= 0.0:
        raise ValueError("every(...) requires a positive interval")

    spec = PeriodicSpec(
        interval=interval,
        start_immediately=start_immediately,
        when_connected=when_connected,
        name=name,
    )

    def decorator(func: F) -> F:
        """Tag ``func`` with the schedule and return it unchanged."""
        setattr(func, _SPEC_ATTR, spec)
        return func

    return decorator

iter_periodic

iter_periodic(obj: object) -> Iterator[tuple[PeriodicSpec, Callable[[], Any]]]

Yield the schedule and bound method for each @every job on obj.

Walks the full MRO so tagged methods on base classes are found, while an override in a subclass shadows the base entry (whether or not the override is itself tagged) - standard method-resolution semantics.

Parameters:

Name	Type	Description	Default
obj	object	The instance to scan (typically a ~indikit.driver.device.Device).	required

Yields:

Name	Type	Description
spec	PeriodicSpec	The schedule for a tagged job.
method	Callable	The bound method to run for that job.

Source code in src/indikit/driver/scheduling.py

def iter_periodic(obj: object) -> Iterator[tuple[PeriodicSpec, Callable[[], Any]]]:
    """Yield the schedule and bound method for each ``@every`` job on ``obj``.

    Walks the full MRO so tagged methods on base classes are found, while an
    override in a subclass shadows the base entry (whether or not the override is
    itself tagged) - standard method-resolution semantics.

    Parameters
    ----------
    obj : object
        The instance to scan (typically a `~indikit.driver.device.Device`).

    Yields
    ------
    spec : PeriodicSpec
        The schedule for a tagged job.
    method : Callable
        The bound method to run for that job.
    """
    seen: set[str] = set()
    for klass in type(obj).__mro__:
        for attr, value in vars(klass).items():
            if attr in seen:
                continue
            seen.add(attr)
            spec = getattr(value, _SPEC_ATTR, None)
            if isinstance(spec, PeriodicSpec):
                yield spec, getattr(obj, attr)

Dispatch (@on_new)

indikit.driver.dispatch

The @on_new decorator: route client writes to typed handlers.

A handler is tagged with the property name it serves; the device builds a per-instance name -> handler map and hands each incoming newXxxVector to the matching handler as a fully typed, parsed vector.

on_new

on_new(name: str) -> Callable[[F], F]

Tag a method as the handler for client writes to property name.

The handler receives the parsed vector for the property the client is trying to change.

Parameters:

Name	Type	Description	Default
name	str	The property name (the vector's name) this handler serves.	required

Returns:

Name	Type	Description
decorator	Callable	A decorator that tags and returns the method unchanged.

Examples:

>>> @on_new("CONNECTION")
... async def _connect(self, vector: SwitchVector) -> None:
...     connect = vector["CONNECT"].value == ISState.ON
...     ...

Source code in src/indikit/driver/dispatch.py

def on_new(name: str) -> Callable[[F], F]:
    """Tag a method as the handler for client writes to property ``name``.

    The handler receives the parsed vector for the property the client is trying
    to change.

    Parameters
    ----------
    name : str
        The property name (the vector's ``name``) this handler serves.

    Returns
    -------
    decorator : Callable
        A decorator that tags and returns the method unchanged.

    Examples
    --------
    >>> @on_new("CONNECTION")
    ... async def _connect(self, vector: SwitchVector) -> None:
    ...     connect = vector["CONNECT"].value == ISState.ON
    ...     ...
    """

    def decorator(func: F) -> F:
        """Tag ``func`` with the target property name and return it unchanged."""
        setattr(func, _HANDLER_ATTR, name)
        return func

    return decorator

iter_new_handlers

iter_new_handlers(obj: object) -> Iterator[tuple[str, Callable[..., Any]]]

Yield the property name and bound method for each @on_new handler.

Walks the full MRO, with subclass overrides shadowing base entries.

Parameters:

Name	Type	Description	Default
obj	object	The instance to scan (typically a ~indikit.driver.device.Device).	required

Yields:

Name	Type	Description
name	str	The property name a handler serves.
method	Callable	The bound handler for that property.

Source code in src/indikit/driver/dispatch.py

def iter_new_handlers(obj: object) -> Iterator[tuple[str, Callable[..., Any]]]:
    """Yield the property name and bound method for each ``@on_new`` handler.

    Walks the full MRO, with subclass overrides shadowing base entries.

    Parameters
    ----------
    obj : object
        The instance to scan (typically a `~indikit.driver.device.Device`).

    Yields
    ------
    name : str
        The property name a handler serves.
    method : Callable
        The bound handler for that property.
    """
    seen: set[str] = set()
    for klass in type(obj).__mro__:
        for attr, value in vars(klass).items():
            if attr in seen:
                continue
            seen.add(attr)
            prop_name = getattr(value, _HANDLER_ATTR, None)
            if isinstance(prop_name, str):
                yield prop_name, getattr(obj, attr)

Configuration persistence

indikit.driver.config

The file a device's saved configuration lives in, and how it is written.

A driver that has been pointed at a site, given a focuser offset or told which filter sits in slot 3 should still know all of that after a power cut. libindi solves this with CONFIG_PROCESS and an XML file under $HOME/.indi; this module is the same idea in the vocabulary the rest of the SDK uses.

Three decisions are worth reading before changing anything here.

Values, never definitions. The document holds element values and nothing else - no labels, no permissions, no min/max. A definition belongs to the code, which is the only thing that knows what the current version of the driver publishes; a saved definition is a stale copy that outranks it forever, and restoring one would let yesterday's driver decide today's property shapes.

JSON, not libindi's XML, and not libindi's directory. The file is ours: a different schema under the same name would put two frameworks in a fight over one path with no way for either to tell whose file it found. So the directory is ~/.indikit (see :class:~indikit.settings.Settings) - beside ~/.indi, never in it.

Whole-document replace, written atomically. :func:write_document renders the entire configuration each time, into a temporary file created 0600 in the destination directory, and then renames it into place with :func:os.replace. There is no read-modify-write, so a second process holding the same file cannot lose an update to an interleaving; there is no window in which the final name exists half-written or world-readable; and a failure part-way through leaves the previous configuration exactly as it was.

Nothing here knows what a :class:~indikit.driver.device.Device is. It imports :mod:indikit.protocol and the standard library, and that is what keeps tests/test_layering.py flat.

ConfigDocument

Bases: BaseModel

One device's saved configuration: element values, keyed by property.

Attributes:

Name	Type	Description
version	int	The schema version, :data:CONFIG_VERSION for anything written today.
device	str	The INDI device the configuration belongs to. Written for the benefit of somebody reading the file; the filename is what actually locates it.
saved	datetime	When the document was written, in UTC.
properties	dict	Property name to a mapping of element name to value. Switch values are the wire tokens "On" / "Off".

values_of

values_of(vector: Vector) -> dict[str, Any]

Return one vector's element values, reduced to JSON scalars.

The conversion is explicit rather than left to the serialiser: a switch's :class:~indikit.protocol.ISState has to reach the file as the wire token a client would send back, and a value that has no JSON form at all - a BLOB payload, most of it - must never get there by accident.

Parameters:

Name	Type	Description	Default
vector	Vector	The vector to read.	required

Returns:

Name	Type	Description
values	dict	Element name to value, holding only elements that can be persisted.

Source code in src/indikit/driver/config.py

def values_of(vector: Vector) -> dict[str, Any]:
    """Return one vector's element values, reduced to JSON scalars.

    The conversion is explicit rather than left to the serialiser: a switch's
    :class:`~indikit.protocol.ISState` has to reach the file as the wire token
    a client would send back, and a value that has no JSON form at all - a BLOB
    payload, most of it - must never get there by accident.

    Parameters
    ----------
    vector : Vector
        The vector to read.

    Returns
    -------
    values : dict
        Element name to value, holding only elements that can be persisted.
    """
    values: dict[str, Any] = {}
    for element in vector.elements:
        if isinstance(element, Number):
            values[element.name] = float(element.value)
        elif isinstance(element, Text):
            values[element.name] = element.value
        elif isinstance(element, Switch):
            values[element.name] = element.value.value
        # Lights and BLOBs fall through unpersisted. define_*(persist=True)
        # refuses both kinds, so this is unreachable through the SDK; skipping
        # rather than raising is what stops a hand-built vector turning a Save
        # into a failure.
    return values

config_path

config_path(directory: Path, device: str) -> Path

Return the configuration file for one device inside directory.

Parameters:

Name	Type	Description	Default
directory	Path	The configuration directory.	required
device	str	The INDI device name, which becomes the filename stem.	required

Returns:

Name	Type	Description
path	Path	<directory>/<device>.json.

Raises:

Type	Description
ConfigError	Raised if the device name cannot safely be a filename. Also an OSError.

Source code in src/indikit/driver/config.py

def config_path(directory: Path, device: str) -> Path:
    """Return the configuration file for one device inside ``directory``.

    Parameters
    ----------
    directory : Path
        The configuration directory.
    device : str
        The INDI device name, which becomes the filename stem.

    Returns
    -------
    path : Path
        ``<directory>/<device>.json``.

    Raises
    ------
    ConfigError
        Raised if the device name cannot safely be a filename. Also an OSError.
    """
    if not _is_safe_name(device):
        raise ConfigError(f"{device!r} cannot be used as a configuration filename")
    return directory / f"{device}.json"

read_document

read_document(path: Path) -> ConfigDocument

Read and validate one device's configuration file.

Parameters:

Name	Type	Description	Default
path	Path	The file to read.	required

Returns:

Name	Type	Description
document	ConfigDocument	The parsed configuration.

Raises:

Type	Description
ConfigError	Raised if the file is absent, larger than :data:MAX_CONFIG_BYTES, not valid JSON, or not a configuration document this version understands. Also an OSError.

Source code in src/indikit/driver/config.py

def read_document(path: Path) -> ConfigDocument:
    """Read and validate one device's configuration file.

    Parameters
    ----------
    path : Path
        The file to read.

    Returns
    -------
    document : ConfigDocument
        The parsed configuration.

    Raises
    ------
    ConfigError
        Raised if the file is absent, larger than :data:`MAX_CONFIG_BYTES`, not
        valid JSON, or not a configuration document this version understands.
        Also an OSError.
    """
    try:
        size = path.stat().st_size
    except FileNotFoundError:
        raise ConfigError("no saved configuration") from None
    except OSError as exc:
        logger.error("cannot stat %s: %s", path, exc)
        raise ConfigError("saved configuration could not be read") from exc
    if size > MAX_CONFIG_BYTES:
        # Checked before a byte is read: the point is not to parse it at all.
        logger.error("%s is %d bytes, over the %d limit", path, size, MAX_CONFIG_BYTES)
        raise ConfigError("saved configuration is too large to read")
    try:
        raw = path.read_bytes()
    except OSError as exc:
        logger.error("cannot read %s: %s", path, exc)
        raise ConfigError("saved configuration could not be read") from exc
    try:
        # RecursionError, not a ValueError: a deeply nested payload exhausts the
        # decoder's stack rather than failing its grammar, so catching
        # json.JSONDecodeError alone would let it out as a crash.
        payload = json.loads(raw)
        document = ConfigDocument.model_validate(payload)
    except (ValueError, RecursionError) as exc:
        logger.error("cannot parse %s: %s", path, exc)
        raise ConfigError("saved configuration is not readable") from exc
    if document.version != CONFIG_VERSION:
        raise ConfigError(
            f"saved configuration is version {document.version}, not {CONFIG_VERSION}"
        )
    return document

write_document

write_document(path: Path, document: ConfigDocument) -> None

Write one device's configuration, atomically and privately.

The directory is created if it is missing - on the way out only, never on the way in, so a load cannot leave an empty directory behind on a machine that has never saved anything.

Parameters:

Name	Type	Description	Default
path	Path	The destination file.	required
document	ConfigDocument	The configuration to write.	required

Raises:

Type	Description
ConfigError	Raised if the directory cannot be created or the file cannot be written. Also an OSError.

Source code in src/indikit/driver/config.py

def write_document(path: Path, document: ConfigDocument) -> None:
    """Write one device's configuration, atomically and privately.

    The directory is created if it is missing - on the way out only, never on
    the way in, so a load cannot leave an empty directory behind on a machine
    that has never saved anything.

    Parameters
    ----------
    path : Path
        The destination file.
    document : ConfigDocument
        The configuration to write.

    Raises
    ------
    ConfigError
        Raised if the directory cannot be created or the file cannot be
        written. Also an OSError.
    """
    body = document.model_dump_json(indent=2) + "\n"
    try:
        path.parent.mkdir(parents=True, exist_ok=True)
        # mkstemp creates 0600, and the mode is never widened afterwards: a
        # chmod after the rename would leave a window in which the final name
        # is readable by anyone on the machine.
        handle, temporary = tempfile.mkstemp(
            dir=path.parent, prefix=f".{path.name}.", suffix=".tmp"
        )
        try:
            with os.fdopen(handle, "w", encoding="utf-8") as stream:
                stream.write(body)
                stream.flush()
                os.fsync(stream.fileno())
            os.replace(temporary, path)
        except OSError:
            # The replace never happened, so the previous configuration is
            # still whole; all that is left is not to litter beside it.
            Path(temporary).unlink(missing_ok=True)
            raise
    except OSError as exc:
        logger.error("cannot write %s: %s", path, exc)
        raise ConfigError("configuration could not be saved") from exc

remove_document

remove_document(path: Path) -> None

Delete one device's configuration file, if it is there.

Deleting what is already gone is a success, not an error: purging is how an operator says "forget the saved configuration", and that is true whether or not a file was found. libindi's CONFIG_PURGE is a bare remove() for the same reason, and leaves no backup beside it.

Parameters:

Name	Type	Description	Default
path	Path	The file to remove.	required

Raises:

Type	Description
ConfigError	Raised if the file exists and cannot be removed. Also an OSError.

Source code in src/indikit/driver/config.py

def remove_document(path: Path) -> None:
    """Delete one device's configuration file, if it is there.

    Deleting what is already gone is a success, not an error: purging is how an
    operator says "forget the saved configuration", and that is true whether or
    not a file was found. libindi's ``CONFIG_PURGE`` is a bare ``remove()`` for
    the same reason, and leaves no backup beside it.

    Parameters
    ----------
    path : Path
        The file to remove.

    Raises
    ------
    ConfigError
        Raised if the file exists and cannot be removed. Also an OSError.
    """
    try:
        path.unlink(missing_ok=True)
    except OSError as exc:
        logger.error("cannot remove %s: %s", path, exc)
        raise ConfigError("saved configuration could not be removed") from exc

Runtime

indikit.driver.runtime

DriverRuntime: the transport and supervision loop behind a Device.

The runtime does three things:

• read the INDI XML stream from indiserver (stdin), frame it with the M1 :class:~indikit.protocol.xml.XMLStreamParser, and dispatch each message to every device it serves (getProperties -> setup; newXxxVector -> @on_new);
• write every message those devices emit back out (stdout), serialised by the M1 codec;
• supervise each device's @every periodic jobs.

One runtime serves one or more devices, which is the shape libindi drivers have always had: one executable, one stdio pipe, several devices announcing themselves on the first getProperties. There is one stream, so there is one parser, one outbox and one writer; the devices differ only in which of them a message is addressed to.

Concurrency is plain :mod:asyncio: an outbox :class:asyncio.Queue, a writer task draining it, one task per periodic job, and the reader driving the whole thing until stdin reaches EOF. The class takes plain read/write callables so it can be exercised by in-memory streams in tests; :func:run wires it to the real stdin/stdout.

Both ends log one line per message on the shared indikit.wire logger when it is turned up (INDIKIT_WIRE_LOG=1, or indikit --wire), which :func:run reads from the environment. Logging goes to stderr: stdout here is the INDI wire itself.

DriverRuntime

DriverRuntime(devices: Device | Sequence[Device], read: ReadFn, write: WriteFn, *, config_dir: Path | None = None)

Serve one or more :class:~indikit.driver.device.Device over a byte stream.

Inbound dispatch is sequential across co-located devices. The reader awaits each dispatch inline, so while one device's @on_new handler or setup() is running, the next inbound message waits - whichever device it is addressed to. That is head-of-line blocking in the reader, not lock contention: a message naming device A never reaches device B's guard at all, because :meth:~indikit.driver.device.Device._dispatch_get_properties and :meth:~indikit.driver.device.Device._dispatch_new return on the device-name check before entering it. Two things follow, and both are the opposite of the obvious guess:

• off_thread does not help here. It moves the blocking call off the loop, but the handler still awaits it, so the reader stays parked for its whole duration.
• serialize_dispatch = False does not help either. It drops a device's own guard, and the guard was never what B was waiting behind.

What is not affected: outbound traffic, because every device shares one outbox drained by a separate writer task; and @every jobs, which are one task per job taking only their own device's guard, so B keeps polling and publishing throughout A's handler. That is the whole concurrency story of a multi-device driver, and it matches libindi, whose one process dispatches ISNew* inline for exactly the same reason.

When two devices must never delay each other's inbound writes, run them as two drivers. indiserver launches both.

Parameters:

Name	Type	Description	Default
devices	Device or Sequence of Device	The device, or devices, to serve on this stream.	required
read	Callable	Awaitable returning the next chunk of inbound bytes, or b"" at EOF.	required
write	Callable	Awaitable that writes one serialised message to the transport.	required
config_dir	Path or None	Where the devices keep their saved configuration, resolved by whichever entrypoint started the driver. None leaves every device's persistence method raising :class:~indikit.ConfigError, which is what a driver with no CONFIG_PROCESS never notices.	None

Raises:

Type	Description
ValueError	Raised if devices is empty, or if two of them resolve to the same INDI device name. Two devices answering to one name on one stream is not resolvable by any client, and both would answer every message addressed to it.

Bind the devices to their shared transport and outbound-message callback.

Source code in src/indikit/driver/runtime.py

def __init__(
    self,
    devices: Device | Sequence[Device],
    read: ReadFn,
    write: WriteFn,
    *,
    config_dir: Path | None = None,
) -> None:
    """Bind the devices to their shared transport and outbound-message callback."""
    self._devices = (devices,) if isinstance(devices, Device) else tuple(devices)
    if not self._devices:
        raise ValueError("a DriverRuntime needs at least one device to serve")
    names = [device.device for device in self._devices]
    duplicates = sorted({name for name in names if names.count(name) > 1})
    if duplicates:
        raise ValueError(f"duplicate device name(s) on one stream: {', '.join(duplicates)}")
    self._read = read
    self._write = write
    # Unbounded outbox, shared by every device; ``None`` is the writer's
    # shutdown sentinel. The queue is unbounded so a device's synchronous
    # emit never blocks.
    self._outbox: asyncio.Queue[IndiMessage | None] = asyncio.Queue()
    for device in self._devices:
        device._bind(self._emit, config_dir=config_dir)

serve async

serve() -> None

Run until stdin reaches EOF, or the writer fails, or this is cancelled.

On EOF the periodic jobs are cancelled and the writer is allowed to drain any still-queued messages before returning, so a driver that emits and then immediately sees EOF still gets its final messages out.

The reader runs as a task rather than inline because it is no longer the only end that can finish. A writer that dies takes the driver with it: left running, the reader would keep accepting work and the @every jobs would keep filling an outbox nothing drains, and the driver would look perfectly alive to indiserver while answering nothing.

Source code in src/indikit/driver/runtime.py

async def serve(self) -> None:
    """Run until stdin reaches EOF, or the writer fails, or this is cancelled.

    On EOF the periodic jobs are cancelled and the writer is allowed to drain
    any still-queued messages before returning, so a driver that emits and
    then immediately sees EOF still gets its final messages out.

    The reader runs as a task rather than inline because it is no longer the
    only end that can finish. A writer that dies takes the driver with it:
    left running, the reader would keep accepting work and the ``@every``
    jobs would keep filling an outbox nothing drains, and the driver would
    look perfectly alive to ``indiserver`` while answering nothing.
    """
    writer = asyncio.create_task(self._writer_loop())
    reader = asyncio.create_task(self._reader_loop())
    periodic = [
        asyncio.create_task(self._run_periodic(device, spec, method))
        for device in self._devices
        for spec, method in iter_periodic(device)
    ]
    try:
        await asyncio.wait({reader, writer}, return_when=asyncio.FIRST_COMPLETED)
    finally:
        for task in periodic:
            task.cancel()
        reader.cancel()  # a no-op on the EOF path, where it has already returned
        await asyncio.gather(*periodic, reader, return_exceptions=True)
        self._outbox.put_nowait(None)  # let the writer drain, then stop
        await asyncio.wait({writer})
    # Surface whichever end failed. The reader comes first: when a reader
    # failure is what stopped the driver, a writer failure behind it is the
    # symptom rather than the cause.
    for task in (reader, writer):
        failure = None if task.cancelled() else task.exception()
        if failure is not None:
            raise failure

message_name

message_name(msg: IndiMessage) -> str

Return a readable identifier for a message, for log messages.

Used in both directions: an inbound write being dispatched, and an outbound message the writer could not serialise.

Parameters:

Name	Type	Description	Default
msg	IndiMessage	The message being handled.	required

Returns:

Name	Type	Description
name	str	device.property for any message carrying a vector, else the message tag.

Source code in src/indikit/driver/runtime.py

def message_name(msg: IndiMessage) -> str:
    """Return a readable identifier for a message, for log messages.

    Used in both directions: an inbound write being dispatched, and an outbound
    message the writer could not serialise.

    Parameters
    ----------
    msg : IndiMessage
        The message being handled.

    Returns
    -------
    name : str
        ``device.property`` for any message carrying a vector, else the message
        tag.
    """
    if isinstance(msg, (DefVector, SetVector, NewVector)):
        return f"{msg.vector.device}.{msg.vector.name}"
    return type(msg).__name__

task_name

task_name(method: Callable[..., Any]) -> str

Return a readable name for a scheduled method, for log messages.

Parameters:

Name	Type	Description	Default
method	Callable	The scheduled method.	required

Returns:

Name	Type	Description
name	str	The method's __name__ if present, else its repr.

Source code in src/indikit/driver/runtime.py

def task_name(method: Callable[..., Any]) -> str:
    """Return a readable name for a scheduled method, for log messages.

    Parameters
    ----------
    method : Callable
        The scheduled method.

    Returns
    -------
    name : str
        The method's ``__name__`` if present, else its `repr`.
    """
    return getattr(method, "__name__", repr(method))

serve_stdio async

serve_stdio(devices: Device | Sequence[Device], *, config_dir: Path | None = None) -> None

Serve one or more devices over real stdin/stdout (async entrypoint).

Parameters:

Name	Type	Description	Default
devices	Device or Sequence of Device	The device, or devices, to serve on this process's stdio.	required
config_dir	Path or None	Where the devices keep their saved configuration. Resolved by the caller, because this coroutine is what tests and embedders await and reading the environment here would make every one of them do so.	None

Source code in src/indikit/driver/runtime.py

async def serve_stdio(
    devices: Device | Sequence[Device], *, config_dir: Path | None = None
) -> None:
    """Serve one or more devices over real stdin/stdout (async entrypoint).

    Parameters
    ----------
    devices : Device or Sequence of Device
        The device, or devices, to serve on this process's stdio.
    config_dir : Path or None, optional
        Where the devices keep their saved configuration. Resolved by the
        caller, because this coroutine is what tests and embedders await and
        reading the environment here would make every one of them do so.
    """
    read, write = await _open_stdio()
    await DriverRuntime(devices, read, write, config_dir=config_dir).serve()

run

run(devices: Device | Sequence[Device]) -> None

Serve one or more devices over real stdin/stdout until stdin closes.

A list runs several devices from one executable, the shape indiserver has always supported::

run([Camera(), GuideChip(), FilterWheel()])

This is where a driver's logging is configured, from INDIKIT_LOG_LEVEL and INDIKIT_WIRE_LOG in the environment indiserver was started in. That is the whole answer to "what is on the wire" for a driver author with no CLI in the loop: a driver launched as ./my_driver.py reaches here through :meth:~indikit.driver.device.Device.run and picks the variables up.

It is done here, the process entrypoint of the two, and not in :func:serve_stdio, which is a coroutine that tests and embedders await: configuring inside it would have every one of them mutate global logging state as a side effect of running a driver.

Parameters:

Name	Type	Description	Default
devices	Device or Sequence of Device	The device, or devices, to run as an indiserver stdio child.	required

Source code in src/indikit/driver/runtime.py

def run(devices: Device | Sequence[Device]) -> None:
    """Serve one or more devices over real stdin/stdout until stdin closes.

    A list runs several devices from one executable, the shape ``indiserver``
    has always supported::

        run([Camera(), GuideChip(), FilterWheel()])

    **This is where a driver's logging is configured**, from
    ``INDIKIT_LOG_LEVEL`` and ``INDIKIT_WIRE_LOG`` in the environment
    ``indiserver`` was started in. That is the whole answer to "what is on the
    wire" for a driver author with no CLI in the loop: a driver launched as
    ``./my_driver.py`` reaches here through
    :meth:`~indikit.driver.device.Device.run` and picks the variables up.

    It is done here, the process entrypoint of the two, and **not** in
    :func:`serve_stdio`, which is a coroutine that tests and embedders await:
    configuring inside it would have every one of them mutate global logging
    state as a side effect of running a driver.

    Parameters
    ----------
    devices : Device or Sequence of Device
        The device, or devices, to run as an ``indiserver`` stdio child.
    """
    config = settings()
    configure_logging(config.log_level, wire=config.wire_log)
    asyncio.run(serve_stdio(devices, config_dir=config.config_dir))