`indi_nexus.protocol`¶

The single source of truth for the INDI 1.7 wire format: typed enums and Pydantic models that serialize to both INDI XML (for indiserver) and JSON (for browsers).

Enums¶

indi_nexus.protocol.enums ¶

INDI protocol enumerations.

Each enum subclasses :class:enum.StrEnum, so a member is the exact token used on the INDI wire (e.g. IPState.OK == "Ok" is True) and Pydantic serialises it to that token directly.

IPState ¶

Bases: _StrEnum

State of a vector property (the coloured status light in a GUI).

IPerm ¶

Bases: _StrEnum

Client access permission for a vector property.

ISRule ¶

Bases: _StrEnum

Constraint on how many switches in a switch vector may be On.

ISState ¶

Bases: _StrEnum

On/Off state of a single switch.

BLOBPolicy ¶

Bases: _StrEnum

How indiserver should deliver BLOBs to a client.

A client must send an enableBLOB with one of these policies before indiserver will forward any BLOB; the default on the wire is Never.

Models¶

indi_nexus.protocol.models ¶

Typed Pydantic models for INDI properties and wire messages.

Design notes

A vector (NumberVector etc.) is the canonical, in-memory representation of a property. It carries the full metadata plus its elements. A driver holds vectors; a client caches them and applies incoming updates onto them.
Element metadata (format/min/max/step on numbers, rule on switch vectors, ...) is only present in a def message. In set and new messages the wire carries just name + value per element. We model that by making the metadata fields optional with defaults, so the same element class round-trips through both contexts. Clients are expected to merge set values onto the previously-defined vector, which is standard INDI behavior.
The def / set / new distinction is a wire intent, not a different data shape, so it is expressed by the thin event wrappers at the bottom of this module rather than by duplicating every vector five times.

Number ¶

Bases: _Element

A single numeric element (defNumber / oneNumber).

Text ¶

Bases: _Element

A single text element (defText / oneText).

Switch ¶

Bases: _Element

A single switch element (defSwitch / oneSwitch).

Light ¶

Bases: _Element

A single light element (defLight / oneLight). Read-only status.

BLOB ¶

Bases: _Element

A single BLOB element (defBLOB / oneBLOB).

data holds the decoded binary payload; the base64/size framing lives in the codec, not the model.

NumberVector ¶

Bases: _Vector

A vector of numeric elements (defNumberVector / setNumberVector).

TextVector ¶

Bases: _Vector

A vector of text elements (defTextVector / setTextVector).

SwitchVector ¶

Bases: _Vector

A vector of switch elements with a selection rule.

selected ¶

selected() -> str | None

Return the name of the first element that is On, or None.

The idiomatic way to read a OneOfMany/AtMostOne client write: such a write names the newly selected member (often only that member), so the question is "which element is On in this request" - never "what is element X", which raises when X was not sent.

Returns:

Name	Type	Description
`name`	`str or None`	The first On element's name, or `None` when none is On (an `AtMostOne` deselect).

Source code in src/indi_nexus/protocol/models.py

def selected(self) -> str | None:
    """Return the name of the first element that is On, or `None`.

    The idiomatic way to read a ``OneOfMany``/``AtMostOne`` client write:
    such a write names the newly selected member (often *only* that
    member), so the question is "which element is On in this request" -
    never "what is element X", which raises when X was not sent.

    Returns
    -------
    name : str or None
        The first On element's name, or `None` when none is On (an
        ``AtMostOne`` deselect).
    """
    for el in self.elements:
        if el.value is ISState.ON:
            return el.name
    return None

LightVector ¶

Bases: _Vector

A vector of read-only light elements (no perm; lights are always RO).

BLOBVector ¶

Bases: _Vector

A vector of BLOB elements (binary payloads).

GetProperties ¶

Bases: _Model

Client -> device/server request to enumerate properties.

DelProperty ¶

Bases: _Model

Notification that a property (or a whole device) has gone away.

Message ¶

Bases: _Model

A free-form log/notification message.

EnableBLOB ¶

Bases: _Model

Client -> server request controlling BLOB delivery.

indiserver withholds BLOBs from a client until it sends this; the policy applies to one device (and optionally one property when name is set).

DefVector ¶

Bases: _Model

A property definition (device -> client).

SetVector ¶

Bases: _Model

A value update to an already-defined property (device -> client).

NewVector ¶

Bases: _Model

A client's request to change a property's value (client -> device).

XML codec¶

