p4net.control

P4Runtime control plane: gRPC client, P4Info index, value codecs.

This package depends on the p4runtime PyPI distribution for the generated proto stubs. Those stubs are descriptor-incompatible with the C++ google.protobuf runtime shipped in protobuf 4.x and later, so we set PROTOCOL_BUFFERS_PYTHON_IMPLEMENTATION=python here, before any p4 stubs are imported transitively. This is the upstream-recommended workaround until the p4runtime distribution is rebuilt against modern protoc.

__all__ module-attribute

__all__ = ['AsyncOperationCancelledError', 'AsyncP4RuntimeClient', 'ConnectionError', 'CounterData', 'DuplicateEntryError', 'EncodingError', 'EntryNotFoundError', 'NoSuchActionError', 'NoSuchFieldError', 'NoSuchRegisterError', 'NoSuchTableError', 'NotPrimaryError', 'P4InfoIndex', 'P4RuntimeClient', 'P4RuntimeError', 'PipelineError', 'RegisterSpec', 'canonicalize', 'decode_int', 'decode_ipv4', 'decode_ipv6', 'decode_mac', 'encode_int', 'encode_ipv4', 'encode_mac', 'encode_value', 'format_exact', 'format_lpm', 'format_range', 'format_ternary', 'parse_lpm', 'parse_range', 'parse_ternary']

AsyncP4RuntimeClient

Async parallel to :class:P4RuntimeClient. Stable in 1.x since 1.7.0.

Source code in src/p4net/control/async_client.py

class AsyncP4RuntimeClient:
    """Async parallel to :class:`P4RuntimeClient`. **Stable** in 1.x since 1.7.0."""

    def __init__(
        self,
        grpc_address: tuple[str, int],
        device_id: int,
        info_index: P4InfoIndex | None = None,
        thrift_address: tuple[str, int] | None = None,
        election_id: tuple[int, int] | None = None,
    ) -> None:
        self._host, self._port = grpc_address
        self._device_id = int(device_id)
        self._info_index = info_index
        self._thrift_address = thrift_address
        self._election_id = election_id  # resolved at connect() if None

        self._channel: Any = None
        self._stub: Any = None
        self._outgoing: asyncio.Queue[Any] | None = None
        self._stream_call: Any = None
        self._stream_task: asyncio.Task[None] | None = None
        self._arbitration_event: asyncio.Event | None = None
        self._arbitration: Any = None
        self._stream_error: BaseException | None = None
        self._packet_in_queue: asyncio.Queue[tuple[bytes, dict[str, int]]] | None = None
        self._packet_handlers: list[Callable[[bytes, dict[str, int]], Awaitable[None]]] = []
        self._handlers_lock = asyncio.Lock()
        self._connected = False
        self._is_primary = False
        self._closed = False

    # ----------------------------------------------------------------- properties

    @property
    def grpc_address(self) -> tuple[str, int]:
        return (self._host, self._port)

    @property
    def target(self) -> str:
        return f"{self._host}:{self._port}"

    @property
    def device_id(self) -> int:
        return self._device_id

    @property
    def election_id(self) -> tuple[int, int]:
        if self._election_id is None:
            raise P4RuntimeError("election_id not yet assigned (connect() pending)")
        return self._election_id

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

    @property
    def is_primary(self) -> bool:
        return self._is_primary

    @property
    def info_index(self) -> P4InfoIndex:
        if self._info_index is None:
            raise P4RuntimeError(
                "no P4Info index attached; push a pipeline first or pass info_index= to ctor"
            )
        return self._info_index

    # ----------------------------------------------------------------- lifecycle

    async def connect(self, *, timeout: float = 10.0) -> None:
        if self._connected:
            return
        self._closed = False
        self._channel = grpc.aio.insecure_channel(self.target)
        self._stub = p4runtime_pb2_grpc.P4RuntimeStub(self._channel)
        self._outgoing = asyncio.Queue()
        self._arbitration_event = asyncio.Event()
        self._arbitration = None
        self._stream_error = None
        self._packet_in_queue = asyncio.Queue()

        if self._election_id is None:
            millis = time.time_ns() // 1_000_000
            self._election_id = (int(millis), 0)

        self._stream_call = self._stub.StreamChannel(self._request_generator())
        self._stream_task = asyncio.create_task(
            self._stream_consumer(), name=f"p4rt-async-stream-{self._host}:{self._port}"
        )

        req = p4runtime_pb2.StreamMessageRequest()
        req.arbitration.device_id = self._device_id
        req.arbitration.election_id.high = self._election_id[0]
        req.arbitration.election_id.low = self._election_id[1]
        await self._outgoing.put(req)

        try:
            await asyncio.wait_for(self._arbitration_event.wait(), timeout=timeout)
        except asyncio.TimeoutError as exc:
            await self._teardown()
            raise ConnectionError(
                f"P4Runtime async arbitration timed out after {timeout}s for {self.target!r}"
            ) from exc
        if self._stream_error is not None:
            err = self._stream_error
            await self._teardown()
            raise P4RuntimeError(f"async stream error: {err!r}") from err
        if self._arbitration is None:
            await self._teardown()
            raise ConnectionError("no arbitration response received")
        status_code = int(self._arbitration.status.code)
        self._is_primary = status_code == 0
        if status_code != 0 and self._election_id != (0, 0):
            # Explicit secondary (election_id=(0,0)) is allowed and stays connected.
            # Any other non-primary status is a failure.
            await self._teardown()
            raise NotPrimaryError(
                f"async client is not primary for device {self._device_id} "
                f"(status code {status_code})"
            )
        self._connected = True
        logger.info(
            "AsyncP4RuntimeClient %s connected (primary=%s, election=%s)",
            self.target,
            self._is_primary,
            self._election_id,
        )

    async def disconnect(self) -> None:
        if self._closed:
            return
        await self._teardown()

    async def _teardown(self) -> None:
        self._closed = True
        was_connected = self._connected
        self._connected = False
        self._is_primary = False
        if self._outgoing is not None:
            with self._suppress_queue_full():
                self._outgoing.put_nowait(None)  # sentinel
        if self._stream_call is not None:
            try:
                self._stream_call.cancel()
            except Exception as exc:
                logger.debug("async stream cancel raised: %r", exc)
        if self._stream_task is not None and not self._stream_task.done():
            self._stream_task.cancel()
            try:
                await asyncio.shield(asyncio.wait_for(self._stream_task, timeout=2.0))
            except (asyncio.CancelledError, asyncio.TimeoutError):
                pass
            except Exception as exc:
                logger.debug("async stream task awaited with: %r", exc)
        if self._channel is not None:
            try:
                await self._channel.close()
            except Exception as exc:
                logger.debug("async channel close raised: %r", exc)
        self._channel = None
        self._stub = None
        self._stream_call = None
        self._stream_task = None
        self._outgoing = None
        self._packet_in_queue = None
        if was_connected:
            logger.info("AsyncP4RuntimeClient %s disconnected", self.target)

    @staticmethod
    def _suppress_queue_full() -> Any:
        return contextlib.suppress(asyncio.QueueFull)

    async def __aenter__(self) -> AsyncP4RuntimeClient:
        await self.connect()
        return self

    async def __aexit__(self, *exc_info: Any) -> None:
        await self.disconnect()

    # ----------------------------------------------------------------- stream

    async def _request_generator(self) -> AsyncIterator[Any]:
        assert self._outgoing is not None
        while True:
            msg = await self._outgoing.get()
            if msg is None:
                return
            yield msg

    async def _stream_consumer(self) -> None:
        try:
            assert self._stream_call is not None
            async for resp in self._stream_call:
                if resp.HasField("arbitration"):
                    self._arbitration = resp.arbitration
                    assert self._arbitration_event is not None
                    self._arbitration_event.set()
                elif resp.HasField("packet"):
                    await self._dispatch_packet_in(resp.packet)
        except asyncio.CancelledError:
            raise
        except grpc.aio.AioRpcError as exc:
            if not self._closed:
                self._stream_error = exc
                if self._arbitration_event is not None:
                    self._arbitration_event.set()
        except Exception as exc:
            if not self._closed:
                self._stream_error = exc
                if self._arbitration_event is not None:
                    self._arbitration_event.set()

    async def _dispatch_packet_in(self, packet: Any) -> None:
        payload = bytes(packet.payload)
        meta: dict[str, int] = {}
        if self._info_index is not None:
            try:
                meta = self._info_index.decode_packet_in_metadata(packet.metadata)
            except Exception as exc:
                logger.debug("async decode_packet_in_metadata raised: %r", exc)
        # Buffer for expect_packet_in callers.
        if self._packet_in_queue is not None:
            with self._suppress_queue_full():
                self._packet_in_queue.put_nowait((payload, meta))
        # Dispatch to async handlers.
        async with self._handlers_lock:
            handlers = list(self._packet_handlers)
        for h in handlers:
            try:
                await h(payload, meta)
            except Exception as exc:
                logger.warning("async packet_in handler %r raised: %r", h, exc)

    # ----------------------------------------------------------------- pipeline

    async def push_pipeline(
        self,
        p4info_bytes: bytes,
        json_bytes: bytes,
        *,
        action: str = "VERIFY_AND_COMMIT",
        timeout: float = 10.0,
    ) -> None:
        """Push a compiled pipeline (raw bytes)."""
        self._require_connected()
        if action not in _PIPELINE_ACTIONS:
            raise P4RuntimeError(f"invalid pipeline action {action!r}")
        msg = p4info_pb2.P4Info()
        text_format.Merge(p4info_bytes.decode("utf-8"), msg)
        req = p4runtime_pb2.SetForwardingPipelineConfigRequest()
        req.device_id = self._device_id
        req.election_id.high = self._election_id[0]  # type: ignore[index]
        req.election_id.low = self._election_id[1]  # type: ignore[index]
        req.action = p4runtime_pb2.SetForwardingPipelineConfigRequest.Action.Value(action)
        req.config.p4info.CopyFrom(msg)
        req.config.p4_device_config = bytes(json_bytes)
        try:
            await asyncio.wait_for(self._stub.SetForwardingPipelineConfig(req), timeout=timeout)
        except asyncio.CancelledError as exc:
            raise AsyncOperationCancelledError("push_pipeline cancelled") from exc
        except grpc.aio.AioRpcError as exc:
            raise self._translate_rpc_error(exc, pipeline=True) from exc
        self._info_index = P4InfoIndex(msg)

    # ----------------------------------------------------------------- tables

    async def insert_table_entry(
        self,
        table: str,
        match: Mapping[str, object],
        action: str,
        params: Mapping[str, object] | None = None,
        *,
        priority: int | None = None,
        timeout: float = 5.0,
    ) -> None:
        await self._write_entry(table, match, action, params, priority, "INSERT", timeout)

    async def modify_table_entry(
        self,
        table: str,
        match: Mapping[str, object],
        action: str,
        params: Mapping[str, object] | None = None,
        *,
        priority: int | None = None,
        timeout: float = 5.0,
    ) -> None:
        await self._write_entry(table, match, action, params, priority, "MODIFY", timeout)

    async def delete_table_entry(
        self,
        table: str,
        match: Mapping[str, object],
        *,
        priority: int | None = None,
        timeout: float = 5.0,
    ) -> None:
        await self._write_entry(table, match, None, None, priority, "DELETE", timeout)

    async def _write_entry(
        self,
        table: str,
        match: Mapping[str, object],
        action: str | None,
        params: Mapping[str, object] | None,
        priority: int | None,
        update_type: str,
        timeout: float,
    ) -> None:
        self._require_connected_with_index()
        idx = self.info_index
        table_id = idx.table_id(table)
        requires_priority = idx.table_requires_priority(table)
        if requires_priority and priority is None and update_type != "DELETE":
            raise EncodingError(f"table {table!r} has ternary/range fields; priority required")
        if not requires_priority and priority is not None:
            raise EncodingError(f"table {table!r} is exact/lpm-only; priority must be None")
        fms = idx.encode_match(table, match)
        entry = p4runtime_pb2.TableEntry()
        entry.table_id = table_id
        for fm in fms:
            entry.match.add().CopyFrom(fm)
        if action is not None:
            entry.action.action.CopyFrom(idx.encode_action(action, params))
        if priority is not None:
            entry.priority = int(priority)
        upd = p4runtime_pb2.Update()
        upd.type = p4runtime_pb2.Update.Type.Value(update_type)
        upd.entity.table_entry.CopyFrom(entry)
        req = p4runtime_pb2.WriteRequest()
        req.device_id = self._device_id
        req.election_id.high = self._election_id[0]  # type: ignore[index]
        req.election_id.low = self._election_id[1]  # type: ignore[index]
        req.updates.add().CopyFrom(upd)
        try:
            await asyncio.wait_for(self._stub.Write(req), timeout=timeout)
        except asyncio.CancelledError as exc:
            raise AsyncOperationCancelledError(f"{update_type} on {table!r} cancelled") from exc
        except grpc.aio.AioRpcError as exc:
            raise self._translate_rpc_error(exc) from exc

    def list_table_entries(
        self,
        table: str | None = None,
        *,
        timeout: float = 5.0,
    ) -> AsyncIterator[dict[str, Any]]:
        """Async iterator over decoded table entries.

        Note: this is a regular ``def`` returning an ``AsyncIterator``.
        Use it as ``async for entry in client.list_table_entries(...)``.
        """
        self._require_connected_with_index()
        idx = self.info_index
        req = p4runtime_pb2.ReadRequest()
        req.device_id = self._device_id
        entity = req.entities.add()
        if table is not None:
            entity.table_entry.table_id = idx.table_id(table)
        else:
            entity.table_entry.table_id = 0

        async def _gen() -> AsyncIterator[dict[str, Any]]:
            try:
                async for resp in self._stub.Read(req, timeout=timeout):
                    for ent in resp.entities:
                        if ent.HasField("table_entry"):
                            yield self._decode_table_entry(ent.table_entry)
            except asyncio.CancelledError:
                raise
            except grpc.aio.AioRpcError as exc:
                raise self._translate_rpc_error(exc) from exc

        return _gen()

    def _decode_table_entry(self, entry: Any) -> dict[str, Any]:
        idx = self.info_index
        table_name = idx.table_name(int(entry.table_id))
        table = None
        for t in idx.raw.tables:
            if t.preamble.id == entry.table_id:
                table = t
                break
        assert table is not None
        fields_by_id = {int(mf.id): mf for mf in table.match_fields}
        match: dict[str, Any] = {}
        for fm in entry.match:
            mf = fields_by_id.get(int(fm.field_id))
            if mf is None:
                continue
            if fm.HasField("exact"):
                match[mf.name] = bytes(fm.exact.value)
            elif fm.HasField("lpm"):
                match[mf.name] = (bytes(fm.lpm.value), int(fm.lpm.prefix_len))
            elif fm.HasField("ternary"):
                match[mf.name] = (bytes(fm.ternary.value), bytes(fm.ternary.mask))
            elif fm.HasField("range"):
                match[mf.name] = (bytes(fm.range.low), bytes(fm.range.high))
            elif fm.HasField("optional"):
                match[mf.name] = bytes(fm.optional.value)
        action_name: str | None = None
        params: dict[str, bytes] = {}
        if entry.HasField("action") and entry.action.HasField("action"):
            a = entry.action.action
            action_name = idx.action_name(int(a.action_id))
            for am in idx.raw.actions:
                if am.preamble.id == a.action_id:
                    params_by_id = {int(p.id): p for p in am.params}
                    for ap in a.params:
                        p = params_by_id.get(int(ap.param_id))
                        if p is not None:
                            params[p.name] = bytes(ap.value)
                    break
        return {
            "table": table_name,
            "match": match,
            "action": action_name,
            "params": params,
            "priority": int(entry.priority) if entry.priority else None,
        }

    # ----------------------------------------------------------------- counters

    async def read_counter(
        self,
        counter: str,
        index: int | None = None,
        *,
        timeout: float = 5.0,
    ) -> int | dict[int, int]:
        """Read indirect counter. Returns packet_count only for simplicity.

        Single-index returns the cell's packet_count (int). No index returns
        ``{cell_index: packet_count}``. The sync client returns the richer
        ``CounterData`` dataclass; the async API keeps things simple per spec.
        """
        self._require_connected_with_index()
        idx = self.info_index
        counter_id = idx.counter_id(counter)
        req = p4runtime_pb2.ReadRequest()
        req.device_id = self._device_id
        entity = req.entities.add()
        entity.counter_entry.counter_id = counter_id
        if index is not None:
            entity.counter_entry.index.index = int(index)
        collected: dict[int, int] = {}
        try:
            async for resp in self._stub.Read(req, timeout=timeout):
                for ent in resp.entities:
                    if not ent.HasField("counter_entry"):
                        continue
                    ce = ent.counter_entry
                    collected[int(ce.index.index)] = int(ce.data.packet_count)
        except asyncio.CancelledError as exc:
            raise AsyncOperationCancelledError("read_counter cancelled") from exc
        except grpc.aio.AioRpcError as exc:
            raise self._translate_rpc_error(exc) from exc
        if index is not None:
            return collected.get(int(index), 0)
        return collected

    # ----------------------------------------------------------------- registers

    async def write_register(
        self,
        name: str,
        index: int,
        value: int,
        *,
        timeout: float = 5.0,
    ) -> None:
        self._require_connected_with_index()
        idx = self.info_index
        spec = idx.register_by_name(name)
        if not 0 <= int(index) < spec.size:
            raise EncodingError(f"register {name!r} index {index} out of range [0, {spec.size})")
        encode_value(int(value), spec.bitwidth)  # pre-flight bitwidth check
        await self._run_thrift_cli(
            f"register_write {name} {int(index)} {int(value)}",
            timeout=timeout,
            op_description=f"register_write {name}[{index}]",
        )

    async def read_register(
        self,
        name: str,
        index: int | None = None,
        *,
        timeout: float = 5.0,
    ) -> int | list[int]:
        self._require_connected_with_index()
        idx = self.info_index
        spec = idx.register_by_name(name)
        if index is not None and not 0 <= int(index) < spec.size:
            raise EncodingError(f"register {name!r} index {index} out of range [0, {spec.size})")
        if index is not None:
            output = await self._run_thrift_cli(
                f"register_read {name} {int(index)}",
                timeout=timeout,
                op_description=f"register_read {name}[{index}]",
            )
            return _parse_register_read_single(output, name, int(index))
        output = await self._run_thrift_cli(
            f"register_read {name}",
            timeout=timeout,
            op_description=f"register_read {name}",
        )
        return _parse_register_read_array(output, name, spec.size)

    async def _run_thrift_cli(
        self,
        command: str,
        *,
        timeout: float,
        op_description: str,
    ) -> str:
        if self._thrift_address is None:
            raise P4RuntimeError(
                f"register operations require thrift_address=(host, port); op: {op_description}"
            )
        host, port = self._thrift_address
        if shutil.which("simple_switch_CLI") is None:
            raise P4RuntimeError("simple_switch_CLI not on PATH; required for register ops")
        try:
            proc = await asyncio.create_subprocess_exec(
                "simple_switch_CLI",
                "--thrift-ip",
                str(host),
                "--thrift-port",
                str(int(port)),
                stdin=asyncio.subprocess.PIPE,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
            )
        except FileNotFoundError as exc:
            raise P4RuntimeError("simple_switch_CLI not found") from exc
        try:
            stdout, stderr = await asyncio.wait_for(
                proc.communicate(input=(command + "\n").encode("utf-8")), timeout=timeout
            )
        except asyncio.TimeoutError as exc:
            proc.kill()
            await proc.wait()
            raise P4RuntimeError(f"{op_description} timed out after {timeout}s") from exc
        except asyncio.CancelledError as exc:
            proc.kill()
            await proc.wait()
            raise AsyncOperationCancelledError(f"{op_description} cancelled") from exc
        if proc.returncode != 0:
            raise P4RuntimeError(
                f"{op_description} failed (rc={proc.returncode}): "
                f"{(stdout + stderr).decode('utf-8', errors='replace').strip()}"
            )
        combined = (stdout + stderr).decode("utf-8", errors="replace")
        for line in combined.splitlines():
            stripped = line.strip()
            if stripped.startswith("Error:") or "Invalid" in stripped:
                raise P4RuntimeError(f"{op_description}: {stripped}")
        return stdout.decode("utf-8", errors="replace")

    # ----------------------------------------------------------------- packets

    async def send_packet_out(
        self,
        payload: bytes,
        metadata: Mapping[str, object] | None = None,
    ) -> None:
        self._require_connected()
        idx = self._info_index
        encoded = idx.encode_packet_out_metadata(metadata or {}) if idx is not None else []
        if metadata and idx is None:
            raise EncodingError("no pipeline set; cannot encode packet_out metadata")
        if not isinstance(payload, (bytes, bytearray)):
            raise EncodingError(f"payload must be bytes-like, got {type(payload).__name__}")
        req = p4runtime_pb2.StreamMessageRequest()
        req.packet.payload = bytes(payload)
        for pm in encoded:
            req.packet.metadata.add().CopyFrom(pm)
        out = self._outgoing
        if out is None:
            raise ConnectionError("client is not connected")
        await out.put(req)

    async def on_packet_in(
        self,
        handler: Callable[[bytes, dict[str, int]], Awaitable[None]],
    ) -> Callable[[], Awaitable[None]]:
        """Register an async PacketIn handler. Returns an async unsubscribe."""
        async with self._handlers_lock:
            self._packet_handlers.append(handler)

        async def deregister() -> None:
            async with self._handlers_lock:
                with contextlib.suppress(ValueError):
                    self._packet_handlers.remove(handler)

        return deregister

    async def expect_packet_in(
        self,
        *,
        timeout: float = 5.0,
    ) -> tuple[bytes, dict[str, int]]:
        """Await the next PacketIn. Raises P4RuntimeError on timeout."""
        self._require_connected()
        q = self._packet_in_queue
        if q is None:
            raise ConnectionError("packet_in queue is not initialised")
        try:
            return await asyncio.wait_for(q.get(), timeout=timeout)
        except asyncio.TimeoutError as exc:
            raise P4RuntimeError(f"no PacketIn within {timeout}s on {self.target!r}") from exc
        except asyncio.CancelledError as exc:
            raise AsyncOperationCancelledError("expect_packet_in cancelled") from exc

    # ----------------------------------------------------------------- internals

    def _require_connected(self) -> None:
        if not self._connected:
            raise ConnectionError(f"AsyncP4RuntimeClient {self.target} is not connected")

    def _require_connected_with_index(self) -> None:
        self._require_connected()
        if self._info_index is None:
            raise P4RuntimeError(
                "no P4Info index; pass info_index= to ctor or call push_pipeline first"
            )

    def _translate_rpc_error(
        self,
        exc: grpc.aio.AioRpcError,
        *,
        pipeline: bool = False,
    ) -> P4RuntimeError:
        code = exc.code()
        details = exc.details() or ""
        if pipeline:
            return PipelineError(f"pipeline rejected: {code} {details}")
        if code == grpc.StatusCode.ALREADY_EXISTS:
            return DuplicateEntryError(details)
        if code == grpc.StatusCode.NOT_FOUND:
            return EntryNotFoundError(details)
        if code in (grpc.StatusCode.PERMISSION_DENIED, grpc.StatusCode.FAILED_PRECONDITION):
            return NotPrimaryError(details)
        if code == grpc.StatusCode.CANCELLED:
            return AsyncOperationCancelledError(details)
        return P4RuntimeError(f"gRPC error {code}: {details}")

