Adapters API Reference

effero.adapters.base

Base interface for device adapters.

DeviceAdapter

Bases: ABC

Base class for all device adapters in the Device Abstraction Layer.

Source code in src/effero/adapters/base.py

class DeviceAdapter(ABC):
    """Base class for all device adapters in the Device Abstraction Layer."""

    @abstractmethod
    async def connect(self) -> None: ...

    @abstractmethod
    async def disconnect(self) -> None: ...

    @abstractmethod
    async def execute(self, command: str, params: dict[str, Any]) -> Any: ...

    @abstractmethod
    async def read_state(self) -> dict[str, Any]: ...

effero.adapters

Effero device adapters.

This package contains various device adapters for communication with IoT devices, robots, cloud APIs, etc.

DeviceAdapter

Bases: ABC

Base class for all device adapters in the Device Abstraction Layer.

Source code in src/effero/adapters/base.py

class DeviceAdapter(ABC):
    """Base class for all device adapters in the Device Abstraction Layer."""

    @abstractmethod
    async def connect(self) -> None: ...

    @abstractmethod
    async def disconnect(self) -> None: ...

    @abstractmethod
    async def execute(self, command: str, params: dict[str, Any]) -> Any: ...

    @abstractmethod
    async def read_state(self) -> dict[str, Any]: ...

CloudAPIAdapter

Bases: DeviceAdapter

Adapter for interacting with RESTful cloud APIs.

Source code in src/effero/adapters/cloud_api/client.py

class CloudAPIAdapter(DeviceAdapter):
    """Adapter for interacting with RESTful cloud APIs."""

    def __init__(self, base_url: str, headers: dict[str, str] | None = None, auth_token: str | None = None) -> None:
        self.base_url = base_url.rstrip("/")
        self.headers = headers or {}
        if auth_token:
            self.headers["Authorization"] = f"Bearer {auth_token}"
        self._client: Any = None
        self._last_state: dict[str, Any] = {}

    async def connect(self) -> None:
        if not HAS_HTTPX:
            raise RuntimeError("httpx is required for CloudAPIAdapter. Install with: pip install httpx")

        self._client = httpx.AsyncClient(base_url=self.base_url, headers=self.headers)
        logger.info(f"Cloud API client initialized for {self.base_url}")

    async def disconnect(self) -> None:
        if self._client:
            await self._client.aclose()
            self._client = None
            logger.info(f"Cloud API client disconnected from {self.base_url}")

    async def execute(self, command: str, params: dict[str, Any]) -> Any:
        method = params.get("method", "POST").upper()
        endpoint = params.get("endpoint", "")
        payload = params.get("payload")

        if self._client:
            response = await self._client.request(method, endpoint, json=payload)
            response.raise_for_status()
            try:
                data = response.json()
            except Exception:
                data = {"text": response.text}
            return data
        else:
            raise RuntimeError("CloudAPIAdapter not connected. Call await adapter.connect() first.")

    async def read_state(self) -> dict[str, Any]:
        if self._client:
            try:
                response = await self._client.get("")
                response.raise_for_status()
                self._last_state = response.json()
            except Exception as e:
                logger.error(f"Failed to read state: {e}")
        return self._last_state

MQTTAdapter

Bases: DeviceAdapter

Pure-asyncio Live MQTT 3.1.1 client.