indi_nexus.protocol.xml ¶

INDI 1.7 XML codec: models <-> canonical INDI XML.

Two directions:

:func:to_xml serializes a model (DefVector/SetVector/NewVector or a bare message) to the exact def*/set*/new* XML that indiserver and C++ INDI clients expect.
:class:XMLStreamParser consumes the raw byte stream from a socket or stdio pipe and yields fully-formed :data:~indi_nexus.protocol.models.IndiMessage objects as complete top-level elements arrive, reassembling messages across arbitrary chunk boundaries (BLOB payloads included).

Number values honour the INDI printf-style format, including the %m sexagesimal form used for RA/Dec, so values round-trip faithfully with libindi.

XMLStreamParser ¶

XMLStreamParser()

Incremental parser for the unbounded INDI element stream.

The INDI wire is a sequence of sibling top-level elements with no enclosing document root, so we feed a synthetic root and emit each depth-1 element as it completes, clearing consumed nodes to keep memory flat.

Start the pull parser and open a synthetic enclosing root element.

Source code in src/indi_nexus/protocol/xml.py

def __init__(self) -> None:
    """Start the pull parser and open a synthetic enclosing root element."""
    self._parser = etree.XMLPullParser(events=("end",), recover=True, huge_tree=True)
    self._parser.feed(b"<indinexus>")

feed ¶

feed(data: bytes | str) -> Iterator[IndiMessage]

Feed the next chunk of bytes and yield any completed messages.

Parameters:

Name	Type	Description	Default
`data`	`bytes or str`	The next bytes or string from the stream.	required

Yields:

Name	Type	Description
`message`	`IndiMessage`	Each top-level message that completed within this chunk.

Source code in src/indi_nexus/protocol/xml.py

def feed(self, data: bytes | str) -> Iterator[IndiMessage]:
    """Feed the next chunk of bytes and yield any completed messages.

    Parameters
    ----------
    data : bytes or str
        The next bytes or string from the stream.

    Yields
    ------
    message : IndiMessage
        Each top-level message that completed within this chunk.
    """
    if isinstance(data, str):
        data = data.encode("utf-8")
    self._parser.feed(data)
    for _event, item in self._parser.read_events():
        # We only subscribed to "end" events, so the payload is always an
        # element; the lxml stubs can't narrow that, hence the cast.
        element = cast("etree._Element", item)
        parent = element.getparent()
        # depth-1: a complete top-level INDI message (parent is our root)
        if parent is None or parent.getparent() is not None:
            continue
        msg = message_from_xml(element)
        if msg is not None:
            yield msg
        # Free memory: drop this element and earlier siblings.
        element.clear()
        prev = element.getprevious()
        while prev is not None:
            parent.remove(prev)
            prev = element.getprevious()

parse_number ¶

parse_number(text: str) -> float

Parse a number that may be decimal or sexagesimal.

Accepts plain decimals as well as dd:mm:ss (or space-separated) sexagesimal forms used for RA/Dec, mirroring libindi's f_scansexa.

Parameters:

Name	Type	Description	Default
`text`	`str`	The raw element text.	required

Returns:

Name	Type	Description
`value`	`float`	The parsed value; `0.0` for empty input.

Source code in src/indi_nexus/protocol/xml.py

def parse_number(text: str) -> float:
    """Parse a number that may be decimal or sexagesimal.

    Accepts plain decimals as well as ``dd:mm:ss`` (or space-separated)
    sexagesimal forms used for RA/Dec, mirroring libindi's ``f_scansexa``.

    Parameters
    ----------
    text : str
        The raw element text.

    Returns
    -------
    value : float
        The parsed value; ``0.0`` for empty input.
    """
    s = text.strip()
    if not s:
        return 0.0
    try:
        return float(s)
    except ValueError:
        pass
    parts = re.split(r"[:\s]+", s)
    sign = -1.0 if parts[0].lstrip().startswith("-") else 1.0
    nums = [abs(float(p)) for p in parts if p]
    acc = 0.0
    for i, n in enumerate(nums):
        acc += n / (60**i)
    return sign * acc

format_number ¶

format_number(value: float, fmt: str) -> str

Format a number per an INDI printf-style format.

Handles ordinary printf conversions as well as the %m sexagesimal form (e.g. %9.6m), field-width padded like libindi's fs_sexa.

Parameters:

Name	Type	Description	Default
`value`	`float`	The number to format.	required
`fmt`	`str`	The INDI `format` string from the element definition.	required