push_pipeline async

push_pipeline(p4info_bytes: bytes, json_bytes: bytes, *, action: str = 'VERIFY_AND_COMMIT', timeout: float = 10.0) -> None

Push a compiled pipeline (raw bytes).

Source code in src/p4net/control/async_client.py

async def push_pipeline(
    self,
    p4info_bytes: bytes,
    json_bytes: bytes,
    *,
    action: str = "VERIFY_AND_COMMIT",
    timeout: float = 10.0,
) -> None:
    """Push a compiled pipeline (raw bytes)."""
    self._require_connected()
    if action not in _PIPELINE_ACTIONS:
        raise P4RuntimeError(f"invalid pipeline action {action!r}")
    msg = p4info_pb2.P4Info()
    text_format.Merge(p4info_bytes.decode("utf-8"), msg)
    req = p4runtime_pb2.SetForwardingPipelineConfigRequest()
    req.device_id = self._device_id
    req.election_id.high = self._election_id[0]  # type: ignore[index]
    req.election_id.low = self._election_id[1]  # type: ignore[index]
    req.action = p4runtime_pb2.SetForwardingPipelineConfigRequest.Action.Value(action)
    req.config.p4info.CopyFrom(msg)
    req.config.p4_device_config = bytes(json_bytes)
    try:
        await asyncio.wait_for(self._stub.SetForwardingPipelineConfig(req), timeout=timeout)
    except asyncio.CancelledError as exc:
        raise AsyncOperationCancelledError("push_pipeline cancelled") from exc
    except grpc.aio.AioRpcError as exc:
        raise self._translate_rpc_error(exc, pipeline=True) from exc
    self._info_index = P4InfoIndex(msg)

list_table_entries

list_table_entries(table: str | None = None, *, timeout: float = 5.0) -> AsyncIterator[dict[str, Any]]

Async iterator over decoded table entries.

Note: this is a regular def returning an AsyncIterator. Use it as async for entry in client.list_table_entries(...).

Source code in src/p4net/control/async_client.py

def list_table_entries(
    self,
    table: str | None = None,
    *,
    timeout: float = 5.0,
) -> AsyncIterator[dict[str, Any]]:
    """Async iterator over decoded table entries.

    Note: this is a regular ``def`` returning an ``AsyncIterator``.
    Use it as ``async for entry in client.list_table_entries(...)``.
    """
    self._require_connected_with_index()
    idx = self.info_index
    req = p4runtime_pb2.ReadRequest()
    req.device_id = self._device_id
    entity = req.entities.add()
    if table is not None:
        entity.table_entry.table_id = idx.table_id(table)
    else:
        entity.table_entry.table_id = 0

    async def _gen() -> AsyncIterator[dict[str, Any]]:
        try:
            async for resp in self._stub.Read(req, timeout=timeout):
                for ent in resp.entities:
                    if ent.HasField("table_entry"):
                        yield self._decode_table_entry(ent.table_entry)
        except asyncio.CancelledError:
            raise
        except grpc.aio.AioRpcError as exc:
            raise self._translate_rpc_error(exc) from exc

    return _gen()

read_counter async

read_counter(counter: str, index: int | None = None, *, timeout: float = 5.0) -> int | dict[int, int]

Read indirect counter. Returns packet_count only for simplicity.

Single-index returns the cell's packet_count (int). No index returns {cell_index: packet_count}. The sync client returns the richer CounterData dataclass; the async API keeps things simple per spec.

Source code in src/p4net/control/async_client.py

async def read_counter(
    self,
    counter: str,
    index: int | None = None,
    *,
    timeout: float = 5.0,
) -> int | dict[int, int]:
    """Read indirect counter. Returns packet_count only for simplicity.

    Single-index returns the cell's packet_count (int). No index returns
    ``{cell_index: packet_count}``. The sync client returns the richer
    ``CounterData`` dataclass; the async API keeps things simple per spec.
    """
    self._require_connected_with_index()
    idx = self.info_index
    counter_id = idx.counter_id(counter)
    req = p4runtime_pb2.ReadRequest()
    req.device_id = self._device_id
    entity = req.entities.add()
    entity.counter_entry.counter_id = counter_id
    if index is not None:
        entity.counter_entry.index.index = int(index)
    collected: dict[int, int] = {}
    try:
        async for resp in self._stub.Read(req, timeout=timeout):
            for ent in resp.entities:
                if not ent.HasField("counter_entry"):
                    continue
                ce = ent.counter_entry
                collected[int(ce.index.index)] = int(ce.data.packet_count)
    except asyncio.CancelledError as exc:
        raise AsyncOperationCancelledError("read_counter cancelled") from exc
    except grpc.aio.AioRpcError as exc:
        raise self._translate_rpc_error(exc) from exc
    if index is not None:
        return collected.get(int(index), 0)
    return collected

on_packet_in async

on_packet_in(handler: Callable[[bytes, dict[str, int]], Awaitable[None]]) -> Callable[[], Awaitable[None]]

Register an async PacketIn handler. Returns an async unsubscribe.

Source code in src/p4net/control/async_client.py

async def on_packet_in(
    self,
    handler: Callable[[bytes, dict[str, int]], Awaitable[None]],
) -> Callable[[], Awaitable[None]]:
    """Register an async PacketIn handler. Returns an async unsubscribe."""
    async with self._handlers_lock:
        self._packet_handlers.append(handler)

    async def deregister() -> None:
        async with self._handlers_lock:
            with contextlib.suppress(ValueError):
                self._packet_handlers.remove(handler)

    return deregister

expect_packet_in async

expect_packet_in(*, timeout: float = 5.0) -> tuple[bytes, dict[str, int]]

Await the next PacketIn. Raises P4RuntimeError on timeout.

Source code in src/p4net/control/async_client.py

async def expect_packet_in(
    self,
    *,
    timeout: float = 5.0,
) -> tuple[bytes, dict[str, int]]:
    """Await the next PacketIn. Raises P4RuntimeError on timeout."""
    self._require_connected()
    q = self._packet_in_queue
    if q is None:
        raise ConnectionError("packet_in queue is not initialised")
    try:
        return await asyncio.wait_for(q.get(), timeout=timeout)
    except asyncio.TimeoutError as exc:
        raise P4RuntimeError(f"no PacketIn within {timeout}s on {self.target!r}") from exc
    except asyncio.CancelledError as exc:
        raise AsyncOperationCancelledError("expect_packet_in cancelled") from exc

CounterData dataclass

Decoded value of a P4 indirect counter cell.

Source code in src/p4net/control/client.py

@dataclass(frozen=True)
class CounterData:
    """Decoded value of a P4 indirect counter cell."""

    packet_count: int
    byte_count: int

P4RuntimeClient

gRPC client for a single P4Runtime device.

Source code in src/p4net/control/client.py

