src/developer/fuchsia-controller/python/fidl/_encoder.py

# Copyright 2026 The Fuchsia Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

import struct
import typing


class Encoder:
    def __init__(self) -> None:
        self.buf = bytearray()
        self.handles: list[typing.Any] = []
        self.ool_tasks: list[typing.Callable[[], None]] = []

    def write_inline(self, offset: int, data: bytes) -> None:
        needed = offset + len(data)
        if len(self.buf) < needed:
            self.buf.extend(b"\x00" * (needed - len(self.buf)))
        self.buf[offset : offset + len(data)] = data

    def write_bool(self, val: bool, offset: int) -> None:
        if not isinstance(val, bool):
            raise TypeError(f"Expected bool, got {type(val)}")
        self.write_inline(offset, b"\x01" if val else b"\x00")

    def _write_primitive(
        self, fmt: str, val: typing.Any, offset: int, subtype: str
    ) -> None:
        if val is None:
            raise TypeError(
                f"None value not allowed for primitive subtype {subtype}"
            )
        if "int" in subtype and not isinstance(val, int):
            raise TypeError(f"Expected int, got {type(val)}")
        if "float" in subtype and not isinstance(val, (int, float)):
            raise TypeError(f"Expected float, got {type(val)}")
        try:
            self.write_inline(offset, struct.pack(fmt, val))
        except (struct.error, OverflowError) as e:
            raise OverflowError(
                f"Value {val} out of range for {subtype}"
            ) from e

    def write_int8(self, val: int, offset: int) -> None:
        self._write_primitive("<b", val, offset, "int8")

    def write_uint8(self, val: int, offset: int) -> None:
        self._write_primitive("<B", val, offset, "uint8")

    def write_int16(self, val: int, offset: int) -> None:
        self._write_primitive("<h", val, offset, "int16")

    def write_uint16(self, val: int, offset: int) -> None:
        self._write_primitive("<H", val, offset, "uint16")

    def write_int32(self, val: int, offset: int) -> None:
        self._write_primitive("<i", val, offset, "int32")

    def write_uint32(self, val: int, offset: int) -> None:
        self._write_primitive("<I", val, offset, "uint32")

    def write_int64(self, val: int, offset: int) -> None:
        self._write_primitive("<q", val, offset, "int64")

    def write_uint64(self, val: int, offset: int) -> None:
        self._write_primitive("<Q", val, offset, "uint64")

    def write_float32(self, val: float, offset: int) -> None:
        self._write_primitive("<f", val, offset, "float32")

    def write_float64(self, val: float, offset: int) -> None:
        self._write_primitive("<d", val, offset, "float64")

    def write_handle(
        self,
        val: int | None,
        offset: int,
        obj_type: int = 0,
        rights: int = 0x80000000,
    ) -> None:
        if val is None or val == 0:
            self.write_inline(offset, struct.pack("<I", 0))
        else:
            hdl_val = int(val)
            if hdl_val < 0 or hdl_val > 0xFFFFFFFF:
                raise OverflowError(f"Handle value {hdl_val} out of range")
            self.write_inline(offset, struct.pack("<I", 0xFFFFFFFF))
            self.handles.append((0, hdl_val, obj_type, rights, 0))

    def write_string(self, val: str | None, offset: int) -> None:
        if val is None:
            self.write_inline(offset, struct.pack("<Q", 0))
            self.write_inline(offset + 8, struct.pack("<Q", 0))
        else:
            encoded = val.encode("utf-8")
            self.write_inline(offset, struct.pack("<Q", len(encoded)))
            self.write_inline(offset + 8, struct.pack("<Q", 0xFFFFFFFFFFFFFFFF))
            self.ool_tasks.append(lambda: self.write_ool_bytes(encoded))

    def write_ool_bytes(self, data: bytes) -> None:
        pad_mod = len(data) % 8
        if pad_mod != 0:
            data = data + b"\x00" * (8 - pad_mod)
        self.buf.extend(data)

    def write_vector(
        self,
        val: typing.Sequence[typing.Any] | None,
        offset: int,
        element_encoder_fn: typing.Callable[[typing.Any, "Encoder", int], None],
        element_size: int,
    ) -> None:
        if val is None:
            self.write_inline(offset, struct.pack("<Q", 0))
            self.write_inline(offset + 8, struct.pack("<Q", 0))
        else:
            self.write_inline(offset, struct.pack("<Q", len(val)))
            self.write_inline(offset + 8, struct.pack("<Q", 0xFFFFFFFFFFFFFFFF))
            self.ool_tasks.append(
                lambda: self.write_ool_vector(
                    val, element_encoder_fn, element_size
                )
            )

    def write_ool_vector(
        self,
        val: typing.Sequence[typing.Any],
        element_encoder_fn: typing.Callable[[typing.Any, "Encoder", int], None],
        element_size: int,
    ) -> None:
        parent_tasks = self.ool_tasks
        self.ool_tasks = []

        pad_mod = len(self.buf) % 8
        if pad_mod != 0:
            self.buf.extend(b"\x00" * (8 - pad_mod))
        start_offset = len(self.buf)
        self.buf.extend(b"\x00" * (element_size * len(val)))
        for i, item in enumerate(val):
            element_encoder_fn(item, self, start_offset + i * element_size)

        pad_mod = len(self.buf) % 8
        if pad_mod != 0:
            self.buf.extend(b"\x00" * (8 - pad_mod))

        i = 0
        while i < len(self.ool_tasks):
            self.ool_tasks[i]()
            i += 1
        self.ool_tasks = parent_tasks

    def write_array(
        self,
        val: typing.Sequence[typing.Any],
        offset: int,
        element_encoder_fn: typing.Callable[[typing.Any, "Encoder", int], None],
        element_size: int,
        count: int,
    ) -> None:
        if len(val) != count:
            raise ValueError(
                f"Array length {len(val)} does not match expected count {count}"
            )
        for i in range(count):
            element_encoder_fn(val[i], self, offset + i * element_size)

    def write_optional_struct(
        self,
        val: typing.Any,
        offset: int,
        encode_fn: typing.Callable[[typing.Any, "Encoder", int], None],
        inline_size: int,
    ) -> None:
        if val is None:
            self.write_inline(offset, struct.pack("<Q", 0))
        else:
            self.write_inline(offset, struct.pack("<Q", 0xFFFFFFFFFFFFFFFF))
            self.ool_tasks.append(
                lambda: self.write_ool_struct(val, encode_fn, inline_size)
            )

    def write_ool_struct(
        self,
        val: typing.Any,
        encode_fn: typing.Callable[[typing.Any, "Encoder", int], None],
        inline_size: int,
    ) -> None:
        parent_tasks = self.ool_tasks
        self.ool_tasks = []

        pad_mod = len(self.buf) % 8
        if pad_mod != 0:
            self.buf.extend(b"\x00" * (8 - pad_mod))
        start_offset = len(self.buf)
        self.buf.extend(b"\x00" * inline_size)
        encode_fn(val, self, start_offset)

        pad_mod = len(self.buf) % 8
        if pad_mod != 0:
            self.buf.extend(b"\x00" * (8 - pad_mod))

        i = 0
        while i < len(self.ool_tasks):
            self.ool_tasks[i]()
            i += 1
        self.ool_tasks = parent_tasks

    def write_envelope(
        self,
        val: typing.Any,
        env_offset: int,
        encode_fn: typing.Callable[[typing.Any, "Encoder", int], None],
        inline_size: int,
        is_inline: bool,
    ) -> None:
        if val is None:
            self.write_inline(env_offset, b"\x00" * 8)
            return

        if is_inline:
            temp_encoder = Encoder()
            encode_fn(val, temp_encoder, 0)
            temp_bytes = bytes(temp_encoder.buf)
            if len(temp_bytes) < 4:
                temp_bytes = temp_bytes + b"\x00" * (4 - len(temp_bytes))
            elif len(temp_bytes) > 4:
                raise RuntimeError("Inlined value exceeds 4 bytes")
            self.write_inline(env_offset, temp_bytes)
            self.handles.extend(temp_encoder.handles)
            self.write_inline(
                env_offset + 4, struct.pack("<H", len(temp_encoder.handles))
            )
            self.write_inline(env_offset + 6, struct.pack("<H", 1))
        else:
            self.write_inline(env_offset, struct.pack("<I", 0))
            self.write_inline(env_offset + 4, struct.pack("<H", 0))
            self.write_inline(env_offset + 6, struct.pack("<H", 0))
            self.ool_tasks.append(
                lambda: self.write_ool_envelope_member(
                    val, encode_fn, inline_size, env_offset
                )
            )

    def write_ool_envelope_member(
        self,
        val: typing.Any,
        encode_fn: typing.Callable[[typing.Any, "Encoder", int], None],
        inline_size: int,
        env_offset: int,
    ) -> None:
        parent_tasks = self.ool_tasks
        self.ool_tasks = []

        pad_mod = len(self.buf) % 8
        if pad_mod != 0:
            self.buf.extend(b"\x00" * (8 - pad_mod))
        start_offset = len(self.buf)
        self.buf.extend(b"\x00" * inline_size)

        start_handles = len(self.handles)
        encode_fn(val, self, start_offset)

        pad_mod = len(self.buf) % 8
        if pad_mod != 0:
            self.buf.extend(b"\x00" * (8 - pad_mod))

        i = 0
        while i < len(self.ool_tasks):
            self.ool_tasks[i]()
            i += 1

        total_size = len(self.buf) - start_offset
        total_handles = len(self.handles) - start_handles

        self.write_inline(env_offset, struct.pack("<I", total_size))
        self.write_inline(env_offset + 4, struct.pack("<H", total_handles))

        self.ool_tasks = parent_tasks


def encode_struct(
    obj: typing.Any,
    encode_fn: typing.Callable[[typing.Any, Encoder, int], None],
    inline_size: int,
) -> tuple[bytes, list[typing.Any]]:
    encoder = Encoder()
    encoder.buf.extend(b"\x00" * inline_size)
    encode_fn(obj, encoder, 0)
    i = 0
    while i < len(encoder.ool_tasks):
        encoder.ool_tasks[i]()
        i += 1
    # Pad final buffer to 8 bytes alignment
    pad_mod = len(encoder.buf) % 8
    if pad_mod != 0:
        encoder.buf.extend(b"\x00" * (8 - pad_mod))
    return bytes(encoder.buf), encoder.handles