Returns:

Name	Type	Description
`text`	`str`	The formatted value.

Source code in src/indi_nexus/protocol/xml.py

def format_number(value: float, fmt: str) -> str:
    """Format a number per an INDI printf-style format.

    Handles ordinary printf conversions as well as the ``%m`` sexagesimal form
    (e.g. ``%9.6m``), field-width padded like libindi's ``fs_sexa``.

    Parameters
    ----------
    value : float
        The number to format.
    fmt : str
        The INDI ``format`` string from the element definition.

    Returns
    -------
    text : str
        The formatted value.
    """
    m = re.fullmatch(r"%(\d+)\.(\d+)m", fmt.strip())
    if not m:
        try:
            return (fmt % value).strip()
        except (TypeError, ValueError):
            return repr(value)

    width, frac = int(m.group(1)), int(m.group(2))
    fracbase = {9: 360000, 8: 36000, 6: 3600, 5: 600}.get(frac, 60)
    neg = value < 0
    n = round(abs(value) * fracbase)
    d, f = divmod(n, fracbase)
    dd = f"{'-' if neg else ''}{d}"
    field = max(width - frac, 1)
    dd = dd.rjust(field)
    if fracbase == 60:  # dd:mm
        return f"{dd}:{f:02d}"
    if fracbase == 600:  # dd:mm.m
        return f"{dd}:{f // 10:02d}.{f % 10:1d}"
    if fracbase == 3600:  # dd:mm:ss
        return f"{dd}:{f // 60:02d}:{f % 60:02d}"
    if fracbase == 36000:  # dd:mm:ss.s
        return f"{dd}:{f // 600:02d}:{(f % 600) // 10:02d}.{f % 10:1d}"
    # dd:mm:ss.ss
    return f"{dd}:{f // 6000:02d}:{(f % 6000) // 100:02d}.{f % 100:02d}"

to_xml ¶

to_xml(msg: IndiMessage, *, pretty: bool = False) -> bytes

Serialise an INDI message model to canonical INDI XML bytes.

Parameters:

Name	Type	Description	Default
`msg`	`IndiMessage`	The message model to serialise.	required
`pretty`	`bool`	Whether to pretty-print the output.	`False`

Returns:

Name	Type	Description
`xml`	`bytes`	The encoded XML.

Source code in src/indi_nexus/protocol/xml.py

def to_xml(msg: IndiMessage, *, pretty: bool = False) -> bytes:
    """Serialise an INDI message model to canonical INDI XML bytes.

    Parameters
    ----------
    msg : IndiMessage
        The message model to serialise.
    pretty : bool, optional
        Whether to pretty-print the output.

    Returns
    -------
    xml : bytes
        The encoded XML.
    """
    return etree.tostring(_message_xml(msg), pretty_print=pretty)

message_from_xml ¶

message_from_xml(node: _Element) -> IndiMessage | None

Convert a single top-level INDI element node to a message model.

Parameters:

Name	Type	Description	Default
`node`	`_Element`	A top-level INDI element node.	required

Returns:

Name	Type	Description
`message`	`IndiMessage or None`	The parsed message, or `None` for comments/PIs or unrecognised tags.

Source code in src/indi_nexus/protocol/xml.py

def message_from_xml(node: etree._Element) -> IndiMessage | None:
    """Convert a single top-level INDI element node to a message model.

    Parameters
    ----------
    node : lxml.etree._Element
        A top-level INDI element node.

    Returns
    -------
    message : IndiMessage or None
        The parsed message, or `None` for comments/PIs or unrecognised tags.
    """
    tag = node.tag
    if not isinstance(tag, str):  # comments / PIs
        return None

    if tag == "getProperties":
        return GetProperties(
            version=node.get("version") or "1.7",
            device=node.get("device"),
            name=node.get("name"),
        )
    if tag == "delProperty":
        return DelProperty(
            device=node.get("device") or "",
            name=node.get("name"),
            timestamp=_optts(node.get("timestamp")),
            message=node.get("message"),
        )
    if tag == "message":
        return Message(
            device=node.get("device"),
            timestamp=_optts(node.get("timestamp")),
            message=node.get("message") or "",
        )
    if tag == "enableBLOB":
        text = (node.text or "").strip()
        return EnableBLOB(
            device=node.get("device") or "",
            name=node.get("name"),
            policy=BLOBPolicy(text) if text else BLOBPolicy.ALSO,
        )

    m = re.fullmatch(r"(def|set|new)(Number|Text|Switch|Light|BLOB)Vector", tag)
    if m:
        mode, stem = m.group(1), m.group(2)
        vector = _vector_from_xml(node, stem)
        if mode == "def":
            return DefVector(vector=vector)
        if mode == "set":
            return SetVector(vector=vector)
        return NewVector(vector=vector)

    return None