class P4RuntimeClient:
    """gRPC client for a single P4Runtime device."""

    def __init__(
        self,
        target: str,
        device_id: int,
        *,
        election_id: tuple[int, int] = (1, 0),
        role: str | None = None,
        channel_options: Sequence[tuple[str, object]] | None = None,
        thrift_address: tuple[str, int] | None = None,
    ) -> None:
        self._target = target
        self._device_id = int(device_id)
        self._election_id_high = int(election_id[0])
        self._election_id_low = int(election_id[1])
        self._role_name = role
        self._channel_options: list[tuple[str, Any]] = (
            list(channel_options) if channel_options is not None else list(_DEFAULT_CHANNEL_OPTIONS)
        )
        self._thrift_address = thrift_address

        self._channel: Any = None
        self._stub: Any = None
        self._outgoing: queue.Queue[Any] | None = None
        self._stream_call: Any = None
        self._stream_thread: threading.Thread | None = None
        self._stream_event = threading.Event()
        self._arbitration: Any = None
        self._stream_error: BaseException | None = None
        self._connected = False
        self._closed = False
        self._index: P4InfoIndex | None = None
        # Controller packet I/O. Handlers run on the stream-consumer thread.
        self._packet_in_handlers: list[Callable[[bytes, dict[str, int]], None]] = []
        self._packet_in_lock = threading.Lock()

    # Properties ---------------------------------------------------------

    @property
    def target(self) -> str:
        """gRPC target string (``host:port``) this client is bound to."""
        return self._target

    @property
    def device_id(self) -> int:
        """P4Runtime device ID this client identifies as."""
        return self._device_id

    @property
    def election_id(self) -> tuple[int, int]:
        """Mastership election ID as a ``(high, low)`` tuple."""
        return (self._election_id_high, self._election_id_low)

    @property
    def index(self) -> P4InfoIndex:
        """The :class:`P4InfoIndex` for the currently-pushed pipeline.

        Raises:
            P4RuntimeError: if no pipeline has been set yet.
        """
        if self._index is None:
            raise P4RuntimeError(
                "no pipeline is set; call set_pipeline_config or get_pipeline_config first"
            )
        return self._index

    def is_connected(self) -> bool:
        """``True`` while the gRPC channel is open and arbitration succeeded."""
        return self._connected and not self._closed

    # Lifecycle ----------------------------------------------------------

    def connect(self, *, timeout: float = 10.0) -> None:
        """Open the gRPC channel, start the StreamChannel, do master arbitration."""
        if self._connected:
            return
        self._closed = False
        self._channel = grpc.insecure_channel(self._target, options=self._channel_options)
        self._stub = p4runtime_pb2_grpc.P4RuntimeStub(self._channel)
        self._outgoing = queue.Queue()
        self._stream_event.clear()
        self._arbitration = None
        self._stream_error = None
        try:
            self._stream_call = self._stub.StreamChannel(self._request_generator())
        except grpc.RpcError as exc:
            raise self._translate_rpc_error(exc) from exc
        self._stream_thread = threading.Thread(
            target=self._stream_consumer, name="p4rt-stream", daemon=True
        )
        self._stream_thread.start()

        # Send the initial arbitration update.
        req = p4runtime_pb2.StreamMessageRequest()
        req.arbitration.device_id = self._device_id
        req.arbitration.election_id.high = self._election_id_high
        req.arbitration.election_id.low = self._election_id_low
        if self._role_name:
            req.arbitration.role.name = self._role_name
        assert self._outgoing is not None
        self._outgoing.put(req)

        if not self._stream_event.wait(timeout):
            self._teardown()
            raise ConnectionError(
                f"P4Runtime arbitration timed out after {timeout}s for {self._target!r}"
            )
        if self._stream_error is not None:
            err = self._stream_error
            self._teardown()
            if isinstance(err, grpc.RpcError):
                raise self._translate_rpc_error(err) from err
            raise P4RuntimeError(f"stream error: {err!r}") from err
        if self._arbitration is None:
            self._teardown()
            raise ConnectionError("no arbitration response received")
        status_code = int(self._arbitration.status.code)
        if status_code != 0:
            self._teardown()
            raise NotPrimaryError(
                f"client is not primary for device {self._device_id} "
                f"(status code {status_code}, message {self._arbitration.status.message!r})"
            )
        self._connected = True
        logger.debug(
            "P4Runtime client %r connected, election_id=(%d, %d)",
            self._target,
            self._election_id_high,
            self._election_id_low,
        )

    def disconnect(self) -> None:
        """Close the stream channel and the gRPC channel. Idempotent."""
        if self._closed:
            return
        self._teardown()

    def _teardown(self) -> None:
        self._closed = True
        was_connected = self._connected
        self._connected = False
        if self._outgoing is not None:
            with contextlib.suppress(queue.Full):
                self._outgoing.put_nowait(_SENTINEL)
        if self._stream_call is not None:
            try:
                self._stream_call.cancel()
            except Exception as exc:
                logger.debug("stream cancel raised: %r", exc)
        if self._stream_thread is not None and self._stream_thread.is_alive():
            self._stream_thread.join(timeout=2.0)
        if self._channel is not None:
            try:
                self._channel.close()
            except Exception as exc:
                logger.debug("channel close raised: %r", exc)
        self._channel = None
        self._stub = None
        self._stream_call = None
        self._stream_thread = None
        self._outgoing = None
        if was_connected:
            logger.debug("P4Runtime client %r disconnected", self._target)

    def _request_generator(self) -> Iterator[Any]:
        assert self._outgoing is not None
        while True:
            msg = self._outgoing.get()
            if msg is _SENTINEL:
                return
            yield msg

    def _stream_consumer(self) -> None:
        try:
            assert self._stream_call is not None
            for resp in self._stream_call:
                if resp.HasField("arbitration"):
                    self._arbitration = resp.arbitration
                    self._stream_event.set()
                elif resp.HasField("packet"):
                    self._dispatch_packet_in(resp.packet)
                # Future: digest / idle-notification / etc.
        except grpc.RpcError as exc:
            if not self._closed:
                self._stream_error = exc
                self._stream_event.set()
        except Exception as exc:
            if not self._closed:
                self._stream_error = exc
                self._stream_event.set()

    def _dispatch_packet_in(self, packet: Any) -> None:
        """Decode a PacketIn and fan out to registered handlers."""
        payload = bytes(packet.payload)
        metadata: dict[str, int] = {}
        if self._index is not None:
            try:
                metadata = self._index.decode_packet_in_metadata(packet.metadata)
            except Exception as exc:
                logger.debug("decode_packet_in_metadata raised: %r", exc)
        with self._packet_in_lock:
            handlers = list(self._packet_in_handlers)
        for h in handlers:
            try:
                h(payload, metadata)
            except Exception as exc:
                logger.warning("packet_in handler %r raised: %r", h, exc)

    # Pipeline -----------------------------------------------------------

    def set_pipeline_config(
        self,
        *,
        bmv2_json: Path,
        p4info: Path,
        action: str = "VERIFY_AND_COMMIT",
        timeout: float = 10.0,
    ) -> None:
        """Push a compiled pipeline to the device."""
        self._require_connected()
        if action not in _PIPELINE_ACTIONS:
            raise P4RuntimeError(
                f"invalid pipeline action {action!r}; must be one of {sorted(_PIPELINE_ACTIONS)}"
            )
        p4info_msg = p4info_pb2.P4Info()
        text_format.Merge(Path(p4info).read_text(), p4info_msg)
        bmv2_data = Path(bmv2_json).read_bytes()

        req = p4runtime_pb2.SetForwardingPipelineConfigRequest()
        req.device_id = self._device_id
        if self._role_name:
            req.role = self._role_name
        req.election_id.high = self._election_id_high
        req.election_id.low = self._election_id_low
        req.action = p4runtime_pb2.SetForwardingPipelineConfigRequest.Action.Value(action)
        req.config.p4info.CopyFrom(p4info_msg)
        req.config.p4_device_config = bmv2_data

        try:
            self._stub.SetForwardingPipelineConfig(req, timeout=timeout)
        except grpc.RpcError as exc:
            raise self._translate_rpc_error(exc, pipeline=True) from exc
        self._index = P4InfoIndex(p4info_msg)
        logger.debug(
            "P4Runtime pipeline pushed (action=%s, p4info_tables=%d)",
            action,
            len(p4info_msg.tables),
        )

    def get_pipeline_config(self, *, timeout: float = 10.0) -> tuple[bytes, P4InfoIndex]:
        """Read the current pipeline back from the device."""
        self._require_connected()
        req = p4runtime_pb2.GetForwardingPipelineConfigRequest()
        req.device_id = self._device_id
        req.response_type = p4runtime_pb2.GetForwardingPipelineConfigRequest.ResponseType.Value(
            "ALL"
        )
        try:
            resp = self._stub.GetForwardingPipelineConfig(req, timeout=timeout)
        except grpc.RpcError as exc:
            raise self._translate_rpc_error(exc) from exc
        index = P4InfoIndex(resp.config.p4info)
        self._index = index
        return resp.config.p4_device_config, index

    # Table CRUD ---------------------------------------------------------

    def insert_table_entry(
        self,
        table: str,
        match: Mapping[str, object],
        action: str,
        params: Mapping[str, object] | None = None,
        *,
        priority: int | None = None,
        timeout: float = 5.0,
    ) -> None:
        """Insert a new entry into ``table``.

        Args:
            table: Fully qualified table name (e.g. ``MyIngress.ipv4_lpm``).
            match: ``{field_name: value}`` for the match key. Values may be
                IPv4/IPv6/MAC strings, decimal/hex integers, or raw bytes.
                LPM fields take ``"<addr>/<plen>"``; ternary fields take
                ``("<value>", "<mask>")``.
            action: Fully qualified action name.
            params: Action parameters keyed by P4 param name.
            priority: Required for tables with ternary or range keys.
            timeout: Per-call gRPC deadline in seconds.

        Raises:
            DuplicateEntryError: if an entry with the same key already exists.
            EncodingError: if any field can't be encoded.
        """
        self._write_entry(
            table,
            match,
            action,
            params,
            priority=priority,
            update_type="INSERT",
            timeout=timeout,
        )

    def modify_table_entry(
        self,
        table: str,
        match: Mapping[str, object],
        action: str,
        params: Mapping[str, object] | None = None,
        *,
        priority: int | None = None,
        timeout: float = 5.0,
    ) -> None:
        """Modify an existing entry in ``table``. Same arguments as
        :meth:`insert_table_entry`. Raises :class:`EntryNotFoundError`
        if no matching entry exists."""
        self._write_entry(
            table,
            match,
            action,
            params,
            priority=priority,
            update_type="MODIFY",
            timeout=timeout,
        )

    def delete_table_entry(
        self,
        table: str,
        match: Mapping[str, object],
        *,
        priority: int | None = None,
        timeout: float = 5.0,
    ) -> None:
        """Delete an entry from ``table`` by match key.

        Raises:
            EntryNotFoundError: if no matching entry exists.
        """
        self._write_entry(
            table,
            match,
            action=None,
            params=None,
            priority=priority,
            update_type="DELETE",
            timeout=timeout,
        )

    def _write_entry(
        self,
        table: str,
        match: Mapping[str, object],
        action: str | None,
        params: Mapping[str, object] | None,
        *,
        priority: int | None,
        update_type: str,
        timeout: float,
    ) -> None:
        self._require_connected_with_index()
        index = self._index
        assert index is not None
        table_id = index.table_id(table)
        requires_priority = index.table_requires_priority(table)
        if requires_priority and priority is None and update_type != "DELETE":
            raise EncodingError(
                f"table {table!r} has ternary/range match fields; priority is required"
            )
        if not requires_priority and priority is not None:
            raise EncodingError(f"table {table!r} is exact/lpm-only; priority must be None")
        field_matches = index.encode_match(table, match)
        entry = p4runtime_pb2.TableEntry()
        entry.table_id = table_id
        for fm in field_matches:
            entry.match.add().CopyFrom(fm)
        if action is not None:
            entry.action.action.CopyFrom(index.encode_action(action, params))
        if priority is not None:
            entry.priority = int(priority)

        update = p4runtime_pb2.Update()
        update.type = p4runtime_pb2.Update.Type.Value(update_type)
        update.entity.table_entry.CopyFrom(entry)

        req = p4runtime_pb2.WriteRequest()
        req.device_id = self._device_id
        if self._role_name:
            req.role = self._role_name
        req.election_id.high = self._election_id_high
        req.election_id.low = self._election_id_low
        req.updates.add().CopyFrom(update)

        try:
            self._stub.Write(req, timeout=timeout)
        except grpc.RpcError as exc:
            raise self._translate_rpc_error(exc) from exc
        logger.debug(
            "P4Runtime %s table_entry table=%s match=%s action=%s",
            update_type,
            table,
            dict(match),
            action,
        )

    def list_table_entries(
        self,
        table: str | None = None,
        *,
        timeout: float = 5.0,
    ) -> list[dict[str, Any]]:
        """Return decoded entries for one table, or all tables if `table` is None.

        The byte values in each entry's ``match`` mapping are returned in
        P4Runtime canonical form (P4Runtime spec §8.4): they may be shorter
        than the bitwidth-rounded width because leading zero bytes are
        stripped. They round-trip correctly through ``insert_table_entry``,
        ``modify_table_entry``, and ``delete_table_entry`` because
        ``encode_value`` accepts shorter ``bytes`` inputs.
        """
        self._require_connected_with_index()
        req = p4runtime_pb2.ReadRequest()
        req.device_id = self._device_id
        entity = req.entities.add()
        if table is not None:
            entity.table_entry.table_id = self.index.table_id(table)
        else:
            entity.table_entry.table_id = 0
        try:
            response_iter = self._stub.Read(req, timeout=timeout)
            entries: list[dict[str, Any]] = []
            for resp in response_iter:
                for ent in resp.entities:
                    if ent.HasField("table_entry"):
                        entries.append(self._decode_table_entry(ent.table_entry))
            return entries
        except grpc.RpcError as exc:
            raise self._translate_rpc_error(exc) from exc

    def clear_table(self, table: str, *, timeout: float = 10.0) -> int:
        """Delete every entry from `table`. Returns the count deleted."""
        self._require_connected_with_index()
        entries = self.list_table_entries(table, timeout=timeout)
        if not entries:
            return 0
        index = self._index
        assert index is not None
        table_id = index.table_id(table)
        req = p4runtime_pb2.WriteRequest()
        req.device_id = self._device_id
        if self._role_name:
            req.role = self._role_name
        req.election_id.high = self._election_id_high
        req.election_id.low = self._election_id_low
        for entry in entries:
            te = p4runtime_pb2.TableEntry()
            te.table_id = table_id
            # Re-encode the match fields. The decoded form holds raw bytes;
            # we feed those back as bytes into encode_value, which is fine.
            field_matches = index.encode_match(table, entry["match"])
            for fm in field_matches:
                te.match.add().CopyFrom(fm)
            if entry.get("priority") is not None:
                te.priority = int(entry["priority"])
            update = req.updates.add()
            update.type = p4runtime_pb2.Update.Type.Value("DELETE")
            update.entity.table_entry.CopyFrom(te)
        try:
            self._stub.Write(req, timeout=timeout)
        except grpc.RpcError as exc:
            raise self._translate_rpc_error(exc) from exc
        return len(entries)

    def _decode_table_entry(self, entry: Any) -> dict[str, Any]:
        index = self._index
        assert index is not None
        table_name = index.table_name(int(entry.table_id))
        table = index.raw.tables[0]
        for t in index.raw.tables:
            if t.preamble.id == entry.table_id:
                table = t
                break
        fields_by_id = {int(mf.id): mf for mf in table.match_fields}
        match: dict[str, Any] = {}
        for fm in entry.match:
            mf = fields_by_id.get(int(fm.field_id))
            if mf is None:
                continue
            if fm.HasField("exact"):
                match[mf.name] = bytes(fm.exact.value)
            elif fm.HasField("lpm"):
                match[mf.name] = (bytes(fm.lpm.value), int(fm.lpm.prefix_len))
            elif fm.HasField("ternary"):
                match[mf.name] = (bytes(fm.ternary.value), bytes(fm.ternary.mask))
            elif fm.HasField("range"):
                match[mf.name] = (bytes(fm.range.low), bytes(fm.range.high))
            elif fm.HasField("optional"):
                match[mf.name] = bytes(fm.optional.value)
        action_name: str | None = None
        params: dict[str, bytes] = {}
        if entry.HasField("action") and entry.action.HasField("action"):
            a = entry.action.action
            action_name = index.action_name(int(a.action_id))
            for action_msg in index.raw.actions:
                if action_msg.preamble.id == a.action_id:
                    params_by_id = {int(p.id): p for p in action_msg.params}
                    for ap in a.params:
                        p = params_by_id.get(int(ap.param_id))
                        if p is not None:
                            params[p.name] = bytes(ap.value)
                    break
        return {
            "table": table_name,
            "match": match,
            "action": action_name,
            "params": params,
            "priority": int(entry.priority) if entry.priority else None,
        }

    # Counters -----------------------------------------------------------

    def read_counter(
        self,
        counter: str,
        index: int | None = None,
        *,
        timeout: float = 5.0,
    ) -> CounterData | dict[int, CounterData]:
        """Read one or all populated cells of an indirect counter."""
        self._require_connected_with_index()
        idx = self._index
        assert idx is not None
        counter_id = idx.counter_id(counter)
        req = p4runtime_pb2.ReadRequest()
        req.device_id = self._device_id
        entity = req.entities.add()
        entity.counter_entry.counter_id = counter_id
        if index is not None:
            entity.counter_entry.index.index = int(index)
        try:
            response_iter = self._stub.Read(req, timeout=timeout)
            collected: dict[int, CounterData] = {}
            for resp in response_iter:
                for ent in resp.entities:
                    if not ent.HasField("counter_entry"):
                        continue
                    ce = ent.counter_entry
                    cell_index = int(ce.index.index)
                    collected[cell_index] = CounterData(
                        packet_count=int(ce.data.packet_count),
                        byte_count=int(ce.data.byte_count),
                    )
        except grpc.RpcError as exc:
            raise self._translate_rpc_error(exc) from exc
        if index is not None:
            return collected.get(int(index), CounterData(0, 0))
        return collected

    def reset_counter(
        self,
        counter: str,
        index: int | None = None,
        *,
        timeout: float = 5.0,
    ) -> None:
        """Zero one or all indices of an indirect counter."""
        self._require_connected_with_index()
        idx = self._index
        assert idx is not None
        counter_id = idx.counter_id(counter)
        targets: list[int]
        if index is None:
            current = self.read_counter(counter, timeout=timeout)
            assert isinstance(current, dict)
            targets = sorted(current.keys())
            if not targets:
                return
        else:
            targets = [int(index)]
        req = p4runtime_pb2.WriteRequest()
        req.device_id = self._device_id
        if self._role_name:
            req.role = self._role_name
        req.election_id.high = self._election_id_high
        req.election_id.low = self._election_id_low
        for cell_idx in targets:
            update = req.updates.add()
            update.type = p4runtime_pb2.Update.Type.Value("MODIFY")
            ce = update.entity.counter_entry
            ce.counter_id = counter_id
            ce.index.index = int(cell_idx)
            ce.data.packet_count = 0
            ce.data.byte_count = 0
        try:
            self._stub.Write(req, timeout=timeout)
        except grpc.RpcError as exc:
            raise self._translate_rpc_error(exc) from exc

    # Registers ----------------------------------------------------------

    # BMv2's P4Runtime backend currently returns UNIMPLEMENTED for RegisterEntry
    # over gRPC ("Register reads are not supported yet" in libpifeproto). The
    # P4Runtime contract is honored at the Python API surface — same method
    # names and semantics — but the implementation delegates to BMv2's Thrift
    # control channel via ``simple_switch_CLI``. Targets with a compliant
    # P4Runtime RegisterEntry implementation can be migrated by swapping the
    # transport here without changing the public API.

    def write_register(
        self,
        name: str,
        index: int,
        value: int,
        *,
        timeout: float = 5.0,
    ) -> None:
        """Write a single cell of a P4 register.

        Args:
            name: Fully qualified P4 register name
                (e.g. ``MyIngress.switch_id``).
            index: Cell index in the register array. Must be in range
                ``[0, size)``.
            value: Integer value to write. Must fit in the register's
                bitwidth.

        Raises:
            NoSuchRegisterError: if the register doesn't exist.
            EncodingError: if ``index`` is out of range or ``value``
                exceeds the register's bitwidth.
            P4RuntimeError: if the underlying control channel returns
                an error or this client has no Thrift address configured.
        """
        self._require_connected_with_index()
        idx = self._index
        assert idx is not None
        spec = idx.register_by_name(name)
        if not 0 <= int(index) < spec.size:
            raise EncodingError(f"register {name!r} index {index} out of range [0, {spec.size})")
        # Force bitwidth validation. Result bytes are unused — the Thrift
        # CLI takes a decimal integer.
        encode_value(int(value), spec.bitwidth)
        self._run_thrift_cli(
            f"register_write {name} {int(index)} {int(value)}",
            timeout=timeout,
            op_description=f"register_write {name}[{index}]",
        )
        logger.debug(
            "BMv2 thrift register_write %s[%d] = %d (bitwidth=%d)",
            name,
            int(index),
            int(value),
            spec.bitwidth,
        )

    def read_register(
        self,
        name: str,
        index: int | None = None,
        *,
        timeout: float = 5.0,
    ) -> int | list[int]:
        """Read a P4 register.

        Args:
            name: Fully qualified P4 register name.
            index: Cell index, or ``None`` to read every cell.

        Returns:
            If ``index`` is given: the integer value of that cell.
            If ``index`` is ``None``: a list of integers indexed by
            position ``[0, size)``. Cells default to ``0`` (BMv2
            initializes register elements to zero).

        Raises:
            NoSuchRegisterError: if the register doesn't exist.
            EncodingError: if ``index`` is out of range.
            P4RuntimeError: if the underlying control channel returns
                an error or this client has no Thrift address configured.
        """
        self._require_connected_with_index()
        idx = self._index
        assert idx is not None
        spec = idx.register_by_name(name)
        if index is not None and not 0 <= int(index) < spec.size:
            raise EncodingError(f"register {name!r} index {index} out of range [0, {spec.size})")
        if index is not None:
            output = self._run_thrift_cli(
                f"register_read {name} {int(index)}",
                timeout=timeout,
                op_description=f"register_read {name}[{index}]",
            )
            return _parse_register_read_single(output, name, int(index))
        output = self._run_thrift_cli(
            f"register_read {name}",
            timeout=timeout,
            op_description=f"register_read {name}",
        )
        return _parse_register_read_array(output, name, spec.size)

    def _run_thrift_cli(
        self,
        command: str,
        *,
        timeout: float,
        op_description: str,
    ) -> str:
        """Run a single command against ``simple_switch_CLI`` and return stdout."""
        if self._thrift_address is None:
            raise P4RuntimeError(
                "register operations require a Thrift sidecar address; "
                f"construct P4RuntimeClient with thrift_address=(host, port). "
                f"Operation: {op_description}"
            )
        host, port = self._thrift_address
        import subprocess

        try:
            result = subprocess.run(
                [
                    "simple_switch_CLI",
                    "--thrift-ip",
                    str(host),
                    "--thrift-port",
                    str(int(port)),
                ],
                input=command + "\n",
                capture_output=True,
                text=True,
                timeout=timeout,
            )
        except FileNotFoundError as exc:
            raise P4RuntimeError(
                "simple_switch_CLI not found on PATH; required for register operations"
            ) from exc
        except subprocess.TimeoutExpired as exc:
            raise P4RuntimeError(
                f"{op_description} timed out after {timeout}s against thrift {host}:{port}"
            ) from exc
        combined = result.stdout + result.stderr
        if result.returncode != 0:
            raise P4RuntimeError(
                f"{op_description} failed (rc={result.returncode}): {combined.strip()}"
            )
        for line in combined.splitlines():
            stripped = line.strip()
            if stripped.startswith("Error:") or "Invalid" in stripped:
                raise P4RuntimeError(f"{op_description}: {stripped}")
        return result.stdout

    # Multicast groups ---------------------------------------------------

    def add_multicast_group(
        self,
        group_id: int,
        ports: Sequence[int],
        *,
        timeout: float = 5.0,
    ) -> None:
        """Create a multicast group with one replica per port (instance=1)."""
        self._mcast_write(group_id, ports, update_type="INSERT", timeout=timeout)

    def modify_multicast_group(
        self,
        group_id: int,
        ports: Sequence[int],
        *,
        timeout: float = 5.0,
    ) -> None:
        """Replace the replica list of an existing multicast group."""
        self._mcast_write(group_id, ports, update_type="MODIFY", timeout=timeout)

    def delete_multicast_group(
        self,
        group_id: int,
        *,
        timeout: float = 5.0,
    ) -> None:
        """Delete a multicast group."""
        self._mcast_write(group_id, ports=(), update_type="DELETE", timeout=timeout)

    def _mcast_write(
        self,
        group_id: int,
        ports: Sequence[int],
        *,
        update_type: str,
        timeout: float,
    ) -> None:
        self._require_connected()
        if group_id <= 0:
            raise EncodingError(f"multicast group_id must be positive, got {group_id}")
        req = p4runtime_pb2.WriteRequest()
        req.device_id = self._device_id
        if self._role_name:
            req.role = self._role_name
        req.election_id.high = self._election_id_high
        req.election_id.low = self._election_id_low
        update = req.updates.add()
        update.type = p4runtime_pb2.Update.Type.Value(update_type)
        mge = update.entity.packet_replication_engine_entry.multicast_group_entry
        mge.multicast_group_id = int(group_id)
        if update_type != "DELETE":
            for port in ports:
                replica = mge.replicas.add()
                replica.egress_port = int(port)
                replica.instance = 1
        try:
            self._stub.Write(req, timeout=timeout)
        except grpc.RpcError as exc:
            raise self._translate_rpc_error(exc) from exc

    def list_multicast_groups(self, *, timeout: float = 5.0) -> dict[int, list[int]]:
        """Return ``{group_id: [egress_port, ...]}``.

        Replica instance numbers are flattened away — each port appears once
        per replica regardless of its instance value (we always write
        instance=1 ourselves; foreign instance values are still listed but
        not exposed in this dict shape).
        """
        self._require_connected()
        req = p4runtime_pb2.ReadRequest()
        req.device_id = self._device_id
        entity = req.entities.add()
        entity.packet_replication_engine_entry.multicast_group_entry.multicast_group_id = 0
        try:
            response_iter = self._stub.Read(req, timeout=timeout)
            groups: dict[int, list[int]] = {}
            for resp in response_iter:
                for ent in resp.entities:
                    if not ent.HasField("packet_replication_engine_entry"):
                        continue
                    pre = ent.packet_replication_engine_entry
                    if not pre.HasField("multicast_group_entry"):
                        continue
                    mge = pre.multicast_group_entry
                    groups[int(mge.multicast_group_id)] = [int(r.egress_port) for r in mge.replicas]
            return groups
        except grpc.RpcError as exc:
            raise self._translate_rpc_error(exc) from exc

    # Controller packet I/O ---------------------------------------------

    def send_packet_out(
        self,
        payload: bytes,
        metadata: Mapping[str, object] | None = None,
        *,
        timeout: float = 5.0,
    ) -> None:
        """Send a PacketOut over the StreamChannel.

        ``payload`` is the full packet to inject — controller headers are
        rebuilt from ``metadata`` per the loaded P4Info. PacketOut is
        fire-and-forget in P4Runtime; this method does not wait for a switch
        response. ``timeout`` is reserved for future flow-control limits and
        currently only bounds the queue put.
        """
        self._require_connected()
        idx = self._index
        encoded = idx.encode_packet_out_metadata(metadata or {}) if idx is not None else []
        if metadata and idx is None:
            raise EncodingError(
                "no pipeline is set; cannot encode packet_out metadata "
                "(call set_pipeline_config or get_pipeline_config first)"
            )
        if not isinstance(payload, (bytes, bytearray)):
            raise EncodingError(f"payload must be bytes-like, got {type(payload).__name__}")
        req = p4runtime_pb2.StreamMessageRequest()
        req.packet.payload = bytes(payload)
        for pm in encoded:
            req.packet.metadata.add().CopyFrom(pm)
        out = self._outgoing
        if out is None:
            raise ConnectionError("client is not connected; outgoing queue is closed")
        out.put(req, timeout=timeout)

    def on_packet_in(
        self,
        handler: Callable[[bytes, dict[str, int]], None],
    ) -> Callable[[], None]:
        """Register a PacketIn handler. Returns a deregister function.

        Handlers run on the StreamChannel consumer thread. Multiple handlers
        are invoked in registration order; an exception from one is logged
        and does not prevent later handlers from running. The returned
        deregister function tolerates double-unregister silently.
        """
        with self._packet_in_lock:
            self._packet_in_handlers.append(handler)

        def deregister() -> None:
            with self._packet_in_lock, contextlib.suppress(ValueError):
                self._packet_in_handlers.remove(handler)

        return deregister

    def expect_packet_in(
        self,
        *,
        timeout: float = 5.0,
    ) -> tuple[bytes, dict[str, int]]:
        """Block until the next PacketIn arrives. Raises ``P4RuntimeError`` on timeout."""
        self._require_connected()
        q: queue.Queue[tuple[bytes, dict[str, int]]] = queue.Queue(maxsize=1)

        def _push(payload: bytes, meta: dict[str, int]) -> None:
            with contextlib.suppress(queue.Full):
                q.put_nowait((payload, meta))

        deregister = self.on_packet_in(_push)
        try:
            try:
                return q.get(timeout=timeout)
            except queue.Empty as exc:
                raise P4RuntimeError(f"no PacketIn within {timeout}s on {self._target!r}") from exc
        finally:
            deregister()

    # Internals ----------------------------------------------------------

    def _require_connected(self) -> None:
        if not self.is_connected():
            raise ConnectionError("client is not connected; call connect() first")

    def _require_connected_with_index(self) -> None:
        self._require_connected()
        if self._index is None:
            raise P4RuntimeError(
                "no pipeline is set; call set_pipeline_config or get_pipeline_config first"
            )

    def _translate_rpc_error(self, exc: grpc.RpcError, *, pipeline: bool = False) -> P4RuntimeError:
        try:
            code = exc.code()
        except Exception:
            return P4RuntimeError(f"gRPC error: {exc!r}")
        try:
            detail = exc.details() or ""
        except Exception:
            detail = ""
        # P4Runtime batches per-update statuses inside an outer UNKNOWN gRPC
        # error; the per-update canonical_code is encoded as a `p4.v1.Error`
        # entry in `grpc-status-details-bin`. Resolve to the real code so
        # callers see DuplicateEntryError / EntryNotFoundError instead of a
        # generic UNKNOWN.
        if code == grpc.StatusCode.UNKNOWN:
            for canonical in _extract_p4_canonical_codes(exc):
                if canonical == 0:
                    continue
                resolved = _grpc_code_for(canonical)
                if resolved is not None:
                    code = resolved
                    if not detail:
                        detail = f"P4Runtime canonical_code={canonical}"
                    break
        if code == grpc.StatusCode.UNAVAILABLE:
            return ConnectionError(f"gRPC unavailable for {self._target!r}: {detail}")
        if code == grpc.StatusCode.NOT_FOUND:
            return EntryNotFoundError(detail or "entry not found")
        if code == grpc.StatusCode.ALREADY_EXISTS:
            return DuplicateEntryError(detail or "entry already exists")
        if code == grpc.StatusCode.FAILED_PRECONDITION and pipeline:
            return PipelineError(detail or "pipeline rejected by switch")
        if code == grpc.StatusCode.INVALID_ARGUMENT:
            return P4RuntimeError(f"invalid argument: {detail}")
        return P4RuntimeError(f"gRPC error {code.name}: {detail}")

    # Context manager ----------------------------------------------------

    def __enter__(self) -> P4RuntimeClient:
        self.connect()
        return self

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc: BaseException | None,
        tb: TracebackType | None,
    ) -> None:
        self.disconnect()

    def __del__(self) -> None:
        with contextlib.suppress(Exception):
            self.disconnect()