Operates directly over asyncio TCP sockets with zero external dependencies. Supports auto-reconnection, exponential backoff, wildcard topic filters (+, #), and QoS 1 & QoS 2 delivery acknowledgment handshakes.

Source code in src/effero/adapters/mqtt_matter/client.py

class MQTTAdapter(DeviceAdapter):
    """Pure-asyncio Live MQTT 3.1.1 client.

    Operates directly over asyncio TCP sockets with zero external dependencies.
    Supports auto-reconnection, exponential backoff, wildcard topic filters (+, #),
    and QoS 1 & QoS 2 delivery acknowledgment handshakes.
    """

    def __init__(
        self,
        broker_host: str = "localhost",
        broker_port: int = 1883,
        client_id: str | None = None,
        username: str | None = None,
        password: str | None = None,
        keep_alive: int = 60,
        clean_session: bool = True,
        auto_reconnect: bool = True,
        base_reconnect_delay: float = 0.5,
        max_reconnect_delay: float = 15.0,
        reconnect_jitter: float = 0.2,
        use_tls: bool = False,
        ssl_context: ssl.SSLContext | None = None,
        ca_certs: str | None = None,
        certfile: str | None = None,
        keyfile: str | None = None,
        tls_insecure: bool = False,
    ) -> None:
        self.broker_host = broker_host
        self.broker_port = broker_port
        self.client_id = client_id or f"effero_{uuid.uuid4().hex[:8]}"
        self.username = username
        self.password = password
        self.keep_alive = keep_alive
        self.clean_session = clean_session
        self.auto_reconnect = auto_reconnect
        self.base_reconnect_delay = base_reconnect_delay
        self.max_reconnect_delay = max_reconnect_delay
        self.reconnect_jitter = reconnect_jitter
        self.use_tls = use_tls
        self.ssl_context = ssl_context
        self.ca_certs = ca_certs
        self.certfile = certfile
        self.keyfile = keyfile
        self.tls_insecure = tls_insecure

        self._reader: asyncio.StreamReader | None = None
        self._writer: asyncio.StreamWriter | None = None
        self._connected: bool = False
        self._manual_disconnect: bool = False

        self._reader_task: asyncio.Task[None] | None = None
        self._keepalive_task: asyncio.Task[None] | None = None
        self._reconnect_task: asyncio.Task[None] | None = None

        # Subscriptions: topic_filter -> list of callbacks
        self._subscriptions: dict[str, list[Callable[[str, Any], Awaitable[None] | None]]] = {}
        self._subscriptions_qos: dict[str, int] = {}

        # Local device state cache
        self._state: dict[str, Any] = {}

        # 16-bit packet identifier tracking
        self._next_packet_id: int = 1
        self._send_lock = asyncio.Lock()

        # Inflight tracking
        self._inflight_publishes: dict[int, asyncio.Future[None]] = {}
        self._inflight_packets: dict[int, PublishPacket] = {}
        self._inflight_subscribes: dict[int, asyncio.Future[list[int]]] = {}
        self._inflight_unsubscribes: dict[int, asyncio.Future[None]] = {}
        self._connack_future: asyncio.Future[ConnackPacket] | None = None

        # Incoming QoS 2 message state: packet_id -> (PublishPacket)
        self._incoming_qos2_packets: dict[int, PublishPacket] = {}

        # Last communication timestamp
        self._last_packet_sent: float = 0.0

    @property
    def is_connected(self) -> bool:
        """Return True if currently connected to MQTT broker."""
        return self._connected and self._writer is not None and not self._writer.is_closing()

    def _get_next_packet_id(self) -> int:
        """Generate the next available 16-bit packet identifier (1..65535)."""
        pid = self._next_packet_id
        self._next_packet_id = (self._next_packet_id % 65535) + 1
        return pid

    async def connect(self) -> None:
        """Establish connection to MQTT broker."""
        self._manual_disconnect = False
        await self._establish_connection()

    def _build_ssl_context(self) -> ssl.SSLContext | None:
        """Build or return configured SSLContext for TLS broker connections."""
        if self.ssl_context is not None:
            return self.ssl_context
        if not self.use_tls and self.broker_port != 8883:
            return None
        if self.tls_insecure:
            ctx = ssl._create_unverified_context()
        else:
            ctx = ssl.create_default_context(cafile=self.ca_certs)
        if self.certfile:
            ctx.load_cert_chain(certfile=self.certfile, keyfile=self.keyfile)
        return ctx

    async def _establish_connection(self) -> None:
        """Perform socket connection and MQTT CONNECT / CONNACK handshake."""
        if self._writer and not self._writer.is_closing():
            try:
                self._writer.close()
                await self._writer.wait_closed()
            except Exception:
                pass
        self._reader = None
        self._writer = None

        ssl_ctx = self._build_ssl_context()
        tls_note = " (TLS enabled)" if ssl_ctx is not None else ""
        logger.info(f"Connecting to MQTT broker at {self.broker_host}:{self.broker_port}{tls_note}...")
        self._reader, self._writer = await asyncio.open_connection(self.broker_host, self.broker_port, ssl=ssl_ctx)

        # Send CONNECT packet
        conn_packet = ConnectPacket(
            client_id=self.client_id,
            clean_session=self.clean_session,
            keep_alive=self.keep_alive,
            username=self.username,
            password=self.password,
        )

        loop = asyncio.get_running_loop()
        self._connack_future = loop.create_future()

        # Start reader loop
        if self._reader_task and not self._reader_task.done():
            self._reader_task.cancel()
        self._reader_task = asyncio.create_task(self._read_loop())

        await self._send_packet(conn_packet)

        # Await CONNACK
        try:
            connack = await asyncio.wait_for(self._connack_future, timeout=10.0)
            if connack.return_code != 0:
                raise ConnectionRefusedError(f"MQTT connection rejected by broker with code {connack.return_code}")
            self._connected = True
            logger.info(f"Successfully connected to MQTT broker {self.broker_host}:{self.broker_port}")
        except Exception:
            self._connected = False
            if self._writer and not self._writer.is_closing():
                self._writer.close()
            raise
        finally:
            self._connack_future = None

        # Start keepalive heartbeat
        if self.keep_alive > 0:
            if self._keepalive_task and not self._keepalive_task.done():
                self._keepalive_task.cancel()
            self._keepalive_task = asyncio.create_task(self._keepalive_loop())

        # Resubscribe existing subscriptions
        if self._subscriptions:
            for topic, qos in self._subscriptions_qos.items():
                pid = self._get_next_packet_id()
                sub_pkt = SubscribePacket(packet_id=pid, topics=[(topic, qos)])
                await self._send_packet(sub_pkt)

        # Retransmit unacknowledged QoS 1 / QoS 2 messages with DUP flag
        if self._inflight_packets:
            for pkt in list(self._inflight_packets.values()):
                pkt.dup = True
                await self._send_packet(pkt)

    async def disconnect(self) -> None:
        """Gracefully disconnect from MQTT broker."""
        self._manual_disconnect = True
        self._connected = False

        if self._keepalive_task and not self._keepalive_task.done():
            self._keepalive_task.cancel()
        if self._reconnect_task and not self._reconnect_task.done():
            self._reconnect_task.cancel()

        if self._writer and not self._writer.is_closing():
            try:
                await self._send_packet(DisconnectPacket())
            except Exception as e:
                logger.debug(f"Error sending DISCONNECT: {e}")
            try:
                self._writer.close()
                await self._writer.wait_closed()
            except Exception as e:
                logger.debug(f"Error closing writer: {e}")

        if self._reader_task and not self._reader_task.done():
            self._reader_task.cancel()

        self._reader = None
        self._writer = None
        logger.info("Disconnected from MQTT broker")

    async def _send_packet(self, packet: MQTTPacket) -> None:
        """Encode and transmit a packet through the active socket."""
        data = encode_packet(packet)
        async with self._send_lock:
            if self._writer is None or self._writer.is_closing():
                raise ConnectionResetError("Cannot send packet: MQTT client not connected")
            self._writer.write(data)
            await self._writer.drain()
            self._last_packet_sent = time.monotonic()

    async def _read_loop(self) -> None:
        """Continuous background loop reading and decoding binary MQTT packets."""
        buffer = bytearray()
        try:
            while not self._manual_disconnect:
                if self._reader is None:
                    break
                chunk = await self._reader.read(4096)
                if not chunk:
                    # Remote broker closed socket
                    logger.warning("MQTT broker connection closed (EOF received)")
                    break

                buffer.extend(chunk)
                while buffer:
                    try:
                        packet, consumed = decode_packet(bytes(buffer))
                    except ValueError as err:
                        logger.error(f"Malformed MQTT packet received: {err}")
                        buffer.clear()
                        break

                    if packet is None or consumed == 0:
                        # Wait for more bytes
                        break

                    del buffer[:consumed]
                    await self._handle_packet(packet)

        except asyncio.CancelledError:
            return
        except Exception as e:
            logger.warning(f"Error in MQTT reader loop: {e}")
        finally:
            self._connected = False
            if self._writer and not self._writer.is_closing():
                try:
                    self._writer.close()
                except Exception:
                    pass
            if not self._manual_disconnect and self.auto_reconnect:
                self._schedule_reconnect()

    def _schedule_reconnect(self) -> None:
        """Schedule automatic reconnection in the background."""
        if self._reconnect_task is None or self._reconnect_task.done():
            self._reconnect_task = asyncio.create_task(self._reconnect_loop())

    async def _reconnect_loop(self) -> None:
        """Exponential backoff reconnection watchdog."""
        retry_count = 0
        while not self._manual_disconnect and not self.is_connected:
            delay = min(
                self.max_reconnect_delay,
                self.base_reconnect_delay * (2**retry_count),
            )
            jitter = random.uniform(0, self.reconnect_jitter)
            wait_time = delay + jitter
            logger.info(f"Reconnecting to MQTT broker in {wait_time:.2f}s (attempt {retry_count + 1})...")
            try:
                await asyncio.sleep(wait_time)
                await self._establish_connection()
                logger.info("Successfully reconnected to MQTT broker")
                return
            except Exception as e:
                logger.warning(f"Reconnection attempt {retry_count + 1} failed: {e}")
                retry_count += 1

    async def _keepalive_loop(self) -> None:
        """Periodic keep-alive heartbeat loop."""
        interval = max(1.0, float(self.keep_alive) * 0.75)
        try:
            while self.is_connected:
                await asyncio.sleep(interval)
                elapsed = time.monotonic() - self._last_packet_sent
                if elapsed >= interval:
                    try:
                        await self._send_packet(PingreqPacket())
                    except Exception as e:
                        logger.warning(f"Failed to send PINGREQ: {e}")
                        break
        except asyncio.CancelledError:
            return

    async def _handle_packet(self, packet: Any) -> None:
        """Process a decoded MQTT packet according to protocol rules."""
        ptype = getattr(packet, "packet_type", None)

        if ptype == PacketType.CONNACK:
            if self._connack_future and not self._connack_future.done():
                self._connack_future.set_result(packet)

        elif ptype == PacketType.SUBACK:
            fut_sub = self._inflight_subscribes.pop(packet.packet_id, None)
            if fut_sub and not fut_sub.done():
                fut_sub.set_result(packet.return_codes or [])

        elif ptype == PacketType.UNSUBACK:
            fut_unsub = self._inflight_unsubscribes.pop(packet.packet_id, None)
            if fut_unsub and not fut_unsub.done():
                fut_unsub.set_result(None)

        elif ptype == PacketType.PUBACK:
            self._inflight_packets.pop(packet.packet_id, None)
            fut_pub = self._inflight_publishes.pop(packet.packet_id, None)
            if fut_pub and not fut_pub.done():
                fut_pub.set_result(None)

        elif ptype == PacketType.PUBREC:
            # QoS 2: Broker received message, send PUBREL
            pubrel = PubrelPacket(packet_id=packet.packet_id)
            await self._send_packet(pubrel)

        elif ptype == PacketType.PUBREL:
            # QoS 2: Broker released message, reply with PUBCOMP and dispatch
            pubcomp = PubcompPacket(packet_id=packet.packet_id)
            await self._send_packet(pubcomp)
            saved_packet = self._incoming_qos2_packets.pop(packet.packet_id, None)
            if saved_packet:
                await self._dispatch_publish(saved_packet)

        elif ptype == PacketType.PUBCOMP:
            # QoS 2: Transaction complete
            self._inflight_packets.pop(packet.packet_id, None)
            fut_pub2 = self._inflight_publishes.pop(packet.packet_id, None)
            if fut_pub2 and not fut_pub2.done():
                fut_pub2.set_result(None)

        elif ptype == PacketType.PUBLISH:
            if packet.qos == 0:
                await self._dispatch_publish(packet)
            elif packet.qos == 1:
                puback = PubackPacket(packet_id=packet.packet_id)
                await self._send_packet(puback)
                await self._dispatch_publish(packet)
            elif packet.qos == 2:
                # Store until PUBREL received
                self._incoming_qos2_packets[packet.packet_id] = packet
                pubrec = PubrecPacket(packet_id=packet.packet_id)
                await self._send_packet(pubrec)

        elif ptype == PacketType.PINGRESP:
            logger.debug("Received PINGRESP from broker")

    async def _dispatch_publish(self, packet: PublishPacket) -> None:
        """Dispatch an incoming PUBLISH packet to registered topic subscribers."""
        topic = packet.topic
        payload_bytes = packet.payload
        try:
            payload_str = payload_bytes.decode("utf-8")
            try:
                payload_val: Any = json.loads(payload_str)
            except Exception:
                payload_val = payload_str
        except Exception:
            payload_val = payload_bytes

        # Update local state cache
        self._state[topic] = payload_val

        # Dispatch to matching callbacks
        for filter_topic, callbacks in list(self._subscriptions.items()):
            if topic_matches(filter_topic, topic):
                for cb in callbacks:
                    try:
                        res = cb(topic, payload_val)
                        if asyncio.iscoroutine(res):
                            await res
                    except Exception as e:
                        logger.error(f"Error in MQTT subscription callback for topic {topic}: {e}")

    async def publish(
        self,
        topic: str,
        payload: Any,
        qos: int = 0,
        retain: bool = False,
        timeout: float = 10.0,
    ) -> None:
        """Publish a message to an MQTT topic with specified QoS (0, 1, or 2)."""
        if isinstance(payload, bytes):
            payload_bytes = payload
        elif isinstance(payload, (dict, list)):
            payload_bytes = json.dumps(payload).encode("utf-8")
        else:
            payload_bytes = str(payload).encode("utf-8")

        # Update local state
        try:
            self._state[topic] = json.loads(payload_bytes.decode("utf-8"))
        except Exception:
            self._state[topic] = payload

        if not self.is_connected:
            logger.debug(f"MQTT client not connected to broker; updated local state for {topic}")
            return

        if qos == 0:
            packet = PublishPacket(
                topic=topic,
                payload=payload_bytes,
                qos=0,
                retain=retain,
            )
            await self._send_packet(packet)
            return

        # QoS 1 or QoS 2 requires packet identifier and tracking
        pid = self._get_next_packet_id()
        packet = PublishPacket(
            topic=topic,
            payload=payload_bytes,
            qos=qos,
            packet_id=pid,
            retain=retain,
        )

        loop = asyncio.get_running_loop()
        ack_future: asyncio.Future[None] = loop.create_future()
        self._inflight_publishes[pid] = ack_future
        self._inflight_packets[pid] = packet

        await self._send_packet(packet)

        try:
            await asyncio.wait_for(ack_future, timeout=timeout)
        except TimeoutError:
            logger.warning(f"Publish QoS {qos} timed out for packet ID {pid} on topic {topic}")
            raise

    async def subscribe(
        self,
        topic: str,
        callback: Callable[[str, Any], Awaitable[None] | None],
        qos: int = 0,
        timeout: float = 10.0,
    ) -> None:
        """Subscribe to a topic filter with callback and requested QoS."""
        if topic not in self._subscriptions:
            self._subscriptions[topic] = []
        if callback not in self._subscriptions[topic]:
            self._subscriptions[topic].append(callback)
        self._subscriptions_qos[topic] = qos

        if self.is_connected:
            pid = self._get_next_packet_id()
            sub_pkt = SubscribePacket(packet_id=pid, topics=[(topic, qos)])
            loop = asyncio.get_running_loop()
            sub_fut: asyncio.Future[list[int]] = loop.create_future()
            self._inflight_subscribes[pid] = sub_fut

            await self._send_packet(sub_pkt)
            try:
                await asyncio.wait_for(sub_fut, timeout=timeout)
            except TimeoutError:
                logger.warning(f"Subscribe timed out for topic {topic}")
                raise

    async def unsubscribe(self, topic: str, timeout: float = 10.0) -> None:
        """Unsubscribe from a topic filter."""
        self._subscriptions.pop(topic, None)
        self._subscriptions_qos.pop(topic, None)

        if self.is_connected:
            pid = self._get_next_packet_id()
            unsub_pkt = UnsubscribePacket(packet_id=pid, topics=[topic])
            loop = asyncio.get_running_loop()
            unsub_fut: asyncio.Future[None] = loop.create_future()
            self._inflight_unsubscribes[pid] = unsub_fut

            await self._send_packet(unsub_pkt)
            try:
                await asyncio.wait_for(unsub_fut, timeout=timeout)
            except TimeoutError:
                logger.warning(f"Unsubscribe timed out for topic {topic}")
                raise

    async def execute(self, command: str, params: dict[str, Any]) -> Any:
        """Execute a device command against the MQTT adapter."""
        if command == "publish":
            topic = params.get("topic")
            if not topic:
                raise ValueError("execute 'publish' requires 'topic' in params")
            payload = params.get("payload", {})
            qos = params.get("qos", 0)
            retain = params.get("retain", False)
            await self.publish(topic, payload, qos=qos, retain=retain)
            return {"status": "success", "command": command, "topic": topic}

        elif command == "subscribe":
            topic = params.get("topic")
            if not topic:
                raise ValueError("execute 'subscribe' requires 'topic' in params")

            # Default state recorder callback
            def _cb(t: str, p: Any) -> None:
                self._state[t] = p

            await self.subscribe(topic, _cb)
            return {"status": "success", "command": command, "topic": topic}

        elif command == "read_state":
            state = await self.read_state()
            return {"status": "success", "command": command, "state": state}

        else:
            topic = params.get("topic")
            payload = params.get("payload", {})
            if topic:
                await self.publish(topic, payload)
                return {"status": "success", "command": command, "topic": topic}
            return {"status": "unknown_command", "command": command}

    async def read_state(self) -> dict[str, Any]:
        """Read currently recorded device state dictionary."""
        return dict(self._state)

is_connected property

Return True if currently connected to MQTT broker.

connect() async

Establish connection to MQTT broker.

Source code in src/effero/adapters/mqtt_matter/client.py

async def connect(self) -> None:
    """Establish connection to MQTT broker."""
    self._manual_disconnect = False
    await self._establish_connection()

disconnect() async

Gracefully disconnect from MQTT broker.

Source code in src/effero/adapters/mqtt_matter/client.py

async def disconnect(self) -> None:
    """Gracefully disconnect from MQTT broker."""
    self._manual_disconnect = True
    self._connected = False

    if self._keepalive_task and not self._keepalive_task.done():
        self._keepalive_task.cancel()
    if self._reconnect_task and not self._reconnect_task.done():
        self._reconnect_task.cancel()

    if self._writer and not self._writer.is_closing():
        try:
            await self._send_packet(DisconnectPacket())
        except Exception as e:
            logger.debug(f"Error sending DISCONNECT: {e}")
        try:
            self._writer.close()
            await self._writer.wait_closed()
        except Exception as e:
            logger.debug(f"Error closing writer: {e}")

    if self._reader_task and not self._reader_task.done():
        self._reader_task.cancel()

    self._reader = None
    self._writer = None
    logger.info("Disconnected from MQTT broker")

publish(topic, payload, qos=0, retain=False, timeout=10.0) async

Publish a message to an MQTT topic with specified QoS (0, 1, or 2).

Source code in src/effero/adapters/mqtt_matter/client.py

async def publish(
    self,
    topic: str,
    payload: Any,
    qos: int = 0,
    retain: bool = False,
    timeout: float = 10.0,
) -> None:
    """Publish a message to an MQTT topic with specified QoS (0, 1, or 2)."""
    if isinstance(payload, bytes):
        payload_bytes = payload
    elif isinstance(payload, (dict, list)):
        payload_bytes = json.dumps(payload).encode("utf-8")
    else:
        payload_bytes = str(payload).encode("utf-8")

    # Update local state
    try:
        self._state[topic] = json.loads(payload_bytes.decode("utf-8"))
    except Exception:
        self._state[topic] = payload

    if not self.is_connected:
        logger.debug(f"MQTT client not connected to broker; updated local state for {topic}")
        return

    if qos == 0:
        packet = PublishPacket(
            topic=topic,
            payload=payload_bytes,
            qos=0,
            retain=retain,
        )
        await self._send_packet(packet)
        return

    # QoS 1 or QoS 2 requires packet identifier and tracking
    pid = self._get_next_packet_id()
    packet = PublishPacket(
        topic=topic,
        payload=payload_bytes,
        qos=qos,
        packet_id=pid,
        retain=retain,
    )

    loop = asyncio.get_running_loop()
    ack_future: asyncio.Future[None] = loop.create_future()
    self._inflight_publishes[pid] = ack_future
    self._inflight_packets[pid] = packet

    await self._send_packet(packet)

    try:
        await asyncio.wait_for(ack_future, timeout=timeout)
    except TimeoutError:
        logger.warning(f"Publish QoS {qos} timed out for packet ID {pid} on topic {topic}")
        raise

subscribe(topic, callback, qos=0, timeout=10.0) async

Subscribe to a topic filter with callback and requested QoS.

Source code in src/effero/adapters/mqtt_matter/client.py

async def subscribe(
    self,
    topic: str,
    callback: Callable[[str, Any], Awaitable[None] | None],
    qos: int = 0,
    timeout: float = 10.0,
) -> None:
    """Subscribe to a topic filter with callback and requested QoS."""
    if topic not in self._subscriptions:
        self._subscriptions[topic] = []
    if callback not in self._subscriptions[topic]:
        self._subscriptions[topic].append(callback)
    self._subscriptions_qos[topic] = qos

    if self.is_connected:
        pid = self._get_next_packet_id()
        sub_pkt = SubscribePacket(packet_id=pid, topics=[(topic, qos)])
        loop = asyncio.get_running_loop()
        sub_fut: asyncio.Future[list[int]] = loop.create_future()
        self._inflight_subscribes[pid] = sub_fut

        await self._send_packet(sub_pkt)
        try:
            await asyncio.wait_for(sub_fut, timeout=timeout)
        except TimeoutError:
            logger.warning(f"Subscribe timed out for topic {topic}")
            raise

unsubscribe(topic, timeout=10.0) async

Unsubscribe from a topic filter.

Source code in src/effero/adapters/mqtt_matter/client.py

async def unsubscribe(self, topic: str, timeout: float = 10.0) -> None:
    """Unsubscribe from a topic filter."""
    self._subscriptions.pop(topic, None)
    self._subscriptions_qos.pop(topic, None)

    if self.is_connected:
        pid = self._get_next_packet_id()
        unsub_pkt = UnsubscribePacket(packet_id=pid, topics=[topic])
        loop = asyncio.get_running_loop()
        unsub_fut: asyncio.Future[None] = loop.create_future()
        self._inflight_unsubscribes[pid] = unsub_fut

        await self._send_packet(unsub_pkt)
        try:
            await asyncio.wait_for(unsub_fut, timeout=timeout)
        except TimeoutError:
            logger.warning(f"Unsubscribe timed out for topic {topic}")
            raise

execute(command, params) async

Execute a device command against the MQTT adapter.

Source code in src/effero/adapters/mqtt_matter/client.py

async def execute(self, command: str, params: dict[str, Any]) -> Any:
    """Execute a device command against the MQTT adapter."""
    if command == "publish":
        topic = params.get("topic")
        if not topic:
            raise ValueError("execute 'publish' requires 'topic' in params")
        payload = params.get("payload", {})
        qos = params.get("qos", 0)
        retain = params.get("retain", False)
        await self.publish(topic, payload, qos=qos, retain=retain)
        return {"status": "success", "command": command, "topic": topic}

    elif command == "subscribe":
        topic = params.get("topic")
        if not topic:
            raise ValueError("execute 'subscribe' requires 'topic' in params")

        # Default state recorder callback
        def _cb(t: str, p: Any) -> None:
            self._state[t] = p

        await self.subscribe(topic, _cb)
        return {"status": "success", "command": command, "topic": topic}

    elif command == "read_state":
        state = await self.read_state()
        return {"status": "success", "command": command, "state": state}

    else:
        topic = params.get("topic")
        payload = params.get("payload", {})
        if topic:
            await self.publish(topic, payload)
            return {"status": "success", "command": command, "topic": topic}
        return {"status": "unknown_command", "command": command}

read_state() async

Read currently recorded device state dictionary.

Source code in src/effero/adapters/mqtt_matter/client.py

async def read_state(self) -> dict[str, Any]:
    """Read currently recorded device state dictionary."""
    return dict(self._state)

ROS2Bridge

Bases: DeviceAdapter

Real communication bridge between Effero and ROS 2 nodes over socket or serial streams.

Source code in src/effero/adapters/ros2/bridge.py

class ROS2Bridge(DeviceAdapter):
    """Real communication bridge between Effero and ROS 2 nodes over socket or serial streams."""

    def __init__(
        self,
        node_name: str = "effero_bridge",
        reader: asyncio.StreamReader | None = None,
        writer: asyncio.StreamWriter | None = None,
    ) -> None:
        self.node_name = node_name
        self._reader: asyncio.StreamReader | None = reader
        self._writer: asyncio.StreamWriter | None = writer
        self._connected: bool = bool(reader and writer)
        self._actions: dict[str, str] = {}
        self._topics: dict[str, str] = {}
        self._published_count: int = 0
        self._latest_joint_state: dict[str, Any] | None = None
        self._latest_twist: dict[str, Any] | None = None

    @property
    def is_connected(self) -> bool:
        return self._connected and self._writer is not None

    async def connect(
        self,
        host: str = "127.0.0.1",
        port: int = 9090,
    ) -> None:
        """Connect to real ROS 2 bridge socket daemon or verify injected stream."""
        if self._reader is not None and self._writer is not None:
            self._connected = True
            logger.info(f"ROS 2 node '{self.node_name}' stream connected")
            return

        try:
            self._reader, self._writer = await asyncio.open_connection(host, port)
            self._connected = True
            logger.info(f"ROS 2 node '{self.node_name}' connected to {host}:{port}")
        except Exception as err:
            self._connected = False
            logger.warning(f"Could not connect to ROS 2 bridge at {host}:{port}: {err}")
            raise RuntimeError(f"ROS 2 bridge connection failed: {err}") from err

    async def disconnect(self) -> None:
        """Disconnect and close transport streams."""
        if self._writer:
            self._writer.close()
            try:
                await self._writer.wait_closed()
            except Exception:
                pass
            self._writer = None
            self._reader = None
        self._connected = False
        logger.info(f"ROS 2 node '{self.node_name}' disconnected")

    async def publish_joint_state(self, joint_state: JointState) -> bytes:
        """Serialize JointState to CDR format and send to ROS 2 transport."""
        data = ROS2CDRSerializer.serialize_joint_state(joint_state)
        if self.is_connected and self._writer:
            # Send length-prefixed frame: 4 bytes uint32 length + CDR data
            self._writer.write(struct.pack(">I", len(data)) + data)
            await self._writer.drain()

        self._published_count += 1
        self._latest_joint_state = {
            "names": joint_state.name,
            "positions": joint_state.position,
            "velocities": joint_state.velocity,
        }
        return data

    async def publish_trajectory(self, trajectory: JointTrajectory) -> bytes:
        """Serialize JointTrajectory to CDR format and send to ROS 2 transport."""
        data = ROS2CDRSerializer.serialize_joint_trajectory(trajectory)
        if self.is_connected and self._writer:
            self._writer.write(struct.pack(">I", len(data)) + data)
            await self._writer.drain()

        self._published_count += 1
        return data

    async def publish_cmd_vel(self, twist: Twist) -> bytes:
        """Serialize Twist to CDR format and send to ROS 2 transport."""
        data = ROS2CDRSerializer.serialize_twist(twist)
        if self.is_connected and self._writer:
            self._writer.write(struct.pack(">I", len(data)) + data)
            await self._writer.drain()

        self._published_count += 1
        self._latest_twist = {
            "linear": {"x": twist.linear.x, "y": twist.linear.y, "z": twist.linear.z},
            "angular": {"x": twist.angular.x, "y": twist.angular.y, "z": twist.angular.z},
        }
        return data

    async def send_actuator_command(self, command: ActuatorCommand) -> bytes:
        """Serialize ActuatorCommand to CDR format and send to ROS 2 transport."""
        data = ROS2CDRSerializer.serialize_actuator_command(command)
        if self.is_connected and self._writer:
            self._writer.write(struct.pack(">I", len(data)) + data)
            await self._writer.drain()

        self._published_count += 1
        return data

    async def execute(self, command: str, params: dict[str, Any]) -> Any:
        """Execute ROS 2 action or command."""
        if command == "publish_joint_state":
            js = JointState(
                header=Header(frame_id=params.get("frame_id", "base_link")),
                name=params.get("names", []),
                position=params.get("positions", []),
                velocity=params.get("velocities", []),
                effort=params.get("effort", []),
            )
            raw_cdr = await self.publish_joint_state(js)
            return {
                "status": "success",
                "command": command,
                "bytes_serialized": len(raw_cdr),
                "joint_names": js.name,
            }

        elif command == "actuator_command":
            cmd = ActuatorCommand(
                actuator_id=params.get("actuator_id", 1),
                command_type=params.get("command_type", "position"),
                target_position=params.get("position", 0.0),
                target_velocity=params.get("velocity", 0.0),
                max_torque=params.get("max_torque", 1.0),
            )
            raw_cdr = await self.send_actuator_command(cmd)
            return {
                "status": "success",
                "command": command,
                "actuator_id": cmd.actuator_id,
                "bytes_serialized": len(raw_cdr),
            }

        elif command in ("cmd_vel", "publish_twist"):
            lin = params.get("linear", {})
            ang = params.get("angular", {})
            twist = Twist(
                linear=Vector3(
                    x=float(lin.get("x", params.get("linear_x", 0.0))),
                    y=float(lin.get("y", params.get("linear_y", 0.0))),
                    z=float(lin.get("z", params.get("linear_z", 0.0))),
                ),
                angular=Vector3(
                    x=float(ang.get("x", params.get("angular_x", 0.0))),
                    y=float(ang.get("y", params.get("angular_y", 0.0))),
                    z=float(ang.get("z", params.get("angular_z", 0.0))),
                ),
            )
            raw_cdr = await self.publish_cmd_vel(twist)
            return {
                "status": "success",
                "command": command,
                "bytes_serialized": len(raw_cdr),
                "linear": {"x": twist.linear.x, "y": twist.linear.y, "z": twist.linear.z},
                "angular": {"x": twist.angular.x, "y": twist.angular.y, "z": twist.angular.z},
            }

        elif command in self._actions:
            skill_target = self._actions[command]
            return {
                "status": "success",
                "action": command,
                "routed_skill": skill_target,
                "params": params,
            }

        else:
            raise ValueError(f"Unknown or unsupported ROS 2 command: '{command}'")

    async def read_state(self) -> dict[str, Any]:
        """Read state of ROS 2 node."""
        return {
            "node": self.node_name,
            "connected": self.is_connected,
            "published_count": self._published_count,
            "latest_joint_state": self._latest_joint_state,
            "latest_twist": self._latest_twist,
            "exposed_actions": list(self._actions.keys()),
            "exposed_topics": list(self._topics.keys()),
        }

    def expose_action(self, action_name: str, skill_name: str) -> None:
        self._actions[action_name] = skill_name
        logger.info(f"Exposed ROS 2 action {action_name} as skill {skill_name}")

    def expose_topic(self, topic_name: str, event_name: str) -> None:
        self._topics[topic_name] = event_name
        logger.info(f"Exposed ROS 2 topic {topic_name} as event {event_name}")

connect(host='127.0.0.1', port=9090) async

Connect to real ROS 2 bridge socket daemon or verify injected stream.

Source code in src/effero/adapters/ros2/bridge.py

async def connect(
    self,
    host: str = "127.0.0.1",
    port: int = 9090,
) -> None:
    """Connect to real ROS 2 bridge socket daemon or verify injected stream."""
    if self._reader is not None and self._writer is not None:
        self._connected = True
        logger.info(f"ROS 2 node '{self.node_name}' stream connected")
        return

    try:
        self._reader, self._writer = await asyncio.open_connection(host, port)
        self._connected = True
        logger.info(f"ROS 2 node '{self.node_name}' connected to {host}:{port}")
    except Exception as err:
        self._connected = False
        logger.warning(f"Could not connect to ROS 2 bridge at {host}:{port}: {err}")
        raise RuntimeError(f"ROS 2 bridge connection failed: {err}") from err

disconnect() async

Disconnect and close transport streams.

Source code in src/effero/adapters/ros2/bridge.py

async def disconnect(self) -> None:
    """Disconnect and close transport streams."""
    if self._writer:
        self._writer.close()
        try:
            await self._writer.wait_closed()
        except Exception:
            pass
        self._writer = None
        self._reader = None
    self._connected = False
    logger.info(f"ROS 2 node '{self.node_name}' disconnected")

publish_joint_state(joint_state) async

Serialize JointState to CDR format and send to ROS 2 transport.

Source code in src/effero/adapters/ros2/bridge.py

async def publish_joint_state(self, joint_state: JointState) -> bytes:
    """Serialize JointState to CDR format and send to ROS 2 transport."""
    data = ROS2CDRSerializer.serialize_joint_state(joint_state)
    if self.is_connected and self._writer:
        # Send length-prefixed frame: 4 bytes uint32 length + CDR data
        self._writer.write(struct.pack(">I", len(data)) + data)
        await self._writer.drain()

    self._published_count += 1
    self._latest_joint_state = {
        "names": joint_state.name,
        "positions": joint_state.position,
        "velocities": joint_state.velocity,
    }
    return data

publish_trajectory(trajectory) async

Serialize JointTrajectory to CDR format and send to ROS 2 transport.

Source code in src/effero/adapters/ros2/bridge.py

async def publish_trajectory(self, trajectory: JointTrajectory) -> bytes:
    """Serialize JointTrajectory to CDR format and send to ROS 2 transport."""
    data = ROS2CDRSerializer.serialize_joint_trajectory(trajectory)
    if self.is_connected and self._writer:
        self._writer.write(struct.pack(">I", len(data)) + data)
        await self._writer.drain()

    self._published_count += 1
    return data

publish_cmd_vel(twist) async

Serialize Twist to CDR format and send to ROS 2 transport.

Source code in src/effero/adapters/ros2/bridge.py

async def publish_cmd_vel(self, twist: Twist) -> bytes:
    """Serialize Twist to CDR format and send to ROS 2 transport."""
    data = ROS2CDRSerializer.serialize_twist(twist)
    if self.is_connected and self._writer:
        self._writer.write(struct.pack(">I", len(data)) + data)
        await self._writer.drain()

    self._published_count += 1
    self._latest_twist = {
        "linear": {"x": twist.linear.x, "y": twist.linear.y, "z": twist.linear.z},
        "angular": {"x": twist.angular.x, "y": twist.angular.y, "z": twist.angular.z},
    }
    return data

send_actuator_command(command) async

Serialize ActuatorCommand to CDR format and send to ROS 2 transport.

Source code in src/effero/adapters/ros2/bridge.py

async def send_actuator_command(self, command: ActuatorCommand) -> bytes:
    """Serialize ActuatorCommand to CDR format and send to ROS 2 transport."""
    data = ROS2CDRSerializer.serialize_actuator_command(command)
    if self.is_connected and self._writer:
        self._writer.write(struct.pack(">I", len(data)) + data)
        await self._writer.drain()

    self._published_count += 1
    return data

execute(command, params) async

Execute ROS 2 action or command.

Source code in src/effero/adapters/ros2/bridge.py

async def execute(self, command: str, params: dict[str, Any]) -> Any:
    """Execute ROS 2 action or command."""
    if command == "publish_joint_state":
        js = JointState(
            header=Header(frame_id=params.get("frame_id", "base_link")),
            name=params.get("names", []),
            position=params.get("positions", []),
            velocity=params.get("velocities", []),
            effort=params.get("effort", []),
        )
        raw_cdr = await self.publish_joint_state(js)
        return {
            "status": "success",
            "command": command,
            "bytes_serialized": len(raw_cdr),
            "joint_names": js.name,
        }

    elif command == "actuator_command":
        cmd = ActuatorCommand(
            actuator_id=params.get("actuator_id", 1),
            command_type=params.get("command_type", "position"),
            target_position=params.get("position", 0.0),
            target_velocity=params.get("velocity", 0.0),
            max_torque=params.get("max_torque", 1.0),
        )
        raw_cdr = await self.send_actuator_command(cmd)
        return {
            "status": "success",
            "command": command,
            "actuator_id": cmd.actuator_id,
            "bytes_serialized": len(raw_cdr),
        }

    elif command in ("cmd_vel", "publish_twist"):
        lin = params.get("linear", {})
        ang = params.get("angular", {})
        twist = Twist(
            linear=Vector3(
                x=float(lin.get("x", params.get("linear_x", 0.0))),
                y=float(lin.get("y", params.get("linear_y", 0.0))),
                z=float(lin.get("z", params.get("linear_z", 0.0))),
            ),
            angular=Vector3(
                x=float(ang.get("x", params.get("angular_x", 0.0))),
                y=float(ang.get("y", params.get("angular_y", 0.0))),
                z=float(ang.get("z", params.get("angular_z", 0.0))),
            ),
        )
        raw_cdr = await self.publish_cmd_vel(twist)
        return {
            "status": "success",
            "command": command,
            "bytes_serialized": len(raw_cdr),
            "linear": {"x": twist.linear.x, "y": twist.linear.y, "z": twist.linear.z},
            "angular": {"x": twist.angular.x, "y": twist.angular.y, "z": twist.angular.z},
        }

    elif command in self._actions:
        skill_target = self._actions[command]
        return {
            "status": "success",
            "action": command,
            "routed_skill": skill_target,
            "params": params,
        }

    else:
        raise ValueError(f"Unknown or unsupported ROS 2 command: '{command}'")

read_state() async

Read state of ROS 2 node.

Source code in src/effero/adapters/ros2/bridge.py

async def read_state(self) -> dict[str, Any]:
    """Read state of ROS 2 node."""
    return {
        "node": self.node_name,
        "connected": self.is_connected,
        "published_count": self._published_count,
        "latest_joint_state": self._latest_joint_state,
        "latest_twist": self._latest_twist,
        "exposed_actions": list(self._actions.keys()),
        "exposed_topics": list(self._topics.keys()),
    }

SerialAdapter

Bases: DeviceAdapter

Serial port adapter for communicating with microcontrollers, servos, and PLC devices.

Supports genuine binary packet framing (Dynamixel Protocol 2.0, Modbus RTU, and framed streaming).

Source code in src/effero/adapters/serial_gpio/client.py

class SerialAdapter(DeviceAdapter):
    """Serial port adapter for communicating with microcontrollers, servos, and PLC devices.

    Supports genuine binary packet framing (Dynamixel Protocol 2.0, Modbus RTU, and framed streaming).
    """

    def __init__(
        self,
        port: str,
        baudrate: int = 9600,
        reader: asyncio.StreamReader | None = None,
        writer: asyncio.StreamWriter | None = None,
    ) -> None:
        self.port = port
        self.baudrate = baudrate
        self._reader: asyncio.StreamReader | None = reader
        self._writer: asyncio.StreamWriter | None = writer
        self._connected = bool(reader and writer)
        self._last_state: dict[str, Any] = {}
        self._protocol_parser = SerialFramedProtocol()

    @property
    def is_connected(self) -> bool:
        return self._connected and self._writer is not None

    async def connect(self) -> None:
        """Open physical serial connection or verify existing stream."""
        if self._reader is not None and self._writer is not None:
            self._connected = True
            logger.info(f"Using provided stream connection for {self.port}")
            return

        if not HAS_SERIAL:
            raise RuntimeError(f"pyserial-asyncio is required to open serial port {self.port}")

        self._reader, self._writer = await serial_asyncio.open_serial_connection(
            url=self.port,
            baudrate=self.baudrate,
        )
        self._connected = True
        logger.info(f"Connected to serial port {self.port} at {self.baudrate} baud")

    async def disconnect(self) -> None:
        """Close serial connection."""
        if self._writer:
            self._writer.close()
            try:
                await self._writer.wait_closed()
            except Exception:
                pass
            self._reader = None
            self._writer = None
            self._connected = False
            logger.info(f"Disconnected from serial port {self.port}")

    async def send_raw(self, data: bytes) -> None:
        """Send raw binary bytes to the serial transport."""
        if not self.is_connected or self._writer is None:
            raise RuntimeError(f"Serial port {self.port} is not connected")
        self._writer.write(data)
        await self._writer.drain()

    async def read_raw(self, nbytes: int, timeout: float = 2.0) -> bytes:
        """Read exactly nbytes from the serial transport."""
        if not self.is_connected or self._reader is None:
            raise RuntimeError(f"Serial port {self.port} is not connected")
        return await asyncio.wait_for(self._reader.readexactly(nbytes), timeout=timeout)

    async def send_dynamixel(
        self,
        packet_id: int,
        instruction: int,
        parameters: bytes = b"",
        response_timeout: float = 2.0,
    ) -> DynamixelStatus:
        """Send Dynamixel Protocol 2.0 instruction packet and await status response."""
        packet = DynamixelPacket.build_instruction_packet(packet_id, instruction, parameters)
        await self.send_raw(packet)

        # Read status response: Header(4) + ID(1) + Length(2)
        if self._reader is None:
            raise RuntimeError(f"Serial port {self.port} is not connected")

        header_bytes = await asyncio.wait_for(self._reader.readexactly(7), timeout=response_timeout)
        length = int.from_bytes(header_bytes[5:7], byteorder="little")
        body_bytes = await asyncio.wait_for(
            self._reader.readexactly(length),
            timeout=response_timeout,
        )
        full_status_pkt = header_bytes + body_bytes
        status = DynamixelPacket.parse_status_packet(full_status_pkt)
        self._last_state[f"dynamixel_{packet_id}"] = {
            "error_code": status.error_code,
            "has_error": status.has_error,
            "parameters_len": len(status.parameters),
        }
        return status

    async def send_modbus(
        self,
        slave_address: int,
        function_code: int,
        data: bytes,
        response_len: int,
        response_timeout: float = 2.0,
    ) -> ModbusResponse:
        """Send Modbus RTU frame and parse response frame."""
        frame = ModbusPacket.build_request(slave_address, function_code, data)
        await self.send_raw(frame)

        if self._reader is None:
            raise RuntimeError(f"Serial port {self.port} is not connected")

        resp_bytes = await asyncio.wait_for(
            self._reader.readexactly(response_len),
            timeout=response_timeout,
        )
        response = ModbusPacket.parse_response(resp_bytes)
        self._last_state[f"modbus_{slave_address}"] = {
            "function": response.function_code,
            "is_exception": response.is_exception,
            "registers": response.registers,
        }
        return response

    async def execute(self, command: str, params: dict[str, Any]) -> Any:
        """Execute a high-level actuator or protocol command over serial."""
        if not self.is_connected or self._writer is None or self._reader is None:
            raise RuntimeError(f"Serial port {self.port} is not connected")

        if command == "dynamixel_ping":
            servo_id = params.get("id", 1)
            status = await self.send_dynamixel(servo_id, 0x01)
            return {
                "status": "success",
                "servo_id": status.packet_id,
                "error_code": status.error_code,
                "has_error": status.has_error,
            }

        elif command == "modbus_read_holding":
            slave = params.get("slave", 1)
            start_addr = params.get("address", 0)
            qty = params.get("quantity", 1)
            req_data = ModbusPacket.build_read_holding_registers(slave, start_addr, qty)
            await self.send_raw(req_data)
            # Response length: Slave(1) + Func(1) + ByteCount(1) + qty*2 + CRC(2) = 5 + 2*qty
            expected_len = 5 + 2 * qty
            resp_bytes = await self.read_raw(expected_len)
            modbus_resp = ModbusPacket.parse_response(resp_bytes)
            return {
                "status": "success",
                "slave": modbus_resp.slave_address,
                "registers": modbus_resp.registers,
            }

        elif command == "send_framed":
            seq = params.get("sequence", 0)
            cmd_id = params.get("cmd_id", 1)
            payload = params.get("payload", b"")
            if isinstance(payload, str):
                payload = payload.encode("utf-8")
            frame = SerialFramedProtocol.encode_frame(seq, cmd_id, payload)
            await self.send_raw(frame)
            return {
                "status": "success",
                "sequence": seq,
                "cmd_id": cmd_id,
                "bytes_sent": len(frame),
            }

        else:
            # Standard framed JSON communication
            msg = {"command": command, "params": params}
            encoded = (json.dumps(msg) + "\n").encode()
            self._writer.write(encoded)
            await self._writer.drain()

            line = await self._reader.readline()
            try:
                response = json.loads(line.decode().strip())
                self._last_state.update(response)
                return response
            except json.JSONDecodeError:
                return {"status": "error", "raw": line.decode().strip()}

    async def read_state(self) -> dict[str, Any]:
        return dict(self._last_state)

connect() async

Open physical serial connection or verify existing stream.

Source code in src/effero/adapters/serial_gpio/client.py

async def connect(self) -> None:
    """Open physical serial connection or verify existing stream."""
    if self._reader is not None and self._writer is not None:
        self._connected = True
        logger.info(f"Using provided stream connection for {self.port}")
        return

    if not HAS_SERIAL:
        raise RuntimeError(f"pyserial-asyncio is required to open serial port {self.port}")

    self._reader, self._writer = await serial_asyncio.open_serial_connection(
        url=self.port,
        baudrate=self.baudrate,
    )
    self._connected = True
    logger.info(f"Connected to serial port {self.port} at {self.baudrate} baud")

disconnect() async

Close serial connection.

Source code in src/effero/adapters/serial_gpio/client.py

async def disconnect(self) -> None:
    """Close serial connection."""
    if self._writer:
        self._writer.close()
        try:
            await self._writer.wait_closed()
        except Exception:
            pass
        self._reader = None
        self._writer = None
        self._connected = False
        logger.info(f"Disconnected from serial port {self.port}")

send_raw(data) async

Send raw binary bytes to the serial transport.

Source code in src/effero/adapters/serial_gpio/client.py

async def send_raw(self, data: bytes) -> None:
    """Send raw binary bytes to the serial transport."""
    if not self.is_connected or self._writer is None:
        raise RuntimeError(f"Serial port {self.port} is not connected")
    self._writer.write(data)
    await self._writer.drain()

read_raw(nbytes, timeout=2.0) async

Read exactly nbytes from the serial transport.

Source code in src/effero/adapters/serial_gpio/client.py

async def read_raw(self, nbytes: int, timeout: float = 2.0) -> bytes:
    """Read exactly nbytes from the serial transport."""
    if not self.is_connected or self._reader is None:
        raise RuntimeError(f"Serial port {self.port} is not connected")
    return await asyncio.wait_for(self._reader.readexactly(nbytes), timeout=timeout)

send_dynamixel(packet_id, instruction, parameters=b'', response_timeout=2.0) async

Send Dynamixel Protocol 2.0 instruction packet and await status response.

Source code in src/effero/adapters/serial_gpio/client.py

async def send_dynamixel(
    self,
    packet_id: int,
    instruction: int,
    parameters: bytes = b"",
    response_timeout: float = 2.0,
) -> DynamixelStatus:
    """Send Dynamixel Protocol 2.0 instruction packet and await status response."""
    packet = DynamixelPacket.build_instruction_packet(packet_id, instruction, parameters)
    await self.send_raw(packet)

    # Read status response: Header(4) + ID(1) + Length(2)
    if self._reader is None:
        raise RuntimeError(f"Serial port {self.port} is not connected")

    header_bytes = await asyncio.wait_for(self._reader.readexactly(7), timeout=response_timeout)
    length = int.from_bytes(header_bytes[5:7], byteorder="little")
    body_bytes = await asyncio.wait_for(
        self._reader.readexactly(length),
        timeout=response_timeout,
    )
    full_status_pkt = header_bytes + body_bytes
    status = DynamixelPacket.parse_status_packet(full_status_pkt)
    self._last_state[f"dynamixel_{packet_id}"] = {
        "error_code": status.error_code,
        "has_error": status.has_error,
        "parameters_len": len(status.parameters),
    }
    return status

send_modbus(slave_address, function_code, data, response_len, response_timeout=2.0) async

Send Modbus RTU frame and parse response frame.

Source code in src/effero/adapters/serial_gpio/client.py

async def send_modbus(
    self,
    slave_address: int,
    function_code: int,
    data: bytes,
    response_len: int,
    response_timeout: float = 2.0,
) -> ModbusResponse:
    """Send Modbus RTU frame and parse response frame."""
    frame = ModbusPacket.build_request(slave_address, function_code, data)
    await self.send_raw(frame)

    if self._reader is None:
        raise RuntimeError(f"Serial port {self.port} is not connected")

    resp_bytes = await asyncio.wait_for(
        self._reader.readexactly(response_len),
        timeout=response_timeout,
    )
    response = ModbusPacket.parse_response(resp_bytes)
    self._last_state[f"modbus_{slave_address}"] = {
        "function": response.function_code,
        "is_exception": response.is_exception,
        "registers": response.registers,
    }
    return response

execute(command, params) async

Execute a high-level actuator or protocol command over serial.

Source code in src/effero/adapters/serial_gpio/client.py

async def execute(self, command: str, params: dict[str, Any]) -> Any:
    """Execute a high-level actuator or protocol command over serial."""
    if not self.is_connected or self._writer is None or self._reader is None:
        raise RuntimeError(f"Serial port {self.port} is not connected")

    if command == "dynamixel_ping":
        servo_id = params.get("id", 1)
        status = await self.send_dynamixel(servo_id, 0x01)
        return {
            "status": "success",
            "servo_id": status.packet_id,
            "error_code": status.error_code,
            "has_error": status.has_error,
        }

    elif command == "modbus_read_holding":
        slave = params.get("slave", 1)
        start_addr = params.get("address", 0)
        qty = params.get("quantity", 1)
        req_data = ModbusPacket.build_read_holding_registers(slave, start_addr, qty)
        await self.send_raw(req_data)
        # Response length: Slave(1) + Func(1) + ByteCount(1) + qty*2 + CRC(2) = 5 + 2*qty
        expected_len = 5 + 2 * qty
        resp_bytes = await self.read_raw(expected_len)
        modbus_resp = ModbusPacket.parse_response(resp_bytes)
        return {
            "status": "success",
            "slave": modbus_resp.slave_address,
            "registers": modbus_resp.registers,
        }

    elif command == "send_framed":
        seq = params.get("sequence", 0)
        cmd_id = params.get("cmd_id", 1)
        payload = params.get("payload", b"")
        if isinstance(payload, str):
            payload = payload.encode("utf-8")
        frame = SerialFramedProtocol.encode_frame(seq, cmd_id, payload)
        await self.send_raw(frame)
        return {
            "status": "success",
            "sequence": seq,
            "cmd_id": cmd_id,
            "bytes_sent": len(frame),
        }

    else:
        # Standard framed JSON communication
        msg = {"command": command, "params": params}
        encoded = (json.dumps(msg) + "\n").encode()
        self._writer.write(encoded)
        await self._writer.drain()

        line = await self._reader.readline()
        try:
            response = json.loads(line.decode().strip())
            self._last_state.update(response)
            return response
        except json.JSONDecodeError:
            return {"status": "error", "raw": line.decode().strip()}

get_default_client(config=None)

Return the global default MQTT client instance, optionally configured from IoTConfig.

Source code in src/effero/adapters/mqtt_matter/client.py

def get_default_client(config: Any | None = None) -> MQTTAdapter:
    """Return the global default MQTT client instance, optionally configured from IoTConfig."""
    global _default_client
    if _default_client is None:
        if config is not None:
            _default_client = MQTTAdapter(
                broker_host=getattr(config, "broker_host", "localhost"),
                broker_port=getattr(config, "broker_port", 1883),
                username=getattr(config, "username", None),
                password=getattr(config, "password", None),
                client_id=getattr(config, "client_id", None),
                keep_alive=getattr(config, "keepalive", 60),
                use_tls=getattr(config, "use_tls", False),
                ca_certs=getattr(config, "ca_certs", None),
                certfile=getattr(config, "certfile", None),
                keyfile=getattr(config, "keyfile", None),
                tls_insecure=getattr(config, "tls_insecure", False),
            )
        else:
            _default_client = MQTTAdapter()
    return _default_client