parse_indi ¶

parse_indi(data: bytes | str) -> list[IndiMessage]

Parse a complete chunk of INDI XML into message models.

Convenience wrapper over :class:XMLStreamParser for a self-contained chunk that holds one or more complete top-level elements.

Parameters:

Name	Type	Description	Default
`data`	`bytes or str`	The XML bytes or string to parse.	required

Returns:

Name	Type	Description
`messages`	`list of IndiMessage`	Every message found in the chunk, in order.

Source code in src/indi_nexus/protocol/xml.py

def parse_indi(data: bytes | str) -> list[IndiMessage]:
    """Parse a complete chunk of INDI XML into message models.

    Convenience wrapper over :class:`XMLStreamParser` for a self-contained chunk
    that holds one or more complete top-level elements.

    Parameters
    ----------
    data : bytes or str
        The XML bytes or string to parse.

    Returns
    -------
    messages : list of IndiMessage
        Every message found in the chunk, in order.
    """
    parser = XMLStreamParser()
    return list(parser.feed(data))

JSON codec¶

indi_nexus.protocol.json ¶

INDI JSON codec: models <-> typed JSON for browser clients.

The INDI wire toward indiserver is XML (see :mod:indi_nexus.protocol.xml); toward browsers it is JSON. Both directions serialise the same Pydantic models, so the JSON contract is just the models dumped to JSON - one source of truth, and the frontend's TypeScript types can be generated from the model JSON schema.

:func:to_json and :func:from_json mirror to_xml / parse_indi. The :data:~indi_nexus.protocol.models.IndiMessage union is discriminated by each member's tag literal, so a single :class:pydantic.TypeAdapter round-trips any message. BLOB payloads travel as base64 (configured on the base model).

to_json ¶

to_json(msg: IndiMessage) -> str

Serialise an INDI message model to a JSON string.

Parameters:

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

Returns:

Name	Type	Description
`text`	`str`	The message as JSON (a `bytes` BLOB payload is base64-encoded).

Source code in src/indi_nexus/protocol/json.py

def to_json(msg: IndiMessage) -> str:
    """Serialise an INDI message model to a JSON string.

    Parameters
    ----------
    msg : IndiMessage
        The message model to serialise.

    Returns
    -------
    text : str
        The message as JSON (a ``bytes`` BLOB payload is base64-encoded).
    """
    return _ADAPTER.dump_json(msg).decode("utf-8")

from_json ¶

from_json(data: str | bytes) -> IndiMessage

Parse a JSON string into the matching INDI message model.

Parameters:

Name	Type	Description	Default
`data`	`str or bytes`	A single JSON object with a `tag` field identifying the message.	required

Returns:

Name	Type	Description
`message`	`IndiMessage`	The parsed, fully typed message model.

Source code in src/indi_nexus/protocol/json.py

def from_json(data: str | bytes) -> IndiMessage:
    """Parse a JSON string into the matching INDI message model.

    Parameters
    ----------
    data : str or bytes
        A single JSON object with a ``tag`` field identifying the message.

    Returns
    -------
    message : IndiMessage
        The parsed, fully typed message model.
    """
    return _ADAPTER.validate_json(data)

indi_nexus.protocol¶

Enums¶

indi_nexus.protocol.enums ¶

IPState ¶

IPerm ¶

ISRule ¶

ISState ¶

BLOBPolicy ¶

Models¶

indi_nexus.protocol.models ¶

Number ¶

Text ¶

Switch ¶

Light ¶

BLOB ¶

NumberVector ¶

TextVector ¶

SwitchVector ¶

selected ¶

LightVector ¶

BLOBVector ¶

GetProperties ¶

DelProperty ¶

Message ¶

EnableBLOB ¶

DefVector ¶

SetVector ¶

NewVector ¶

XML codec¶

indi_nexus.protocol.xml ¶

XMLStreamParser ¶

feed ¶

parse_number ¶

format_number ¶

to_xml ¶

message_from_xml ¶

parse_indi ¶

JSON codec¶

indi_nexus.protocol.json ¶

to_json ¶

from_json ¶

`indi_nexus.protocol`¶