target property

target: str

gRPC target string (host:port) this client is bound to.

device_id property

device_id: int

P4Runtime device ID this client identifies as.

election_id property

election_id: tuple[int, int]

Mastership election ID as a (high, low) tuple.

index property

index: P4InfoIndex

The :class:P4InfoIndex for the currently-pushed pipeline.

Raises:

Type	Description
P4RuntimeError	if no pipeline has been set yet.

is_connected

is_connected() -> bool

True while the gRPC channel is open and arbitration succeeded.

Source code in src/p4net/control/client.py

def is_connected(self) -> bool:
    """``True`` while the gRPC channel is open and arbitration succeeded."""
    return self._connected and not self._closed

connect

connect(*, timeout: float = 10.0) -> None

Open the gRPC channel, start the StreamChannel, do master arbitration.

Source code in src/p4net/control/client.py

def connect(self, *, timeout: float = 10.0) -> None:
    """Open the gRPC channel, start the StreamChannel, do master arbitration."""
    if self._connected:
        return
    self._closed = False
    self._channel = grpc.insecure_channel(self._target, options=self._channel_options)
    self._stub = p4runtime_pb2_grpc.P4RuntimeStub(self._channel)
    self._outgoing = queue.Queue()
    self._stream_event.clear()
    self._arbitration = None
    self._stream_error = None
    try:
        self._stream_call = self._stub.StreamChannel(self._request_generator())
    except grpc.RpcError as exc:
        raise self._translate_rpc_error(exc) from exc
    self._stream_thread = threading.Thread(
        target=self._stream_consumer, name="p4rt-stream", daemon=True
    )
    self._stream_thread.start()

    # Send the initial arbitration update.
    req = p4runtime_pb2.StreamMessageRequest()
    req.arbitration.device_id = self._device_id
    req.arbitration.election_id.high = self._election_id_high
    req.arbitration.election_id.low = self._election_id_low
    if self._role_name:
        req.arbitration.role.name = self._role_name
    assert self._outgoing is not None
    self._outgoing.put(req)

    if not self._stream_event.wait(timeout):
        self._teardown()
        raise ConnectionError(
            f"P4Runtime arbitration timed out after {timeout}s for {self._target!r}"
        )
    if self._stream_error is not None:
        err = self._stream_error
        self._teardown()
        if isinstance(err, grpc.RpcError):
            raise self._translate_rpc_error(err) from err
        raise P4RuntimeError(f"stream error: {err!r}") from err
    if self._arbitration is None:
        self._teardown()
        raise ConnectionError("no arbitration response received")
    status_code = int(self._arbitration.status.code)
    if status_code != 0:
        self._teardown()
        raise NotPrimaryError(
            f"client is not primary for device {self._device_id} "
            f"(status code {status_code}, message {self._arbitration.status.message!r})"
        )
    self._connected = True
    logger.debug(
        "P4Runtime client %r connected, election_id=(%d, %d)",
        self._target,
        self._election_id_high,
        self._election_id_low,
    )

disconnect

disconnect() -> None

Close the stream channel and the gRPC channel. Idempotent.

Source code in src/p4net/control/client.py

def disconnect(self) -> None:
    """Close the stream channel and the gRPC channel. Idempotent."""
    if self._closed:
        return
    self._teardown()

set_pipeline_config

set_pipeline_config(*, bmv2_json: Path, p4info: Path, action: str = 'VERIFY_AND_COMMIT', timeout: float = 10.0) -> None

Push a compiled pipeline to the device.

Source code in src/p4net/control/client.py

def set_pipeline_config(
    self,
    *,
    bmv2_json: Path,
    p4info: Path,
    action: str = "VERIFY_AND_COMMIT",
    timeout: float = 10.0,
) -> None:
    """Push a compiled pipeline to the device."""
    self._require_connected()
    if action not in _PIPELINE_ACTIONS:
        raise P4RuntimeError(
            f"invalid pipeline action {action!r}; must be one of {sorted(_PIPELINE_ACTIONS)}"
        )
    p4info_msg = p4info_pb2.P4Info()
    text_format.Merge(Path(p4info).read_text(), p4info_msg)
    bmv2_data = Path(bmv2_json).read_bytes()

    req = p4runtime_pb2.SetForwardingPipelineConfigRequest()
    req.device_id = self._device_id
    if self._role_name:
        req.role = self._role_name
    req.election_id.high = self._election_id_high
    req.election_id.low = self._election_id_low
    req.action = p4runtime_pb2.SetForwardingPipelineConfigRequest.Action.Value(action)
    req.config.p4info.CopyFrom(p4info_msg)
    req.config.p4_device_config = bmv2_data

    try:
        self._stub.SetForwardingPipelineConfig(req, timeout=timeout)
    except grpc.RpcError as exc:
        raise self._translate_rpc_error(exc, pipeline=True) from exc
    self._index = P4InfoIndex(p4info_msg)
    logger.debug(
        "P4Runtime pipeline pushed (action=%s, p4info_tables=%d)",
        action,
        len(p4info_msg.tables),
    )

get_pipeline_config

get_pipeline_config(*, timeout: float = 10.0) -> tuple[bytes, P4InfoIndex]

Read the current pipeline back from the device.

Source code in src/p4net/control/client.py

def get_pipeline_config(self, *, timeout: float = 10.0) -> tuple[bytes, P4InfoIndex]:
    """Read the current pipeline back from the device."""
    self._require_connected()
    req = p4runtime_pb2.GetForwardingPipelineConfigRequest()
    req.device_id = self._device_id
    req.response_type = p4runtime_pb2.GetForwardingPipelineConfigRequest.ResponseType.Value(
        "ALL"
    )
    try:
        resp = self._stub.GetForwardingPipelineConfig(req, timeout=timeout)
    except grpc.RpcError as exc:
        raise self._translate_rpc_error(exc) from exc
    index = P4InfoIndex(resp.config.p4info)
    self._index = index
    return resp.config.p4_device_config, index

insert_table_entry

insert_table_entry(table: str, match: Mapping[str, object], action: str, params: Mapping[str, object] | None = None, *, priority: int | None = None, timeout: float = 5.0) -> None

Insert a new entry into table.

Parameters:

Name	Type	Description	Default
table	str	Fully qualified table name (e.g. MyIngress.ipv4_lpm).	required
match	Mapping[str, object]	{field_name: value} for the match key. Values may be IPv4/IPv6/MAC strings, decimal/hex integers, or raw bytes. LPM fields take "<addr>/<plen>"; ternary fields take ("<value>", "<mask>").	required
action	str	Fully qualified action name.	required
params	Mapping[str, object] | None	Action parameters keyed by P4 param name.	None
priority	int | None	Required for tables with ternary or range keys.	None
timeout	float	Per-call gRPC deadline in seconds.	5.0

Raises:

Type	Description
DuplicateEntryError	if an entry with the same key already exists.
EncodingError	if any field can't be encoded.

Source code in src/p4net/control/client.py

def insert_table_entry(
    self,
    table: str,
    match: Mapping[str, object],
    action: str,
    params: Mapping[str, object] | None = None,
    *,
    priority: int | None = None,
    timeout: float = 5.0,
) -> None:
    """Insert a new entry into ``table``.

    Args:
        table: Fully qualified table name (e.g. ``MyIngress.ipv4_lpm``).
        match: ``{field_name: value}`` for the match key. Values may be
            IPv4/IPv6/MAC strings, decimal/hex integers, or raw bytes.
            LPM fields take ``"<addr>/<plen>"``; ternary fields take
            ``("<value>", "<mask>")``.
        action: Fully qualified action name.
        params: Action parameters keyed by P4 param name.
        priority: Required for tables with ternary or range keys.
        timeout: Per-call gRPC deadline in seconds.

    Raises:
        DuplicateEntryError: if an entry with the same key already exists.
        EncodingError: if any field can't be encoded.
    """
    self._write_entry(
        table,
        match,
        action,
        params,
        priority=priority,
        update_type="INSERT",
        timeout=timeout,
    )

modify_table_entry

modify_table_entry(table: str, match: Mapping[str, object], action: str, params: Mapping[str, object] | None = None, *, priority: int | None = None, timeout: float = 5.0) -> None

Modify an existing entry in table. Same arguments as :meth:insert_table_entry. Raises :class:EntryNotFoundError if no matching entry exists.

Source code in src/p4net/control/client.py

def modify_table_entry(
    self,
    table: str,
    match: Mapping[str, object],
    action: str,
    params: Mapping[str, object] | None = None,
    *,
    priority: int | None = None,
    timeout: float = 5.0,
) -> None:
    """Modify an existing entry in ``table``. Same arguments as
    :meth:`insert_table_entry`. Raises :class:`EntryNotFoundError`
    if no matching entry exists."""
    self._write_entry(
        table,
        match,
        action,
        params,
        priority=priority,
        update_type="MODIFY",
        timeout=timeout,
    )

delete_table_entry

delete_table_entry(table: str, match: Mapping[str, object], *, priority: int | None = None, timeout: float = 5.0) -> None

Delete an entry from table by match key.

Raises:

Type	Description
EntryNotFoundError	if no matching entry exists.

Source code in src/p4net/control/client.py

def delete_table_entry(
    self,
    table: str,
    match: Mapping[str, object],
    *,
    priority: int | None = None,
    timeout: float = 5.0,
) -> None:
    """Delete an entry from ``table`` by match key.

    Raises:
        EntryNotFoundError: if no matching entry exists.
    """
    self._write_entry(
        table,
        match,
        action=None,
        params=None,
        priority=priority,
        update_type="DELETE",
        timeout=timeout,
    )

list_table_entries

list_table_entries(table: str | None = None, *, timeout: float = 5.0) -> list[dict[str, Any]]

Return decoded entries for one table, or all tables if table is None.

The byte values in each entry's match mapping are returned in P4Runtime canonical form (P4Runtime spec §8.4): they may be shorter than the bitwidth-rounded width because leading zero bytes are stripped. They round-trip correctly through insert_table_entry, modify_table_entry, and delete_table_entry because encode_value accepts shorter bytes inputs.

Source code in src/p4net/control/client.py

def list_table_entries(
    self,
    table: str | None = None,
    *,
    timeout: float = 5.0,
) -> list[dict[str, Any]]:
    """Return decoded entries for one table, or all tables if `table` is None.

    The byte values in each entry's ``match`` mapping are returned in
    P4Runtime canonical form (P4Runtime spec §8.4): they may be shorter
    than the bitwidth-rounded width because leading zero bytes are
    stripped. They round-trip correctly through ``insert_table_entry``,
    ``modify_table_entry``, and ``delete_table_entry`` because
    ``encode_value`` accepts shorter ``bytes`` inputs.
    """
    self._require_connected_with_index()
    req = p4runtime_pb2.ReadRequest()
    req.device_id = self._device_id
    entity = req.entities.add()
    if table is not None:
        entity.table_entry.table_id = self.index.table_id(table)
    else:
        entity.table_entry.table_id = 0
    try:
        response_iter = self._stub.Read(req, timeout=timeout)
        entries: list[dict[str, Any]] = []
        for resp in response_iter:
            for ent in resp.entities:
                if ent.HasField("table_entry"):
                    entries.append(self._decode_table_entry(ent.table_entry))
        return entries
    except grpc.RpcError as exc:
        raise self._translate_rpc_error(exc) from exc

clear_table

clear_table(table: str, *, timeout: float = 10.0) -> int

Delete every entry from table. Returns the count deleted.

Source code in src/p4net/control/client.py

def clear_table(self, table: str, *, timeout: float = 10.0) -> int:
    """Delete every entry from `table`. Returns the count deleted."""
    self._require_connected_with_index()
    entries = self.list_table_entries(table, timeout=timeout)
    if not entries:
        return 0
    index = self._index
    assert index is not None
    table_id = index.table_id(table)
    req = p4runtime_pb2.WriteRequest()
    req.device_id = self._device_id
    if self._role_name:
        req.role = self._role_name
    req.election_id.high = self._election_id_high
    req.election_id.low = self._election_id_low
    for entry in entries:
        te = p4runtime_pb2.TableEntry()
        te.table_id = table_id
        # Re-encode the match fields. The decoded form holds raw bytes;
        # we feed those back as bytes into encode_value, which is fine.
        field_matches = index.encode_match(table, entry["match"])
        for fm in field_matches:
            te.match.add().CopyFrom(fm)
        if entry.get("priority") is not None:
            te.priority = int(entry["priority"])
        update = req.updates.add()
        update.type = p4runtime_pb2.Update.Type.Value("DELETE")
        update.entity.table_entry.CopyFrom(te)
    try:
        self._stub.Write(req, timeout=timeout)
    except grpc.RpcError as exc:
        raise self._translate_rpc_error(exc) from exc
    return len(entries)

read_counter

read_counter(counter: str, index: int | None = None, *, timeout: float = 5.0) -> CounterData | dict[int, CounterData]

Read one or all populated cells of an indirect counter.

Source code in src/p4net/control/client.py

def read_counter(
    self,
    counter: str,
    index: int | None = None,
    *,
    timeout: float = 5.0,
) -> CounterData | dict[int, CounterData]:
    """Read one or all populated cells of an indirect counter."""
    self._require_connected_with_index()
    idx = self._index
    assert idx is not None
    counter_id = idx.counter_id(counter)
    req = p4runtime_pb2.ReadRequest()
    req.device_id = self._device_id
    entity = req.entities.add()
    entity.counter_entry.counter_id = counter_id
    if index is not None:
        entity.counter_entry.index.index = int(index)
    try:
        response_iter = self._stub.Read(req, timeout=timeout)
        collected: dict[int, CounterData] = {}
        for resp in response_iter:
            for ent in resp.entities:
                if not ent.HasField("counter_entry"):
                    continue
                ce = ent.counter_entry
                cell_index = int(ce.index.index)
                collected[cell_index] = CounterData(
                    packet_count=int(ce.data.packet_count),
                    byte_count=int(ce.data.byte_count),
                )
    except grpc.RpcError as exc:
        raise self._translate_rpc_error(exc) from exc
    if index is not None:
        return collected.get(int(index), CounterData(0, 0))
    return collected

reset_counter

reset_counter(counter: str, index: int | None = None, *, timeout: float = 5.0) -> None

Zero one or all indices of an indirect counter.

Source code in src/p4net/control/client.py

def reset_counter(
    self,
    counter: str,
    index: int | None = None,
    *,
    timeout: float = 5.0,
) -> None:
    """Zero one or all indices of an indirect counter."""
    self._require_connected_with_index()
    idx = self._index
    assert idx is not None
    counter_id = idx.counter_id(counter)
    targets: list[int]
    if index is None:
        current = self.read_counter(counter, timeout=timeout)
        assert isinstance(current, dict)
        targets = sorted(current.keys())
        if not targets:
            return
    else:
        targets = [int(index)]
    req = p4runtime_pb2.WriteRequest()
    req.device_id = self._device_id
    if self._role_name:
        req.role = self._role_name
    req.election_id.high = self._election_id_high
    req.election_id.low = self._election_id_low
    for cell_idx in targets:
        update = req.updates.add()
        update.type = p4runtime_pb2.Update.Type.Value("MODIFY")
        ce = update.entity.counter_entry
        ce.counter_id = counter_id
        ce.index.index = int(cell_idx)
        ce.data.packet_count = 0
        ce.data.byte_count = 0
    try:
        self._stub.Write(req, timeout=timeout)
    except grpc.RpcError as exc:
        raise self._translate_rpc_error(exc) from exc

write_register

write_register(name: str, index: int, value: int, *, timeout: float = 5.0) -> None

Write a single cell of a P4 register.

Parameters:

Name	Type	Description	Default
name	str	Fully qualified P4 register name (e.g. MyIngress.switch_id).	required
index	int	Cell index in the register array. Must be in range [0, size).	required
value	int	Integer value to write. Must fit in the register's bitwidth.	required

Raises:

Type	Description
NoSuchRegisterError	if the register doesn't exist.
EncodingError	if index is out of range or value exceeds the register's bitwidth.
P4RuntimeError	if the underlying control channel returns an error or this client has no Thrift address configured.

Source code in src/p4net/control/client.py

def write_register(
    self,
    name: str,
    index: int,
    value: int,
    *,
    timeout: float = 5.0,
) -> None:
    """Write a single cell of a P4 register.

    Args:
        name: Fully qualified P4 register name
            (e.g. ``MyIngress.switch_id``).
        index: Cell index in the register array. Must be in range
            ``[0, size)``.
        value: Integer value to write. Must fit in the register's
            bitwidth.

    Raises:
        NoSuchRegisterError: if the register doesn't exist.
        EncodingError: if ``index`` is out of range or ``value``
            exceeds the register's bitwidth.
        P4RuntimeError: if the underlying control channel returns
            an error or this client has no Thrift address configured.
    """
    self._require_connected_with_index()
    idx = self._index
    assert idx is not None
    spec = idx.register_by_name(name)
    if not 0 <= int(index) < spec.size:
        raise EncodingError(f"register {name!r} index {index} out of range [0, {spec.size})")
    # Force bitwidth validation. Result bytes are unused — the Thrift
    # CLI takes a decimal integer.
    encode_value(int(value), spec.bitwidth)
    self._run_thrift_cli(
        f"register_write {name} {int(index)} {int(value)}",
        timeout=timeout,
        op_description=f"register_write {name}[{index}]",
    )
    logger.debug(
        "BMv2 thrift register_write %s[%d] = %d (bitwidth=%d)",
        name,
        int(index),
        int(value),
        spec.bitwidth,
    )

read_register

read_register(name: str, index: int | None = None, *, timeout: float = 5.0) -> int | list[int]

Read a P4 register.

Parameters:

Name	Type	Description	Default
name	str	Fully qualified P4 register name.	required
index	int | None	Cell index, or None to read every cell.	None

Returns:

Type	Description
int | list[int]	If index is given: the integer value of that cell.
int | list[int]	If index is None: a list of integers indexed by
int | list[int]	position [0, size). Cells default to 0 (BMv2
int | list[int]	initializes register elements to zero).

Raises:

Type	Description
NoSuchRegisterError	if the register doesn't exist.
EncodingError	if index is out of range.
P4RuntimeError	if the underlying control channel returns an error or this client has no Thrift address configured.

Source code in src/p4net/control/client.py

def read_register(
    self,
    name: str,
    index: int | None = None,
    *,
    timeout: float = 5.0,
) -> int | list[int]:
    """Read a P4 register.

    Args:
        name: Fully qualified P4 register name.
        index: Cell index, or ``None`` to read every cell.

    Returns:
        If ``index`` is given: the integer value of that cell.
        If ``index`` is ``None``: a list of integers indexed by
        position ``[0, size)``. Cells default to ``0`` (BMv2
        initializes register elements to zero).

    Raises:
        NoSuchRegisterError: if the register doesn't exist.
        EncodingError: if ``index`` is out of range.
        P4RuntimeError: if the underlying control channel returns
            an error or this client has no Thrift address configured.
    """
    self._require_connected_with_index()
    idx = self._index
    assert idx is not None
    spec = idx.register_by_name(name)
    if index is not None and not 0 <= int(index) < spec.size:
        raise EncodingError(f"register {name!r} index {index} out of range [0, {spec.size})")
    if index is not None:
        output = self._run_thrift_cli(
            f"register_read {name} {int(index)}",
            timeout=timeout,
            op_description=f"register_read {name}[{index}]",
        )
        return _parse_register_read_single(output, name, int(index))
    output = self._run_thrift_cli(
        f"register_read {name}",
        timeout=timeout,
        op_description=f"register_read {name}",
    )
    return _parse_register_read_array(output, name, spec.size)

add_multicast_group

add_multicast_group(group_id: int, ports: Sequence[int], *, timeout: float = 5.0) -> None

Create a multicast group with one replica per port (instance=1).

Source code in src/p4net/control/client.py

def add_multicast_group(
    self,
    group_id: int,
    ports: Sequence[int],
    *,
    timeout: float = 5.0,
) -> None:
    """Create a multicast group with one replica per port (instance=1)."""
    self._mcast_write(group_id, ports, update_type="INSERT", timeout=timeout)

modify_multicast_group

modify_multicast_group(group_id: int, ports: Sequence[int], *, timeout: float = 5.0) -> None

Replace the replica list of an existing multicast group.

Source code in src/p4net/control/client.py

def modify_multicast_group(
    self,
    group_id: int,
    ports: Sequence[int],
    *,
    timeout: float = 5.0,
) -> None:
    """Replace the replica list of an existing multicast group."""
    self._mcast_write(group_id, ports, update_type="MODIFY", timeout=timeout)

delete_multicast_group

delete_multicast_group(group_id: int, *, timeout: float = 5.0) -> None

Delete a multicast group.

Source code in src/p4net/control/client.py

def delete_multicast_group(
    self,
    group_id: int,
    *,
    timeout: float = 5.0,
) -> None:
    """Delete a multicast group."""
    self._mcast_write(group_id, ports=(), update_type="DELETE", timeout=timeout)

list_multicast_groups

list_multicast_groups(*, timeout: float = 5.0) -> dict[int, list[int]]

Return {group_id: [egress_port, ...]}.

Replica instance numbers are flattened away — each port appears once per replica regardless of its instance value (we always write instance=1 ourselves; foreign instance values are still listed but not exposed in this dict shape).

Source code in src/p4net/control/client.py

def list_multicast_groups(self, *, timeout: float = 5.0) -> dict[int, list[int]]:
    """Return ``{group_id: [egress_port, ...]}``.

    Replica instance numbers are flattened away — each port appears once
    per replica regardless of its instance value (we always write
    instance=1 ourselves; foreign instance values are still listed but
    not exposed in this dict shape).
    """
    self._require_connected()
    req = p4runtime_pb2.ReadRequest()
    req.device_id = self._device_id
    entity = req.entities.add()
    entity.packet_replication_engine_entry.multicast_group_entry.multicast_group_id = 0
    try:
        response_iter = self._stub.Read(req, timeout=timeout)
        groups: dict[int, list[int]] = {}
        for resp in response_iter:
            for ent in resp.entities:
                if not ent.HasField("packet_replication_engine_entry"):
                    continue
                pre = ent.packet_replication_engine_entry
                if not pre.HasField("multicast_group_entry"):
                    continue
                mge = pre.multicast_group_entry
                groups[int(mge.multicast_group_id)] = [int(r.egress_port) for r in mge.replicas]
        return groups
    except grpc.RpcError as exc:
        raise self._translate_rpc_error(exc) from exc

send_packet_out

send_packet_out(payload: bytes, metadata: Mapping[str, object] | None = None, *, timeout: float = 5.0) -> None

Send a PacketOut over the StreamChannel.

payload is the full packet to inject — controller headers are rebuilt from metadata per the loaded P4Info. PacketOut is fire-and-forget in P4Runtime; this method does not wait for a switch response. timeout is reserved for future flow-control limits and currently only bounds the queue put.

Source code in src/p4net/control/client.py

