indi_nexus.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

indi_nexus.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".

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/indi_nexus/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] = {}
    # 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
    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()

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/indi_nexus/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/indi_nexus/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/indi_nexus/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/indi_nexus/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/indi_nexus/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

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/indi_nexus/driver/device.py

def define_connection(
    self, *, label: str = "Connection", group: str = "Main Control"
) -> BoundProperty:
    """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/indi_nexus/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

define(vector: Vector) -> BoundProperty

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

Parameters:

Name	Type	Description	Default
vector	Vector	The vector to define. If its device is unset, this device's name is filled in.	required

Returns:

Name	Type	Description
prop	BoundProperty	The handle used to push later updates for this property.

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

def define(self, vector: Vector) -> BoundProperty:
    """Register a property vector, emit its ``def``, and return its handle.

    Parameters
    ----------
    vector : Vector
        The vector to define. If its ``device`` is unset, this device's name
        is filled in.

    Returns
    -------
    prop : BoundProperty
        The handle used to push later updates for this property.
    """
    if not vector.device:
        vector.device = self._device
    prop = BoundProperty(vector, self._send)
    self._properties[vector.name] = prop
    self._send(DefVector(vector=vector))
    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) -> BoundProperty

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

Returns:

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

Source code in src/indi_nexus/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,
) -> BoundProperty:
    """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.

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

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

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

Returns:

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

Source code in src/indi_nexus/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,
) -> BoundProperty:
    """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.

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

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

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

Returns:

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

Source code in src/indi_nexus/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,
) -> BoundProperty:
    """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.

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

define_light

define_light(name: str, elements: list[Light], *, label: str | None = None, group: str | None = None, state: IPState = IPState.IDLE) -> BoundProperty

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

Returns:

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

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

def define_light(
    self,
    name: str,
    elements: list[Light],
    *,
    label: str | None = None,
    group: str | None = None,
    state: IPState = IPState.IDLE,
) -> BoundProperty:
    """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.

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

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

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

Returns:

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

Source code in src/indi_nexus/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,
) -> BoundProperty:
    """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.

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

property

property(name: str) -> BoundProperty

Return the handle for a previously defined property.

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
KeyError	Raised if no property with that name has been defined.

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

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

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

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

    Raises
    ------
    KeyError
        Raised if no property with that name has been defined.
    """
    return self._properties[name]

__getitem__

__getitem__(name: str) -> BoundProperty

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

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

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

__contains__

__contains__(name: str) -> bool

Return whether a property named name has been defined.

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

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

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.	None

Source code in src/indi_nexus/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.
    """
    self._send(
        Message(
            device=self._device,
            timestamp=timestamp or dt.datetime.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/indi_nexus/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/indi_nexus/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 indi_nexus.driver.runtime import run

    run(cls(name=name))

BoundProperty

indi_nexus.driver.property

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

The protocol models in :mod:indi_nexus.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.

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

BoundProperty

BoundProperty(vector: Vector, emit: Emit)

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

Parameters:

Name	Type	Description	Default
vector	Vector	The protocol vector this handle wraps and mutates in place.	required
emit	Callable	Callback that queues an outbound message on the runtime.	required

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

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

def __init__(self, vector: Vector, emit: Emit) -> None:
    """Wrap ``vector`` with the runtime's outbound-message callback."""
    self._vector = vector
    self._emit = emit

vector property

vector: Vector

The underlying (mutable) protocol model.

name property

name: str

The property's name.

state property

state: IPState

The property's current vector state.

__getitem__

__getitem__(name: str) -> Element

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

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

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

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/indi_nexus/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, **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 switch vector, turning one element On automatically turns its siblings Off.

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.	None
**kwargs	object	Element values by name (the common case).	{}

Source code in src/indi_nexus/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,
    **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`` switch vector, turning one element
    On automatically turns its siblings Off.

    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.
    **kwargs : object
        Element values by name (the common case).
    """
    merged = {**(values or {}), **kwargs}
    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._vector.timestamp = timestamp or dt.datetime.now()
    self._emit(SetVector(vector=self._vector))

delete

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

Tell the client this property has gone away (delProperty).

Parameters:

Name	Type	Description	Default
message	str	Optional explanation to include with the deletion.	None

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

def delete(self, message: str | None = None) -> None:
    """Tell the client this property has gone away (``delProperty``).

    Parameters
    ----------
    message : str, optional
        Optional explanation to include with the deletion.
    """
    self._emit(DelProperty(device=self._vector.device, name=self._vector.name, message=message))

Scheduling (@every)

indi_nexus.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:~indi_nexus.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/indi_nexus/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:`~indi_nexus.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 ~indi_nexus.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/indi_nexus/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 `~indi_nexus.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)

indi_nexus.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/indi_nexus/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 ~indi_nexus.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/indi_nexus/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 `~indi_nexus.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)

Runtime

indi_nexus.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:~indi_nexus.protocol.xml.XMLStreamParser, and dispatch each message to the device (getProperties -> setup; newXxxVector -> @on_new);
• write every message the device emits back out (stdout), serialised by the M1 codec;
• supervise the device's @every periodic jobs.

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.

DriverRuntime

DriverRuntime(device: Device, read: ReadFn, write: WriteFn)

Serve one :class:~indi_nexus.driver.device.Device over a byte stream.

Parameters:

Name	Type	Description	Default
device	Device	The device to serve.	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

Bind the device to its transport and its outbound-message callback.

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

def __init__(self, device: Device, read: ReadFn, write: WriteFn) -> None:
    """Bind the device to its transport and its outbound-message callback."""
    self._device = device
    self._read = read
    self._write = write
    # Unbounded outbox; ``None`` is the writer's shutdown sentinel. The queue
    # is unbounded so the device's synchronous emit never blocks.
    self._outbox: asyncio.Queue[IndiMessage | None] = asyncio.Queue()
    device._bind(self._emit)

serve async

serve() -> None

Run until stdin reaches EOF (or the caller cancels this coroutine).

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.

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

async def serve(self) -> None:
    """Run until stdin reaches EOF (or the caller cancels this coroutine).

    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.
    """
    writer = asyncio.create_task(self._writer_loop())
    periodic = [
        asyncio.create_task(self._run_periodic(spec, method))
        for spec, method in iter_periodic(self._device)
    ]
    try:
        await self._reader_loop()
    finally:
        for task in periodic:
            task.cancel()
        await asyncio.gather(*periodic, return_exceptions=True)
        self._outbox.put_nowait(None)  # let the writer drain, then stop
        await writer

message_name

message_name(msg: IndiMessage) -> str

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

Parameters:

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

Returns:

Name	Type	Description
name	str	device.property for a property write, else the message tag.

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

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

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

    Returns
    -------
    name : str
        ``device.property`` for a property write, else the message tag.
    """
    if isinstance(msg, 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/indi_nexus/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(device: Device) -> None

Serve device over real stdin/stdout (async entrypoint).

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

async def serve_stdio(device: Device) -> None:
    """Serve ``device`` over real stdin/stdout (async entrypoint)."""
    read, write = await _open_stdio()
    await DriverRuntime(device, read, write).serve()

run

run(device: Device) -> None

Serve device over real stdin/stdout until stdin closes.

Parameters:

Name	Type	Description	Default
device	Device	The device to run as an indiserver stdio child.	required

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

def run(device: Device) -> None:
    """Serve ``device`` over real stdin/stdout until stdin closes.

    Parameters
    ----------
    device : Device
        The device to run as an ``indiserver`` stdio child.
    """
    asyncio.run(serve_stdio(device))