def send_packet_out(
    self,
    payload: bytes,
    metadata: Mapping[str, object] | None = None,
    *,
    timeout: float = 5.0,
) -> None:
    """Send a PacketOut over the StreamChannel.

    ``payload`` is the full packet to inject — controller headers are
    rebuilt from ``metadata`` per the loaded P4Info. PacketOut is
    fire-and-forget in P4Runtime; this method does not wait for a switch
    response. ``timeout`` is reserved for future flow-control limits and
    currently only bounds the queue put.
    """
    self._require_connected()
    idx = self._index
    encoded = idx.encode_packet_out_metadata(metadata or {}) if idx is not None else []
    if metadata and idx is None:
        raise EncodingError(
            "no pipeline is set; cannot encode packet_out metadata "
            "(call set_pipeline_config or get_pipeline_config first)"
        )
    if not isinstance(payload, (bytes, bytearray)):
        raise EncodingError(f"payload must be bytes-like, got {type(payload).__name__}")
    req = p4runtime_pb2.StreamMessageRequest()
    req.packet.payload = bytes(payload)
    for pm in encoded:
        req.packet.metadata.add().CopyFrom(pm)
    out = self._outgoing
    if out is None:
        raise ConnectionError("client is not connected; outgoing queue is closed")
    out.put(req, timeout=timeout)

on_packet_in

on_packet_in(handler: Callable[[bytes, dict[str, int]], None]) -> Callable[[], None]

Register a PacketIn handler. Returns a deregister function.

Handlers run on the StreamChannel consumer thread. Multiple handlers are invoked in registration order; an exception from one is logged and does not prevent later handlers from running. The returned deregister function tolerates double-unregister silently.

Source code in src/p4net/control/client.py

def on_packet_in(
    self,
    handler: Callable[[bytes, dict[str, int]], None],
) -> Callable[[], None]:
    """Register a PacketIn handler. Returns a deregister function.

    Handlers run on the StreamChannel consumer thread. Multiple handlers
    are invoked in registration order; an exception from one is logged
    and does not prevent later handlers from running. The returned
    deregister function tolerates double-unregister silently.
    """
    with self._packet_in_lock:
        self._packet_in_handlers.append(handler)

    def deregister() -> None:
        with self._packet_in_lock, contextlib.suppress(ValueError):
            self._packet_in_handlers.remove(handler)

    return deregister

expect_packet_in

expect_packet_in(*, timeout: float = 5.0) -> tuple[bytes, dict[str, int]]

Block until the next PacketIn arrives. Raises P4RuntimeError on timeout.

Source code in src/p4net/control/client.py

def expect_packet_in(
    self,
    *,
    timeout: float = 5.0,
) -> tuple[bytes, dict[str, int]]:
    """Block until the next PacketIn arrives. Raises ``P4RuntimeError`` on timeout."""
    self._require_connected()
    q: queue.Queue[tuple[bytes, dict[str, int]]] = queue.Queue(maxsize=1)

    def _push(payload: bytes, meta: dict[str, int]) -> None:
        with contextlib.suppress(queue.Full):
            q.put_nowait((payload, meta))

    deregister = self.on_packet_in(_push)
    try:
        try:
            return q.get(timeout=timeout)
        except queue.Empty as exc:
            raise P4RuntimeError(f"no PacketIn within {timeout}s on {self._target!r}") from exc
    finally:
        deregister()

AsyncOperationCancelledError

Bases: P4RuntimeError

An async client operation was cancelled mid-flight.

Raised by :class:p4net.control.AsyncP4RuntimeClient when an in-flight RPC is cancelled (typically because the owning task was cancelled, or because disconnect() is called while another coroutine is awaiting a response). Subclasses :class:P4RuntimeError so existing except P4RuntimeError handlers still cover it, but lets cancellation sites distinguish a clean cancel from a connection failure.

Stable in p4net 1.x since version 1.7.0 — same stability tier as AsyncP4RuntimeClient.

Source code in src/p4net/control/exceptions.py

class AsyncOperationCancelledError(P4RuntimeError):
    """An async client operation was cancelled mid-flight.

    Raised by :class:`p4net.control.AsyncP4RuntimeClient` when an in-flight
    RPC is cancelled (typically because the owning task was cancelled, or
    because ``disconnect()`` is called while another coroutine is awaiting
    a response). Subclasses :class:`P4RuntimeError` so existing
    ``except P4RuntimeError`` handlers still cover it, but lets cancellation
    sites distinguish a clean cancel from a connection failure.

    **Stable** in p4net 1.x since version 1.7.0 — same stability tier as
    ``AsyncP4RuntimeClient``.
    """

ConnectionError

Bases: P4RuntimeError

Failure to open the gRPC channel or complete master arbitration.

Source code in src/p4net/control/exceptions.py

class ConnectionError(P4RuntimeError):
    """Failure to open the gRPC channel or complete master arbitration."""

DuplicateEntryError

Bases: P4RuntimeError

Insert failed because the entry already exists.

Source code in src/p4net/control/exceptions.py

class DuplicateEntryError(P4RuntimeError):
    """Insert failed because the entry already exists."""

EncodingError

Bases: P4RuntimeError

A value could not be encoded for the declared field bitwidth or match type.

Source code in src/p4net/control/exceptions.py

class EncodingError(P4RuntimeError):
    """A value could not be encoded for the declared field bitwidth or match type."""

EntryNotFoundError

Bases: P4RuntimeError

Modify or delete failed because the entry does not exist.

Source code in src/p4net/control/exceptions.py

class EntryNotFoundError(P4RuntimeError):
    """Modify or delete failed because the entry does not exist."""

NoSuchActionError

Bases: P4RuntimeError

Referenced action is not present in the current P4Info.

Source code in src/p4net/control/exceptions.py

class NoSuchActionError(P4RuntimeError):
    """Referenced action is not present in the current P4Info."""

NoSuchFieldError

Bases: P4RuntimeError

Referenced match field or action parameter is not present.

Source code in src/p4net/control/exceptions.py

class NoSuchFieldError(P4RuntimeError):
    """Referenced match field or action parameter is not present."""

NoSuchRegisterError

Bases: P4RuntimeError

Referenced register is not present in the current P4Info.

Source code in src/p4net/control/exceptions.py

class NoSuchRegisterError(P4RuntimeError):
    """Referenced register is not present in the current P4Info."""

NoSuchTableError

Bases: P4RuntimeError

Referenced table is not present in the current P4Info.

Source code in src/p4net/control/exceptions.py

class NoSuchTableError(P4RuntimeError):
    """Referenced table is not present in the current P4Info."""

NotPrimaryError

Bases: P4RuntimeError

The local client is not the primary controller for this device.

Source code in src/p4net/control/exceptions.py

class NotPrimaryError(P4RuntimeError):
    """The local client is not the primary controller for this device."""

P4RuntimeError

Bases: P4NetError

Base class for P4Runtime client failures.

Source code in src/p4net/control/exceptions.py

class P4RuntimeError(P4NetError):
    """Base class for P4Runtime client failures."""

PipelineError

Bases: P4RuntimeError

SetForwardingPipelineConfig was rejected by the switch.

Source code in src/p4net/control/exceptions.py

class PipelineError(P4RuntimeError):
    """SetForwardingPipelineConfig was rejected by the switch."""

P4InfoIndex

Indexed view of a P4Info message with name → id and encoding helpers.

Source code in src/p4net/control/p4info_index.py

class P4InfoIndex:
    """Indexed view of a P4Info message with name → id and encoding helpers."""

    def __init__(self, p4info: Any) -> None:
        self._p4info = p4info
        self._tables_by_name: dict[str, Any] = {t.preamble.name: t for t in p4info.tables}
        self._actions_by_name: dict[str, Any] = {a.preamble.name: a for a in p4info.actions}
        self._counters_by_name: dict[str, Any] = {c.preamble.name: c for c in p4info.counters}
        self._registers_by_name: dict[str, RegisterSpec] = {}
        for r in p4info.registers:
            bitwidth = int(r.type_spec.bitstring.bit.bitwidth)
            self._registers_by_name[r.preamble.name] = RegisterSpec(
                id=int(r.preamble.id),
                name=str(r.preamble.name),
                bitwidth=bitwidth,
                size=int(r.size),
            )
        self._packet_in: Any | None = None
        self._packet_out: Any | None = None
        for cpm in p4info.controller_packet_metadata:
            if cpm.preamble.name == "packet_in":
                self._packet_in = cpm
            elif cpm.preamble.name == "packet_out":
                self._packet_out = cpm

    @classmethod
    def from_file(cls, path: Path) -> P4InfoIndex:
        """Parse a P4Info text-protobuf file."""
        return cls.from_bytes(Path(path).read_bytes(), text_format=True)

    @classmethod
    def from_bytes(cls, data: bytes, *, text_format: bool = True) -> P4InfoIndex:
        """Parse from text or binary protobuf bytes."""
        p4info_pb2 = _import_p4info()
        msg = p4info_pb2.P4Info()
        if text_format:
            from google.protobuf import text_format as _tf

            _tf.Merge(data.decode("utf-8"), msg)
        else:
            msg.ParseFromString(data)
        return cls(msg)

    @property
    def raw(self) -> Any:
        """The underlying ``p4.config.v1.P4Info`` protobuf message."""
        return self._p4info

    @property
    def table_names(self) -> list[str]:
        """Names of every table declared in the P4Info."""
        return list(self._tables_by_name)

    @property
    def action_names(self) -> list[str]:
        """Names of every action declared in the P4Info."""
        return list(self._actions_by_name)

    # Lookup -------------------------------------------------------------

    def table_id(self, name: str) -> int:
        """Return the numeric table ID for ``name``. Raises :class:`NoSuchTableError`."""
        t = self._tables_by_name.get(name)
        if t is None:
            raise NoSuchTableError(f"no table named {name!r}")
        return int(t.preamble.id)

    def action_id(self, name: str) -> int:
        """Return the numeric action ID for ``name``. Raises :class:`NoSuchActionError`."""
        a = self._actions_by_name.get(name)
        if a is None:
            raise NoSuchActionError(f"no action named {name!r}")
        return int(a.preamble.id)

    def counter_id(self, name: str) -> int:
        """Return the numeric counter ID for ``name``. Raises :class:`P4RuntimeError`."""
        c = self._counters_by_name.get(name)
        if c is None:
            raise P4RuntimeError(f"no counter named {name!r}")
        return int(c.preamble.id)

    def table_name(self, table_id: int) -> str:
        """Inverse of :meth:`table_id`."""
        for t in self._tables_by_name.values():
            if t.preamble.id == table_id:
                return str(t.preamble.name)
        raise NoSuchTableError(f"no table with id {table_id}")

    def action_name(self, action_id: int) -> str:
        """Inverse of :meth:`action_id`."""
        for a in self._actions_by_name.values():
            if a.preamble.id == action_id:
                return str(a.preamble.name)
        raise NoSuchActionError(f"no action with id {action_id}")

    def counter_name(self, counter_id: int) -> str:
        """Inverse of :meth:`counter_id`."""
        for c in self._counters_by_name.values():
            if c.preamble.id == counter_id:
                return str(c.preamble.name)
        raise P4RuntimeError(f"no counter with id {counter_id}")

    def register_by_name(self, name: str) -> RegisterSpec:
        """Look up a register by its fully qualified P4 name.

        Args:
            name: Fully qualified P4 name (e.g. ``MyIngress.switch_id``).

        Returns:
            A :class:`RegisterSpec` describing the register.

        Raises:
            NoSuchRegisterError: if no register with that name exists.
        """
        spec = self._registers_by_name.get(name)
        if spec is None:
            raise NoSuchRegisterError(f"no register named {name!r}")
        return spec

    @property
    def register_names(self) -> list[str]:
        """Names of every register declared in the P4Info."""
        return list(self._registers_by_name)

    def table_requires_priority(self, name: str) -> bool:
        """True iff the table has a TERNARY or RANGE match field."""
        t = self._tables_by_name.get(name)
        if t is None:
            raise NoSuchTableError(f"no table named {name!r}")
        p4info_pb2 = _import_p4info()
        return any(
            mf.match_type in (p4info_pb2.MatchField.TERNARY, p4info_pb2.MatchField.RANGE)
            for mf in t.match_fields
        )

    def multicast_group_id_unused(self) -> int:
        """Helper: return 1. The actual selection is delegated to the controller."""
        return 1

    # Encoding -----------------------------------------------------------

    def encode_match(
        self,
        table_name: str,
        match: Mapping[str, object],
    ) -> list[Any]:
        """Build a list of P4Runtime FieldMatch protos for the given table."""
        table = self._tables_by_name.get(table_name)
        if table is None:
            raise NoSuchTableError(f"no table named {table_name!r}")
        p4info_pb2 = _import_p4info()
        p4runtime_pb2 = _import_p4runtime()

        fields_by_name: dict[str, Any] = {mf.name: mf for mf in table.match_fields}
        for key in match:
            if key not in fields_by_name:
                raise NoSuchFieldError(f"field {key!r} not present in table {table_name!r}")
        for mf in table.match_fields:
            if mf.match_type == p4info_pb2.MatchField.EXACT and mf.name not in match:
                raise EncodingError(
                    f"required exact match field {mf.name!r} missing for table {table_name!r}"
                )

        result: list[Any] = []
        for mf_name, value in match.items():
            mf = fields_by_name[mf_name]
            fm = p4runtime_pb2.FieldMatch()
            fm.field_id = int(mf.id)
            bw = int(mf.bitwidth)
            mt = mf.match_type
            if mt == p4info_pb2.MatchField.EXACT:
                fm.exact.value = canonicalize(encode_value(value, bw))  # type: ignore[arg-type]
            elif mt == p4info_pb2.MatchField.LPM:
                encoded, plen = parse_lpm(value, bw)  # type: ignore[arg-type]
                if plen == 0:
                    # P4Runtime: a missing LPM field is wildcard; encoding plen=0 here would
                    # be rejected by the switch. Skip the field instead.
                    continue
                fm.lpm.value = canonicalize(encoded)
                fm.lpm.prefix_len = plen
            elif mt == p4info_pb2.MatchField.TERNARY:
                v, m = parse_ternary(value, bw)  # type: ignore[arg-type]
                if all(b == 0 for b in m):
                    continue  # all-zero mask is wildcard; omit
                fm.ternary.value = canonicalize(v)
                fm.ternary.mask = canonicalize(m)
            elif mt == p4info_pb2.MatchField.RANGE:
                low, high = parse_range(value, bw)  # type: ignore[arg-type]
                fm.range.low = canonicalize(low)
                fm.range.high = canonicalize(high)
            elif mt == p4info_pb2.MatchField.OPTIONAL:
                fm.optional.value = canonicalize(encode_value(value, bw))  # type: ignore[arg-type]
            else:
                raise EncodingError(f"unsupported match type {mt} for field {mf_name!r}")
            result.append(fm)
        return result

    def decode_match(
        self,
        table_name: str,
        match: Mapping[str, object],
    ) -> dict[str, str]:
        """Inverse of `encode_match`: render raw match bytes as human strings.

        For each ``(field_name, raw_value)`` pair the field's bitwidth and
        match type are looked up in the P4Info, then the value is formatted:

        - 32-bit fields → IPv4 dotted-quad (with ``/<plen>`` for LPM,
          ``&<mask>`` for TERNARY, ``[<lo>,<hi>]`` for RANGE).
        - 48-bit fields → MAC ``xx:xx:xx:xx:xx:xx`` (with the same combinators).
        - Other widths → decimal int (with the same combinators).

        Width is taken from P4Info; the bytes may be canonical (shorter than
        width-rounded) and are zero-extended on the high side before decoding.
        """
        table = self._tables_by_name.get(table_name)
        if table is None:
            raise NoSuchTableError(f"no table named {table_name!r}")
        p4info_pb2 = _import_p4info()
        fields_by_name: dict[str, Any] = {mf.name: mf for mf in table.match_fields}
        out: dict[str, str] = {}
        for name, raw in match.items():
            mf = fields_by_name.get(name)
            if mf is None:
                raise NoSuchFieldError(f"field {name!r} not present in table {table_name!r}")
            bw = int(mf.bitwidth)
            mt = mf.match_type
            if mt == p4info_pb2.MatchField.EXACT or mt == p4info_pb2.MatchField.OPTIONAL:
                if not isinstance(raw, bytes):
                    raise EncodingError(
                        f"field {name!r} expected bytes for EXACT/OPTIONAL, "
                        f"got {type(raw).__name__}"
                    )
                out[name] = format_exact(raw, bw)
            elif mt == p4info_pb2.MatchField.LPM:
                if not (isinstance(raw, tuple) and len(raw) == 2):
                    raise EncodingError(
                        f"field {name!r} expected (bytes, plen) for LPM, got {raw!r}"
                    )
                value, plen = raw
                if not isinstance(value, bytes) or not isinstance(plen, int):
                    raise EncodingError(f"field {name!r}: bad LPM tuple {raw!r}")
                out[name] = format_lpm(value, plen, bw)
            elif mt == p4info_pb2.MatchField.TERNARY:
                if not (isinstance(raw, tuple) and len(raw) == 2):
                    raise EncodingError(
                        f"field {name!r} expected (bytes, bytes) for TERNARY, got {raw!r}"
                    )
                v, m = raw
                if not isinstance(v, bytes) or not isinstance(m, bytes):
                    raise EncodingError(f"field {name!r}: bad TERNARY tuple {raw!r}")
                out[name] = format_ternary(v, m, bw)
            elif mt == p4info_pb2.MatchField.RANGE:
                if not (isinstance(raw, tuple) and len(raw) == 2):
                    raise EncodingError(
                        f"field {name!r} expected (bytes, bytes) for RANGE, got {raw!r}"
                    )
                lo, hi = raw
                if not isinstance(lo, bytes) or not isinstance(hi, bytes):
                    raise EncodingError(f"field {name!r}: bad RANGE tuple {raw!r}")
                out[name] = format_range(lo, hi, bw)
            else:
                raise EncodingError(f"unsupported match type {mt} for field {name!r}")
        return out

    def encode_action(
        self,
        action_name: str,
        params: Mapping[str, object] | None = None,
    ) -> Any:
        """Build a P4Runtime Action proto."""
        action = self._actions_by_name.get(action_name)
        if action is None:
            raise NoSuchActionError(f"no action named {action_name!r}")
        p4runtime_pb2 = _import_p4runtime()
        params = params or {}
        params_by_name: dict[str, Any] = {p.name: p for p in action.params}
        for name in params:
            if name not in params_by_name:
                raise NoSuchFieldError(f"param {name!r} not present in action {action_name!r}")
        for p in action.params:
            if p.name not in params:
                raise EncodingError(f"required action param {p.name!r} missing for {action_name!r}")
        action_proto = p4runtime_pb2.Action()
        action_proto.action_id = int(action.preamble.id)
        for p in action.params:
            ap = action_proto.params.add()
            ap.param_id = int(p.id)
            ap.value = canonicalize(encode_value(params[p.name], int(p.bitwidth)))  # type: ignore[arg-type]
        return action_proto

    # Controller packet metadata -----------------------------------------

    def packet_in_metadata_schema(self) -> list[tuple[str, int]]:
        """Return ``[(name, bitwidth), ...]`` for the packet_in controller header.

        Empty list if the program declares no ``@controller_header("packet_in")``.
        """
        if self._packet_in is None:
            return []
        return [(m.name, int(m.bitwidth)) for m in self._packet_in.metadata]

    def packet_out_metadata_schema(self) -> list[tuple[str, int]]:
        """Return ``[(name, bitwidth), ...]`` for the packet_out controller header.

        Empty list if the program declares no ``@controller_header("packet_out")``.
        """
        if self._packet_out is None:
            return []
        return [(m.name, int(m.bitwidth)) for m in self._packet_out.metadata]

    def encode_packet_out_metadata(self, metadata: Mapping[str, object]) -> list[Any]:
        """Encode controller metadata for a PacketOut.

        Each ``(key, value)`` pair is matched against the packet_out schema;
        values go through ``encode_value`` with the schema's bitwidth, then
        wrapped in a ``PacketMetadata`` proto with the corresponding numeric id
        (P4Runtime uses ids on the wire, not field names). Missing keys are
        populated with zero-valued bytes of the correct width. Raises
        ``NoSuchFieldError`` on unknown keys, ``EncodingError`` on width
        overflow.
        """
        if self._packet_out is None:
            if metadata:
                raise NoSuchFieldError(
                    "P4Info declares no packet_out controller header; "
                    f"cannot encode metadata {dict(metadata)!r}"
                )
            return []
        p4runtime_pb2 = _import_p4runtime()
        schema_by_name: dict[str, Any] = {m.name: m for m in self._packet_out.metadata}
        for k in metadata:
            if k not in schema_by_name:
                raise NoSuchFieldError(f"field {k!r} not present in packet_out controller header")
        out: list[Any] = []
        for m in self._packet_out.metadata:
            pm = p4runtime_pb2.PacketMetadata()
            pm.metadata_id = int(m.id)
            value = metadata.get(m.name, 0)
            pm.value = canonicalize(encode_value(value, int(m.bitwidth)))  # type: ignore[arg-type]
            out.append(pm)
        return out

    def decode_packet_in_metadata(self, metadata: Any) -> dict[str, int]:
        """Decode PacketIn controller metadata into ``{field_name: int}``.

        Unknown ids (e.g. switch running a different P4Info than the controller
        has loaded) are silently dropped; the dispatcher logs them at DEBUG.
        """
        if self._packet_in is None:
            return {}
        schema_by_id: dict[int, Any] = {int(m.id): m for m in self._packet_in.metadata}
        out: dict[str, int] = {}
        for pm in metadata:
            mid = int(pm.metadata_id)
            m = schema_by_id.get(mid)
            if m is None:
                continue
            out[str(m.name)] = int.from_bytes(bytes(pm.value), "big")
        return out

raw property

raw: Any

The underlying p4.config.v1.P4Info protobuf message.

table_names property

table_names: list[str]

Names of every table declared in the P4Info.

action_names property

action_names: list[str]

Names of every action declared in the P4Info.

register_names property

register_names: list[str]

Names of every register declared in the P4Info.

from_file classmethod

from_file(path: Path) -> P4InfoIndex

Parse a P4Info text-protobuf file.

Source code in src/p4net/control/p4info_index.py

@classmethod
def from_file(cls, path: Path) -> P4InfoIndex:
    """Parse a P4Info text-protobuf file."""
    return cls.from_bytes(Path(path).read_bytes(), text_format=True)

from_bytes classmethod

from_bytes(data: bytes, *, text_format: bool = True) -> P4InfoIndex

Parse from text or binary protobuf bytes.

Source code in src/p4net/control/p4info_index.py

@classmethod
def from_bytes(cls, data: bytes, *, text_format: bool = True) -> P4InfoIndex:
    """Parse from text or binary protobuf bytes."""
    p4info_pb2 = _import_p4info()
    msg = p4info_pb2.P4Info()
    if text_format:
        from google.protobuf import text_format as _tf

        _tf.Merge(data.decode("utf-8"), msg)
    else:
        msg.ParseFromString(data)
    return cls(msg)

table_id

table_id(name: str) -> int

Return the numeric table ID for name. Raises :class:NoSuchTableError.

Source code in src/p4net/control/p4info_index.py

def table_id(self, name: str) -> int:
    """Return the numeric table ID for ``name``. Raises :class:`NoSuchTableError`."""
    t = self._tables_by_name.get(name)
    if t is None:
        raise NoSuchTableError(f"no table named {name!r}")
    return int(t.preamble.id)

action_id

action_id(name: str) -> int

Return the numeric action ID for name. Raises :class:NoSuchActionError.

Source code in src/p4net/control/p4info_index.py

def action_id(self, name: str) -> int:
    """Return the numeric action ID for ``name``. Raises :class:`NoSuchActionError`."""
    a = self._actions_by_name.get(name)
    if a is None:
        raise NoSuchActionError(f"no action named {name!r}")
    return int(a.preamble.id)

counter_id

counter_id(name: str) -> int

Return the numeric counter ID for name. Raises :class:P4RuntimeError.

Source code in src/p4net/control/p4info_index.py

def counter_id(self, name: str) -> int:
    """Return the numeric counter ID for ``name``. Raises :class:`P4RuntimeError`."""
    c = self._counters_by_name.get(name)
    if c is None:
        raise P4RuntimeError(f"no counter named {name!r}")
    return int(c.preamble.id)

table_name

table_name(table_id: int) -> str

Inverse of :meth:table_id.

Source code in src/p4net/control/p4info_index.py

def table_name(self, table_id: int) -> str:
    """Inverse of :meth:`table_id`."""
    for t in self._tables_by_name.values():
        if t.preamble.id == table_id:
            return str(t.preamble.name)
    raise NoSuchTableError(f"no table with id {table_id}")

action_name

action_name(action_id: int) -> str

Inverse of :meth:action_id.

Source code in src/p4net/control/p4info_index.py

def action_name(self, action_id: int) -> str:
    """Inverse of :meth:`action_id`."""
    for a in self._actions_by_name.values():
        if a.preamble.id == action_id:
            return str(a.preamble.name)
    raise NoSuchActionError(f"no action with id {action_id}")

counter_name

counter_name(counter_id: int) -> str

Inverse of :meth:counter_id.

Source code in src/p4net/control/p4info_index.py

def counter_name(self, counter_id: int) -> str:
    """Inverse of :meth:`counter_id`."""
    for c in self._counters_by_name.values():
        if c.preamble.id == counter_id:
            return str(c.preamble.name)
    raise P4RuntimeError(f"no counter with id {counter_id}")

register_by_name

register_by_name(name: str) -> RegisterSpec

Look up a register by its fully qualified P4 name.

Parameters:

Name	Type	Description	Default
name	str	Fully qualified P4 name (e.g. MyIngress.switch_id).	required

Returns:

Name	Type	Description
A	RegisterSpec	class:RegisterSpec describing the register.

Raises:

Type	Description
NoSuchRegisterError	if no register with that name exists.

Source code in src/p4net/control/p4info_index.py

def register_by_name(self, name: str) -> RegisterSpec:
    """Look up a register by its fully qualified P4 name.

    Args:
        name: Fully qualified P4 name (e.g. ``MyIngress.switch_id``).

    Returns:
        A :class:`RegisterSpec` describing the register.

    Raises:
        NoSuchRegisterError: if no register with that name exists.
    """
    spec = self._registers_by_name.get(name)
    if spec is None:
        raise NoSuchRegisterError(f"no register named {name!r}")
    return spec

table_requires_priority

table_requires_priority(name: str) -> bool

True iff the table has a TERNARY or RANGE match field.

Source code in src/p4net/control/p4info_index.py

def table_requires_priority(self, name: str) -> bool:
    """True iff the table has a TERNARY or RANGE match field."""
    t = self._tables_by_name.get(name)
    if t is None:
        raise NoSuchTableError(f"no table named {name!r}")
    p4info_pb2 = _import_p4info()
    return any(
        mf.match_type in (p4info_pb2.MatchField.TERNARY, p4info_pb2.MatchField.RANGE)
        for mf in t.match_fields
    )

multicast_group_id_unused

multicast_group_id_unused() -> int

Helper: return 1. The actual selection is delegated to the controller.

Source code in src/p4net/control/p4info_index.py

def multicast_group_id_unused(self) -> int:
    """Helper: return 1. The actual selection is delegated to the controller."""
    return 1

encode_match

encode_match(table_name: str, match: Mapping[str, object]) -> list[Any]

Build a list of P4Runtime FieldMatch protos for the given table.

Source code in src/p4net/control/p4info_index.py

def encode_match(
    self,
    table_name: str,
    match: Mapping[str, object],
) -> list[Any]:
    """Build a list of P4Runtime FieldMatch protos for the given table."""
    table = self._tables_by_name.get(table_name)
    if table is None:
        raise NoSuchTableError(f"no table named {table_name!r}")
    p4info_pb2 = _import_p4info()
    p4runtime_pb2 = _import_p4runtime()

    fields_by_name: dict[str, Any] = {mf.name: mf for mf in table.match_fields}
    for key in match:
        if key not in fields_by_name:
            raise NoSuchFieldError(f"field {key!r} not present in table {table_name!r}")
    for mf in table.match_fields:
        if mf.match_type == p4info_pb2.MatchField.EXACT and mf.name not in match:
            raise EncodingError(
                f"required exact match field {mf.name!r} missing for table {table_name!r}"
            )

    result: list[Any] = []
    for mf_name, value in match.items():
        mf = fields_by_name[mf_name]
        fm = p4runtime_pb2.FieldMatch()
        fm.field_id = int(mf.id)
        bw = int(mf.bitwidth)
        mt = mf.match_type
        if mt == p4info_pb2.MatchField.EXACT:
            fm.exact.value = canonicalize(encode_value(value, bw))  # type: ignore[arg-type]
        elif mt == p4info_pb2.MatchField.LPM:
            encoded, plen = parse_lpm(value, bw)  # type: ignore[arg-type]
            if plen == 0:
                # P4Runtime: a missing LPM field is wildcard; encoding plen=0 here would
                # be rejected by the switch. Skip the field instead.
                continue
            fm.lpm.value = canonicalize(encoded)
            fm.lpm.prefix_len = plen
        elif mt == p4info_pb2.MatchField.TERNARY:
            v, m = parse_ternary(value, bw)  # type: ignore[arg-type]
            if all(b == 0 for b in m):
                continue  # all-zero mask is wildcard; omit
            fm.ternary.value = canonicalize(v)
            fm.ternary.mask = canonicalize(m)
        elif mt == p4info_pb2.MatchField.RANGE:
            low, high = parse_range(value, bw)  # type: ignore[arg-type]
            fm.range.low = canonicalize(low)
            fm.range.high = canonicalize(high)
        elif mt == p4info_pb2.MatchField.OPTIONAL:
            fm.optional.value = canonicalize(encode_value(value, bw))  # type: ignore[arg-type]
        else:
            raise EncodingError(f"unsupported match type {mt} for field {mf_name!r}")
        result.append(fm)
    return result

decode_match

decode_match(table_name: str, match: Mapping[str, object]) -> dict[str, str]

Inverse of encode_match: render raw match bytes as human strings.

For each (field_name, raw_value) pair the field's bitwidth and match type are looked up in the P4Info, then the value is formatted:

• 32-bit fields → IPv4 dotted-quad (with /<plen> for LPM, &<mask> for TERNARY, [<lo>,<hi>] for RANGE).
• 48-bit fields → MAC xx:xx:xx:xx:xx:xx (with the same combinators).
• Other widths → decimal int (with the same combinators).

Width is taken from P4Info; the bytes may be canonical (shorter than width-rounded) and are zero-extended on the high side before decoding.

Source code in src/p4net/control/p4info_index.py

def decode_match(
    self,
    table_name: str,
    match: Mapping[str, object],
) -> dict[str, str]:
    """Inverse of `encode_match`: render raw match bytes as human strings.

    For each ``(field_name, raw_value)`` pair the field's bitwidth and
    match type are looked up in the P4Info, then the value is formatted:

    - 32-bit fields → IPv4 dotted-quad (with ``/<plen>`` for LPM,
      ``&<mask>`` for TERNARY, ``[<lo>,<hi>]`` for RANGE).
    - 48-bit fields → MAC ``xx:xx:xx:xx:xx:xx`` (with the same combinators).
    - Other widths → decimal int (with the same combinators).

    Width is taken from P4Info; the bytes may be canonical (shorter than
    width-rounded) and are zero-extended on the high side before decoding.
    """
    table = self._tables_by_name.get(table_name)
    if table is None:
        raise NoSuchTableError(f"no table named {table_name!r}")
    p4info_pb2 = _import_p4info()
    fields_by_name: dict[str, Any] = {mf.name: mf for mf in table.match_fields}
    out: dict[str, str] = {}
    for name, raw in match.items():
        mf = fields_by_name.get(name)
        if mf is None:
            raise NoSuchFieldError(f"field {name!r} not present in table {table_name!r}")
        bw = int(mf.bitwidth)
        mt = mf.match_type
        if mt == p4info_pb2.MatchField.EXACT or mt == p4info_pb2.MatchField.OPTIONAL:
            if not isinstance(raw, bytes):
                raise EncodingError(
                    f"field {name!r} expected bytes for EXACT/OPTIONAL, "
                    f"got {type(raw).__name__}"
                )
            out[name] = format_exact(raw, bw)
        elif mt == p4info_pb2.MatchField.LPM:
            if not (isinstance(raw, tuple) and len(raw) == 2):
                raise EncodingError(
                    f"field {name!r} expected (bytes, plen) for LPM, got {raw!r}"
                )
            value, plen = raw
            if not isinstance(value, bytes) or not isinstance(plen, int):
                raise EncodingError(f"field {name!r}: bad LPM tuple {raw!r}")
            out[name] = format_lpm(value, plen, bw)
        elif mt == p4info_pb2.MatchField.TERNARY:
            if not (isinstance(raw, tuple) and len(raw) == 2):
                raise EncodingError(
                    f"field {name!r} expected (bytes, bytes) for TERNARY, got {raw!r}"
                )
            v, m = raw
            if not isinstance(v, bytes) or not isinstance(m, bytes):
                raise EncodingError(f"field {name!r}: bad TERNARY tuple {raw!r}")
            out[name] = format_ternary(v, m, bw)
        elif mt == p4info_pb2.MatchField.RANGE:
            if not (isinstance(raw, tuple) and len(raw) == 2):
                raise EncodingError(
                    f"field {name!r} expected (bytes, bytes) for RANGE, got {raw!r}"
                )
            lo, hi = raw
            if not isinstance(lo, bytes) or not isinstance(hi, bytes):
                raise EncodingError(f"field {name!r}: bad RANGE tuple {raw!r}")
            out[name] = format_range(lo, hi, bw)
        else:
            raise EncodingError(f"unsupported match type {mt} for field {name!r}")
    return out

encode_action

encode_action(action_name: str, params: Mapping[str, object] | None = None) -> Any

Build a P4Runtime Action proto.

Source code in src/p4net/control/p4info_index.py

def encode_action(
    self,
    action_name: str,
    params: Mapping[str, object] | None = None,
) -> Any:
    """Build a P4Runtime Action proto."""
    action = self._actions_by_name.get(action_name)
    if action is None:
        raise NoSuchActionError(f"no action named {action_name!r}")
    p4runtime_pb2 = _import_p4runtime()
    params = params or {}
    params_by_name: dict[str, Any] = {p.name: p for p in action.params}
    for name in params:
        if name not in params_by_name:
            raise NoSuchFieldError(f"param {name!r} not present in action {action_name!r}")
    for p in action.params:
        if p.name not in params:
            raise EncodingError(f"required action param {p.name!r} missing for {action_name!r}")
    action_proto = p4runtime_pb2.Action()
    action_proto.action_id = int(action.preamble.id)
    for p in action.params:
        ap = action_proto.params.add()
        ap.param_id = int(p.id)
        ap.value = canonicalize(encode_value(params[p.name], int(p.bitwidth)))  # type: ignore[arg-type]
    return action_proto

packet_in_metadata_schema

packet_in_metadata_schema() -> list[tuple[str, int]]

Return [(name, bitwidth), ...] for the packet_in controller header.

Empty list if the program declares no @controller_header("packet_in").

Source code in src/p4net/control/p4info_index.py

def packet_in_metadata_schema(self) -> list[tuple[str, int]]:
    """Return ``[(name, bitwidth), ...]`` for the packet_in controller header.

    Empty list if the program declares no ``@controller_header("packet_in")``.
    """
    if self._packet_in is None:
        return []
    return [(m.name, int(m.bitwidth)) for m in self._packet_in.metadata]

packet_out_metadata_schema

packet_out_metadata_schema() -> list[tuple[str, int]]

Return [(name, bitwidth), ...] for the packet_out controller header.

Empty list if the program declares no @controller_header("packet_out").

Source code in src/p4net/control/p4info_index.py

def packet_out_metadata_schema(self) -> list[tuple[str, int]]:
    """Return ``[(name, bitwidth), ...]`` for the packet_out controller header.

    Empty list if the program declares no ``@controller_header("packet_out")``.
    """
    if self._packet_out is None:
        return []
    return [(m.name, int(m.bitwidth)) for m in self._packet_out.metadata]

encode_packet_out_metadata

encode_packet_out_metadata(metadata: Mapping[str, object]) -> list[Any]

Encode controller metadata for a PacketOut.

Each (key, value) pair is matched against the packet_out schema; values go through encode_value with the schema's bitwidth, then wrapped in a PacketMetadata proto with the corresponding numeric id (P4Runtime uses ids on the wire, not field names). Missing keys are populated with zero-valued bytes of the correct width. Raises NoSuchFieldError on unknown keys, EncodingError on width overflow.

Source code in src/p4net/control/p4info_index.py

def encode_packet_out_metadata(self, metadata: Mapping[str, object]) -> list[Any]:
    """Encode controller metadata for a PacketOut.

    Each ``(key, value)`` pair is matched against the packet_out schema;
    values go through ``encode_value`` with the schema's bitwidth, then
    wrapped in a ``PacketMetadata`` proto with the corresponding numeric id
    (P4Runtime uses ids on the wire, not field names). Missing keys are
    populated with zero-valued bytes of the correct width. Raises
    ``NoSuchFieldError`` on unknown keys, ``EncodingError`` on width
    overflow.
    """
    if self._packet_out is None:
        if metadata:
            raise NoSuchFieldError(
                "P4Info declares no packet_out controller header; "
                f"cannot encode metadata {dict(metadata)!r}"
            )
        return []
    p4runtime_pb2 = _import_p4runtime()
    schema_by_name: dict[str, Any] = {m.name: m for m in self._packet_out.metadata}
    for k in metadata:
        if k not in schema_by_name:
            raise NoSuchFieldError(f"field {k!r} not present in packet_out controller header")
    out: list[Any] = []
    for m in self._packet_out.metadata:
        pm = p4runtime_pb2.PacketMetadata()
        pm.metadata_id = int(m.id)
        value = metadata.get(m.name, 0)
        pm.value = canonicalize(encode_value(value, int(m.bitwidth)))  # type: ignore[arg-type]
        out.append(pm)
    return out

decode_packet_in_metadata

decode_packet_in_metadata(metadata: Any) -> dict[str, int]

Decode PacketIn controller metadata into {field_name: int}.

Unknown ids (e.g. switch running a different P4Info than the controller has loaded) are silently dropped; the dispatcher logs them at DEBUG.

Source code in src/p4net/control/p4info_index.py

def decode_packet_in_metadata(self, metadata: Any) -> dict[str, int]:
    """Decode PacketIn controller metadata into ``{field_name: int}``.

    Unknown ids (e.g. switch running a different P4Info than the controller
    has loaded) are silently dropped; the dispatcher logs them at DEBUG.
    """
    if self._packet_in is None:
        return {}
    schema_by_id: dict[int, Any] = {int(m.id): m for m in self._packet_in.metadata}
    out: dict[str, int] = {}
    for pm in metadata:
        mid = int(pm.metadata_id)
        m = schema_by_id.get(mid)
        if m is None:
            continue
        out[str(m.name)] = int.from_bytes(bytes(pm.value), "big")
    return out

RegisterSpec dataclass

Metadata for a P4 register declared in the pipeline.

Attributes:

Name	Type	Description
id	int	P4Runtime register ID, used in WriteRequest / ReadRequest.
name	str	Fully qualified P4 name (e.g. MyIngress.switch_id).
bitwidth	int	Width of each register element in bits.
size	int	Number of elements in the register array.

Source code in src/p4net/control/p4info_index.py

@dataclass(frozen=True)
class RegisterSpec:
    """Metadata for a P4 register declared in the pipeline.

    Attributes:
        id: P4Runtime register ID, used in WriteRequest / ReadRequest.
        name: Fully qualified P4 name (e.g. ``MyIngress.switch_id``).
        bitwidth: Width of each register element in bits.
        size: Number of elements in the register array.
    """

    id: int
    name: str
    bitwidth: int
    size: int

canonicalize

canonicalize(data: bytes) -> bytes

Return P4Runtime canonical form: strip leading zeros; zero -> b'\x00'.

Source code in src/p4net/control/codec.py

def canonicalize(data: bytes) -> bytes:
    """Return P4Runtime canonical form: strip leading zeros; zero -> b'\\x00'."""
    stripped = data.lstrip(b"\x00")
    return stripped if stripped else b"\x00"

decode_int

decode_int(data: bytes, bitwidth: int) -> int

Inverse of encode_int. Accepts canonical or full-width input.

Source code in src/p4net/control/codec.py

def decode_int(data: bytes, bitwidth: int) -> int:
    """Inverse of `encode_int`. Accepts canonical or full-width input."""
    if not isinstance(data, bytes):
        raise EncodingError(f"data must be bytes, got {type(data).__name__}")
    if bitwidth <= 0:
        raise EncodingError(f"bitwidth must be positive, got {bitwidth}")
    if len(data) > (bitwidth + 7) // 8:
        raise EncodingError(f"byte string of length {len(data)} too wide for bitwidth {bitwidth}")
    return int.from_bytes(data, "big")

decode_ipv4

decode_ipv4(data: bytes) -> str

Inverse of encode_ipv4. Accepts canonical (short) or 4-byte input.

Source code in src/p4net/control/codec.py

def decode_ipv4(data: bytes) -> str:
    """Inverse of `encode_ipv4`. Accepts canonical (short) or 4-byte input."""
    if not isinstance(data, bytes):
        raise EncodingError(f"data must be bytes, got {type(data).__name__}")
    padded = _zero_extend(data, 32)
    return str(ipaddress.IPv4Address(padded))

decode_ipv6

decode_ipv6(data: bytes) -> str

Decode up to 16 canonical bytes as an IPv6 condensed-form string.

Inputs shorter than 16 bytes are zero-extended on the high side, matching P4Runtime canonical encoding (§8.4). The condensed form is whatever :class:ipaddress.IPv6Address.__str__ produces (e.g. "fd00::1").

Source code in src/p4net/control/codec.py

def decode_ipv6(data: bytes) -> str:
    """Decode up to 16 canonical bytes as an IPv6 condensed-form string.

    Inputs shorter than 16 bytes are zero-extended on the high side, matching
    P4Runtime canonical encoding (§8.4). The condensed form is whatever
    :class:`ipaddress.IPv6Address.__str__` produces (e.g. ``"fd00::1"``).
    """
    if not isinstance(data, bytes):
        raise EncodingError(f"data must be bytes, got {type(data).__name__}")
    padded = _zero_extend(data, 128)
    return str(ipaddress.IPv6Address(padded))

decode_mac

decode_mac(data: bytes) -> str

Inverse of encode_mac. Accepts canonical (short) or 6-byte input.

Source code in src/p4net/control/codec.py

def decode_mac(data: bytes) -> str:
    """Inverse of `encode_mac`. Accepts canonical (short) or 6-byte input."""
    if not isinstance(data, bytes):
        raise EncodingError(f"data must be bytes, got {type(data).__name__}")
    padded = _zero_extend(data, 48)
    return ":".join(f"{b:02x}" for b in padded)

encode_int

encode_int(value: int, bitwidth: int) -> bytes

Encode an integer as canonical big-endian bytes for the given bitwidth.

The returned slice is the minimum byte width that holds bitwidth bits. Raises EncodingError if value is negative or does not fit.

Source code in src/p4net/control/codec.py

def encode_int(value: int, bitwidth: int) -> bytes:
    """Encode an integer as canonical big-endian bytes for the given bitwidth.

    The returned slice is the minimum byte width that holds `bitwidth` bits.
    Raises EncodingError if value is negative or does not fit.
    """
    if isinstance(value, bool) or not isinstance(value, int):
        raise EncodingError(f"value must be int, got {type(value).__name__}")
    if value < 0:
        raise EncodingError(f"value {value} must be non-negative")
    if bitwidth <= 0:
        raise EncodingError(f"bitwidth must be positive, got {bitwidth}")
    if value.bit_length() > bitwidth:
        raise EncodingError(f"value {value} does not fit in {bitwidth} bits")
    n_bytes = (bitwidth + 7) // 8
    return value.to_bytes(n_bytes, "big")

encode_ipv4

encode_ipv4(value: str) -> bytes

Encode '10.0.0.1' as 4 big-endian bytes.

Source code in src/p4net/control/codec.py

def encode_ipv4(value: str) -> bytes:
    """Encode '10.0.0.1' as 4 big-endian bytes."""
    if not isinstance(value, str):
        raise EncodingError(f"IPv4 must be a string, got {type(value).__name__}")
    try:
        return ipaddress.IPv4Address(value).packed
    except (ValueError, ipaddress.AddressValueError) as exc:
        raise EncodingError(f"invalid IPv4 address {value!r}") from exc

encode_mac

encode_mac(value: str) -> bytes

Encode 'aa:bb:cc:dd:ee:ff' as 6 bytes.

Source code in src/p4net/control/codec.py

def encode_mac(value: str) -> bytes:
    """Encode 'aa:bb:cc:dd:ee:ff' as 6 bytes."""
    if not isinstance(value, str) or not _MAC_RE.match(value):
        raise EncodingError(f"invalid MAC address {value!r}")
    return bytes(int(part, 16) for part in value.split(":"))

encode_value

encode_value(value: int | str | bytes, bitwidth: int) -> bytes

Auto-dispatch encode for a single field value.

Source code in src/p4net/control/codec.py

def encode_value(value: int | str | bytes, bitwidth: int) -> bytes:
    """Auto-dispatch encode for a single field value."""
    if bitwidth <= 0:
        raise EncodingError(f"bitwidth must be positive, got {bitwidth}")
    if isinstance(value, bytes):
        max_bytes = (bitwidth + 7) // 8
        if len(value) > max_bytes:
            raise EncodingError(
                f"bytes value of length {len(value)} exceeds {max_bytes} for bitwidth {bitwidth}"
            )
        return value
    if isinstance(value, bool):
        return encode_int(int(value), bitwidth)
    if isinstance(value, int):
        return encode_int(value, bitwidth)
    if isinstance(value, str):
        if value.count(".") == 3:
            if bitwidth != 32:
                raise EncodingError(f"IPv4 literal {value!r} requires bitwidth=32, got {bitwidth}")
            return encode_ipv4(value)
        if value.count(":") == 5 and _MAC_RE.match(value):
            if bitwidth != 48:
                raise EncodingError(f"MAC literal {value!r} requires bitwidth=48, got {bitwidth}")
            return encode_mac(value)
        if ":" in value:
            # IPv6 literal (any colon-bearing string that isn't a MAC).
            if bitwidth != 128:
                raise EncodingError(f"IPv6 literal {value!r} requires bitwidth=128, got {bitwidth}")
            try:
                return ipaddress.IPv6Address(value).packed
            except (ValueError, ipaddress.AddressValueError) as exc:
                raise EncodingError(f"invalid IPv6 address {value!r}") from exc
        try:
            n = int(value, 0)
        except ValueError as exc:
            raise EncodingError(f"cannot parse {value!r} as integer") from exc
        return encode_int(n, bitwidth)
    raise EncodingError(f"unsupported value type {type(value).__name__}")

format_exact

format_exact(data: bytes, bitwidth: int) -> str

Render an exact-match value using the width-aware formatter.

Source code in src/p4net/control/codec.py

def format_exact(data: bytes, bitwidth: int) -> str:
    """Render an exact-match value using the width-aware formatter."""
    return _format_value(data, bitwidth)

format_lpm

format_lpm(value: bytes, prefix_len: int, bitwidth: int) -> str

Render an LPM (value, prefix_len) tuple as <addr>/<plen>.

Source code in src/p4net/control/codec.py

def format_lpm(value: bytes, prefix_len: int, bitwidth: int) -> str:
    """Render an LPM (value, prefix_len) tuple as `<addr>/<plen>`."""
    return f"{_format_value(value, bitwidth)}/{int(prefix_len)}"

format_range

format_range(low: bytes, high: bytes, bitwidth: int) -> str

Render a RANGE (low, high) tuple as [<low>,<high>].

Source code in src/p4net/control/codec.py

def format_range(low: bytes, high: bytes, bitwidth: int) -> str:
    """Render a RANGE (low, high) tuple as `[<low>,<high>]`."""
    return f"[{_format_value(low, bitwidth)},{_format_value(high, bitwidth)}]"

format_ternary

format_ternary(value: bytes, mask: bytes, bitwidth: int) -> str

Render a TERNARY (value, mask) tuple as <value>&<mask>.

Source code in src/p4net/control/codec.py

def format_ternary(value: bytes, mask: bytes, bitwidth: int) -> str:
    """Render a TERNARY (value, mask) tuple as `<value>&<mask>`."""
    return f"{_format_value(value, bitwidth)}&{_format_value(mask, bitwidth)}"

parse_lpm

parse_lpm(value: str | tuple[str | int, int], bitwidth: int) -> tuple[bytes, int]

Accept '10.0.0.0/24' or ('10.0.0.0', 24). Returns (encoded_value, prefix_len).

Source code in src/p4net/control/codec.py

def parse_lpm(
    value: str | tuple[str | int, int],
    bitwidth: int,
) -> tuple[bytes, int]:
    """Accept '10.0.0.0/24' or ('10.0.0.0', 24). Returns (encoded_value, prefix_len)."""
    if isinstance(value, str):
        if "/" not in value:
            raise EncodingError(f"LPM string {value!r} must contain '/'")
        addr_part, prefix_part = value.rsplit("/", 1)
        try:
            prefix_len = int(prefix_part)
        except ValueError as exc:
            raise EncodingError(f"invalid LPM prefix {prefix_part!r}") from exc
        addr: int | str = addr_part
    elif isinstance(value, tuple) and len(value) == 2:
        addr, prefix_len = value
        if not isinstance(prefix_len, int) or isinstance(prefix_len, bool):
            raise EncodingError(f"LPM prefix length must be int, got {type(prefix_len).__name__}")
    else:
        raise EncodingError(f"LPM value must be string or 2-tuple, got {value!r}")
    if prefix_len < 0 or prefix_len > bitwidth:
        raise EncodingError(f"LPM prefix_len {prefix_len} out of range [0, {bitwidth}]")
    encoded = encode_value(addr, bitwidth)
    return encoded, prefix_len

parse_range

parse_range(value: tuple[Any, Any], bitwidth: int) -> tuple[bytes, bytes]

Accept (low, high). Returns (encoded_low, encoded_high).

Source code in src/p4net/control/codec.py

def parse_range(
    value: tuple[Any, Any],
    bitwidth: int,
) -> tuple[bytes, bytes]:
    """Accept (low, high). Returns (encoded_low, encoded_high)."""
    if not (isinstance(value, tuple) and len(value) == 2):
        raise EncodingError(f"range value must be a 2-tuple of (low, high), got {value!r}")
    low, high = value
    return encode_value(low, bitwidth), encode_value(high, bitwidth)

parse_ternary

parse_ternary(value: tuple[str | int | bytes, str | int | bytes], bitwidth: int) -> tuple[bytes, bytes]

Accept ('value', 'mask'). Returns (encoded_value, encoded_mask).

Source code in src/p4net/control/codec.py

def parse_ternary(
    value: tuple[str | int | bytes, str | int | bytes],
    bitwidth: int,
) -> tuple[bytes, bytes]:
    """Accept ('value', 'mask'). Returns (encoded_value, encoded_mask)."""
    if not (isinstance(value, tuple) and len(value) == 2):
        raise EncodingError(f"ternary value must be a 2-tuple of (value, mask), got {value!r}")
    val_part, mask_part = value
    enc_val = encode_value(val_part, bitwidth)
    enc_mask = encode_value(mask_part, bitwidth)
    return enc_val, enc_mask