codegen.py

Go to the documentation of this file.

#  Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions.
#  See https://llvm.org/LICENSE.txt for license information.
#  SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
"""Code generation from ESI manifests to source code.
 
Uses a two-pass approach for C++: first collect and name all reachable types,
then emit structs/aliases in a dependency-ordered sequence so headers are
standalone and deterministic.
"""
 
# C++ header support included with the runtime, though it is intended to be
# extensible for other languages.
 
from typing import Dict, List, Set, TextIO, Tuple, Type, Optional
from .accelerator import AcceleratorConnection, Context
from .esiCppAccel import ModuleInfo
from . import types
 
import sys
import os
import textwrap
import argparse
from pathlib import Path
 
_thisdir = Path(__file__).absolute().resolve().parent
 
 
class Generator:
  """Base class for all generators."""
 
  language: Optional[str] = None
 
  def __init__(self, conn: AcceleratorConnection):
    self.manifest = conn.manifest()
 
  def generate(self, output_dir: Path, system_name: str):
    raise NotImplementedError("Generator.generate() must be overridden")
 
 
class CppGenerator(Generator):
  """Generate C++ headers from an ESI manifest."""
 
  language = "C++"
 
  def __init__(self, conn: AcceleratorConnection):
    super().__init__(conn)
    self.type_planner = CppTypePlanner(self.manifest.type_table)
    self.type_emitter = CppTypeEmitter(self.type_planner)
 
  def get_consts_str(self, module_info: ModuleInfo) -> str:
    """Get the C++ code for a constant in a module."""
    const_strs: List[str] = [
        f"static constexpr {self.type_emitter.type_identifier(const.type)} "
        f"{name} = 0x{const.value:x};"
        for name, const in module_info.constants.items()
    ]
    return "\n".join(const_strs)
 
  def write_modules(self, output_dir: Path, system_name: str):
    """Write the C++ header. One for each module in the manifest."""
 
    for module_info in self.manifest.module_infos:
      s = f"""
      /// Generated header for {system_name} module {module_info.name}.
      #pragma once
      #include "types.h"
 
      namespace {system_name} {{
      class {module_info.name} {{
      public:
        {self.get_consts_str(module_info)}
      }};
      }} // namespace {system_name}
      """
 
      hdr_file = output_dir / f"{module_info.name}.h"
      with open(hdr_file, "w") as hdr:
        hdr.write(textwrap.dedent(s))
 
  def generate(self, output_dir: Path, system_name: str):
    self.type_emitter.write_header(output_dir, system_name)
    self.write_modules(output_dir, system_name)
 
 
class CppTypePlanner:
  """Plan C++ type naming and ordering from an ESI manifest."""
 
  def __init__(self, type_table) -> None:
    """Initialize the generator with the manifest and target namespace."""
    # Map manifest type ids to their preferred C++ names.
    self.type_id_map: Dict[types.ESIType, str] = {}
    # Track all names already taken to avoid collisions. True => alias-based.
    self.used_names: Dict[str, bool] = {}
    # Track alias base names to warn on collisions.
    self.alias_base_names: Set[str] = set()
    self.ordered_types: List[types.ESIType] = []
    self.has_cycle = False
    self._prepare_types(type_table)
 
  def _prepare_types(self, type_table) -> None:
    """Name the types and prepare for emission by registering all reachable
    types and assigning."""
    visited: Set[str] = set()
    for t in type_table:
      self._collect_aliases(t, visited)
 
    visited = set()
    for t in type_table:
      self._collect_structs(t, visited)
 
    visited = set()
    for t in type_table:
      self._collect_windows(t, visited)
 
    self.ordered_types, self.has_cycle = self._ordered_emit_types()
 
  def _sanitize_name(self, name: str) -> str:
    """Create a C++-safe identifier from the manifest-provided name."""
    name = name.replace("::", "_")
    if name.startswith("@"):
      name = name[1:]
    sanitized = []
    for ch in name:
      if ch.isalnum() or ch == "_":
        sanitized.append(ch)
      else:
        sanitized.append("_")
    if not sanitized:
      return "Type"
    if sanitized[0].isdigit():
      sanitized.insert(0, "_")
    return "".join(sanitized)
 
  def _reserve_name(self, base: str, is_alias: bool) -> str:
    """Reserve a globally unique identifier using the sanitized base name."""
    base = self._sanitize_name(base)
    if is_alias and base in self.alias_base_names:
      sys.stderr.write(
          f"Warning: duplicate alias name '{base}' detected; disambiguating.\n")
    if is_alias:
      self.alias_base_names.add(base)
    name = base
    idx = 1
    while name in self.used_names:
      name = f"{base}_{idx}"
      idx += 1
    self.used_names[name] = is_alias
    return name
 
  def _auto_struct_name(self, struct_type: types.StructType) -> str:
    """Derive a deterministic name for anonymous structs from their fields."""
    parts = ["_struct"]
    for field_name, field_type in struct_type.fields:
      parts.append(field_name)
      parts.append(self._sanitize_name(field_type.id))
    return self._reserve_name("_".join(parts), is_alias=False)
 
  def _auto_union_name(self, union_type: types.UnionType) -> str:
    """Derive a deterministic name for anonymous unions from their fields."""
    parts = ["_union"]
    for field_name, field_type in union_type.fields:
      parts.append(field_name)
      parts.append(self._sanitize_name(field_type.id))
    return self._reserve_name("_".join(parts), is_alias=False)
 
  def _auto_window_name(self, window_type: types.WindowType) -> str:
    """Derive a deterministic name for generated window helpers."""
    if window_type.name:
      return self._reserve_name(window_type.name, is_alias=False)
    return self._reserve_name(f"_window_{self._sanitize_name(window_type.id)}",
                              is_alias=False)
 
  def _unwrap_aliases(self, wrapped: types.ESIType) -> types.ESIType:
    while isinstance(wrapped, types.TypeAlias):
      wrapped = wrapped.inner_type
    return wrapped
 
  def _is_supported_window(self, current_type: types.ESIType) -> bool:
    if not isinstance(current_type, types.WindowType):
      return False
    into_type = self._unwrap_aliases(current_type.into_type)
    if not isinstance(into_type, types.StructType):
      return False
 
    # The generated window helper only supports struct-shaped payloads with a
    # single logical list field to stream across multiple frames.
    list_fields = []
    for field_name, field_type in into_type.fields:
      if isinstance(self._unwrap_aliases(field_type), types.ListType):
        list_fields.append(field_name)
    if len(list_fields) != 1:
      return False
 
    list_field_name = list_fields[0]
    header_field = None
    data_field = None
    # That list must appear exactly once as a bulk-count field and exactly once
    # as a single-item data field so the helper can synthesize header/data/footer.
    for frame in current_type.frames:
      for field in frame.fields:
        if field.name != list_field_name:
          continue
        if field.bulk_count_width > 0:
          if header_field is not None:
            return False
          header_field = field
        elif field.num_items > 0:
          if data_field is not None:
            return False
          data_field = field
    return (header_field is not None and data_field is not None and
            data_field.num_items == 1)
 
  def _iter_type_children(self, t: types.ESIType) -> List[types.ESIType]:
    """Return child types in a stable order for traversal."""
    if isinstance(t, types.TypeAlias):
      return [t.inner_type] if t.inner_type is not None else []
    if isinstance(t, types.BundleType):
      return [channel.type for channel in t.channels]
    if isinstance(t, types.ChannelType):
      return [t.inner]
    if isinstance(t, types.StructType):
      return [field_type for _, field_type in t.fields]
    if isinstance(t, types.UnionType):
      return [field_type for _, field_type in t.fields]
    if isinstance(t, types.ListType):
      return [t.element_type]
    if isinstance(t, types.WindowType):
      return [t.into_type]
    if isinstance(t, types.ArrayType):
      return [t.element_type]
    return []
 
  def _visit_types(self, t: types.ESIType, visited: Set[str], visit_fn) -> None:
    """Traverse types with alphabetical child ordering in post-order."""
    if not isinstance(t, types.ESIType):
      raise TypeError(f"Expected ESIType, got {type(t)}")
    tid = t.id
    if tid in visited:
      return
    visited.add(tid)
    children = sorted(self._iter_type_children(t), key=lambda child: child.id)
    for child in children:
      self._visit_types(child, visited, visit_fn)
    visit_fn(t)
 
  def _collect_aliases(self, t: types.ESIType, visited: Set[str]) -> None:
    """Scan for aliases and reserve their names (recursive)."""
 
    # Visit callback: reserve alias names and map aliases to identifiers.
    def visit(alias_type: types.ESIType) -> None:
      if not isinstance(alias_type, types.TypeAlias):
        return
      if alias_type not in self.type_id_map:
        alias_name = self._reserve_name(alias_type.name, is_alias=True)
        self.type_id_map[alias_type] = alias_name
 
    self._visit_types(t, visited, visit)
 
  def _collect_structs(self, t: types.ESIType, visited: Set[str]) -> None:
    """Scan for structs/unions needing auto-names and reserve them."""
 
    # Visit callback: assign auto-names to unnamed structs and unions.
    def visit(current_type: types.ESIType) -> None:
      if current_type in self.type_id_map:
        return
      if isinstance(current_type, types.StructType):
        self.type_id_map[current_type] = self._auto_struct_name(current_type)
      elif isinstance(current_type, types.UnionType):
        self.type_id_map[current_type] = self._auto_union_name(current_type)
 
    self._visit_types(t, visited, visit)
 
  def _collect_windows(self, t: types.ESIType, visited: Set[str]) -> None:
    """Scan for supported window types and reserve helper names."""
 
    def visit(current_type: types.ESIType) -> None:
      if not self._is_supported_window(current_type):
        return
      assert isinstance(current_type, types.WindowType)
      if current_type in self.type_id_map:
        return
      self.type_id_map[current_type] = self._auto_window_name(current_type)
 
    self._visit_types(t, visited, visit)
 
  def _collect_decls_from_type(self,
                               wrapped: types.ESIType) -> Set[types.ESIType]:
    """Collect types that require top-level declarations for a given type."""
    deps: Set[types.ESIType] = set()
 
    # Visit callback: collect structs, unions, and non-struct aliases used by a
    # type.
    def visit(current: types.ESIType) -> None:
      if isinstance(current, types.TypeAlias):
        inner = current.inner_type
        if inner is not None and (isinstance(
            inner, (types.StructType, types.UnionType)) or
                                  self._is_supported_window(inner)):
          deps.add(inner)
        else:
          deps.add(current)
      elif isinstance(current, (types.StructType, types.UnionType)):
        deps.add(current)
      elif self._is_supported_window(current):
        deps.add(current)
 
    self._visit_types(wrapped, set(), visit)
    return deps
 
  def _collect_decls_from_window(
      self, window_type: types.WindowType) -> Set[types.ESIType]:
    """Collect only the declarations referenced by a generated window helper."""
    deps: Set[types.ESIType] = set()
    into_type = self._unwrap_aliases(window_type.into_type)
    if not isinstance(into_type, types.StructType):
      return deps
 
    for _, field_type in into_type.fields:
      unwrapped = self._unwrap_aliases(field_type)
      if isinstance(unwrapped, types.ListType):
        deps.update(self._collect_decls_from_type(unwrapped.element_type))
      else:
        deps.update(self._collect_decls_from_type(field_type))
    return deps
 
  def _ordered_emit_types(self) -> Tuple[List[types.ESIType], bool]:
    """Collect and order types for deterministic emission."""
    window_into_types: Set[types.ESIType] = set()
    for esi_type in self.type_id_map.keys():
      if not self._is_supported_window(esi_type):
        continue
      assert isinstance(esi_type, types.WindowType)
      window_into_types.add(self._unwrap_aliases(esi_type.into_type))
    emit_types: List[types.ESIType] = []
    for esi_type in self.type_id_map.keys():
      if isinstance(esi_type,
                    types.StructType) and esi_type in window_into_types:
        continue
      if isinstance(esi_type, types.TypeAlias):
        inner = esi_type.inner_type
        if inner is not None and self._unwrap_aliases(
            inner) in window_into_types:
          continue
      if (isinstance(esi_type,
                     (types.StructType, types.UnionType, types.TypeAlias)) or
          self._is_supported_window(esi_type)):
        emit_types.append(esi_type)
 
    # Prefer alias-reserved names first, then lexicographic for determinism.
    name_to_type = {self.type_id_map[t]: t for t in emit_types}
    sorted_names = sorted(name_to_type.keys(),
                          key=lambda name:
                          (0 if self.used_names.get(name, False) else 1, name))
 
    ordered: List[types.ESIType] = []
    visited: Set[types.ESIType] = set()
    visiting: Set[types.ESIType] = set()
    has_cycle = False
 
    # Visit callback: DFS to emit dependencies before their users.
    def visit(current: types.ESIType) -> None:
      nonlocal has_cycle
      if current in visited:
        return
      if current in visiting:
        has_cycle = True
        return
      visiting.add(current)
 
      deps: Set[types.ESIType] = set()
      if isinstance(current, types.TypeAlias):
        inner = current.inner_type
        if inner is not None:
          deps.update(self._collect_decls_from_type(inner))
      elif isinstance(current, (types.StructType, types.UnionType)):
        for _, field_type in current.fields:
          deps.update(self._collect_decls_from_type(field_type))
      elif self._is_supported_window(current):
        assert isinstance(current, types.WindowType)
        deps.update(self._collect_decls_from_window(current))
      for dep in sorted(deps, key=lambda dep: self.type_id_map[dep]):
        visit(dep)
 
      visiting.remove(current)
      visited.add(current)
      ordered.append(current)
 
    for name in sorted_names:
      visit(name_to_type[name])
 
    return ordered, has_cycle
 
 
class CppTypeEmitter:
  """Emit C++ headers from precomputed type ordering."""
 
  def __init__(self, planner: CppTypePlanner) -> None:
    self.type_id_map = planner.type_id_map
    self.ordered_types = planner.ordered_types
    self.has_cycle = planner.has_cycle
 
  def type_identifier(self, type: types.ESIType) -> str:
    """Get the C++ type string for an ESI type."""
    return self._cpp_type(type)
 
  def _cpp_string_literal(self, value: str) -> str:
    """Escape a Python string for use as a C++ string literal."""
    escaped = value.replace("\\", "\\\\").replace('"', '\\"')
    return f'"{escaped}"'
 
  def _get_bitvector_str(self, type: types.ESIType) -> str:
    """Get the textual code for the storage class of an integer type."""
    assert isinstance(type, (types.BitsType, types.IntType))
 
    return self._storage_type(
        type.bit_width, not isinstance(type, (types.BitsType, types.UIntType)))
 
  def _storage_type(self, bit_width: int, signed: bool) -> str:
    """Get the textual code for a byte-addressable integer storage type."""
 
    if bit_width == 1:
      return "bool"
    elif bit_width <= 8:
      storage_width = 8
    elif bit_width <= 16:
      storage_width = 16
    elif bit_width <= 32:
      storage_width = 32
    elif bit_width <= 64:
      storage_width = 64
    else:
      raise ValueError(f"Unsupported integer width: {bit_width}")
 
    if not signed:
      return f"uint{storage_width}_t"
    return f"int{storage_width}_t"
 
  def _type_byte_width(self, wrapped: types.ESIType) -> int:
    """Return the size of a fixed-width type in bytes."""
    if wrapped.bit_width < 0:
      raise ValueError(f"Unsupported unbounded type width for '{wrapped}'")
    return (wrapped.bit_width + 7) // 8
 
  def _array_base_and_suffix(self,
                             array_type: types.ArrayType) -> Tuple[str, str]:
    """Return the base C++ type and bracket suffix for a nested array."""
    dims: List[int] = []
    inner: types.ESIType = array_type
    while isinstance(inner, types.ArrayType):
      dims.append(inner.size)
      inner = inner.element_type
    base_cpp = self._cpp_type(inner)
    suffix = "".join([f"[{d}]" for d in dims])
    return base_cpp, suffix
 
  def _format_array_type(self, array_type: types.ArrayType) -> str:
    """Return the C++ type string for a nested array alias."""
    base_cpp, suffix = self._array_base_and_suffix(array_type)
    return f"{base_cpp}{suffix}"
 
  def _cpp_type(self, wrapped: types.ESIType) -> str:
    """Resolve an ESI type to its C++ identifier."""
    if isinstance(wrapped, types.WindowType) and wrapped in self.type_id_map:
      return self.type_id_map[wrapped]
    if isinstance(wrapped,
                  (types.TypeAlias, types.StructType, types.UnionType)):
      return self.type_id_map[wrapped]
    if isinstance(wrapped, types.BundleType):
      return "void"
    if isinstance(wrapped, types.ChannelType):
      return self._cpp_type(wrapped.inner)
    if isinstance(wrapped, types.ListType):
      raise ValueError("List types require a generated window wrapper")
    if isinstance(wrapped, types.VoidType):
      return "void"
    if isinstance(wrapped, types.AnyType):
      return "std::any"
    if isinstance(wrapped, (types.BitsType, types.IntType)):
      return self._get_bitvector_str(wrapped)
    if isinstance(wrapped, types.ArrayType):
      return self._format_array_type(wrapped)
    if type(wrapped) is types.ESIType:
      return "std::any"
    raise NotImplementedError(
        f"Type '{wrapped}' not supported for C++ generation")
 
  def _unwrap_aliases(self, wrapped: types.ESIType) -> types.ESIType:
    """Strip alias wrappers to reach the underlying type."""
    while isinstance(wrapped, types.TypeAlias):
      wrapped = wrapped.inner_type
    return wrapped
 
  def _format_array_decl(self, array_type: types.ArrayType, name: str) -> str:
    """Emit a field declaration for a multi-dimensional array member.
 
    The declaration flattens nested arrays into repeated bracketed sizes for C++.
    """
    base_cpp, suffix = self._array_base_and_suffix(array_type)
    return f"{base_cpp} {name}{suffix};"
 
  def _format_window_field_decl(self, field_name: str,
                                field_type: types.ESIType) -> str:
    """Emit a packed field declaration for generated window helpers."""
    if isinstance(field_type, types.ArrayType):
      return self._format_array_decl(field_type, field_name)
    field_cpp = self._cpp_type(field_type)
    wrapped = self._unwrap_aliases(field_type)
    if isinstance(wrapped, (types.BitsType, types.IntType)) and \
       wrapped.bit_width % 8 != 0:
      return f"{field_cpp} {field_name} : {wrapped.bit_width};"
    return f"{field_cpp} {field_name};"
 
  def _format_window_ctor_param(self, field_name: str,
                                field_type: types.ESIType) -> str:
    """Emit a constructor parameter for generated window helpers.
 
    Small scalar header fields are cheaper to pass by value than by reference.
    Larger aggregates stay as const references.
    """
    if isinstance(field_type, types.ArrayType):
      base_cpp, suffix = self._array_base_and_suffix(field_type)
      return f"const {base_cpp} (&{field_name}){suffix}"
    field_cpp = self._cpp_type(field_type)
    wrapped = self._unwrap_aliases(field_type)
    if isinstance(wrapped, (types.BitsType, types.IntType)):
      return f"{field_cpp} {field_name}"
    return f"const {field_cpp} &{field_name}"
 
  def _emit_window_field_copy(self, hdr: TextIO, dest_expr: str, src_expr: str,
                              field_type: types.ESIType) -> None:
    """Copy a generated window field, preserving array semantics."""
    if isinstance(field_type, types.ArrayType):
      hdr.write(
          f"    std::memcpy(&{dest_expr}, &{src_expr}, sizeof({dest_expr}));\n")
      return
    hdr.write(f"    {dest_expr} = {src_expr};\n")
 
  def _field_byte_width(self, field_type: types.ESIType) -> int:
    """Compute the byte width of a field type, rounding up to full bytes."""
    return (field_type.bit_width + 7) // 8
 
  def _analyze_window(self, window_type: types.WindowType):
    """Extract the metadata needed to emit a bulk list window wrapper."""
    into_type = self._unwrap_aliases(window_type.into_type)
    if not isinstance(into_type, types.StructType):
      raise ValueError("window codegen currently requires a struct into-type")
 
    field_map = {name: field_type for name, field_type in into_type.fields}
    list_fields = [
        (name, self._unwrap_aliases(field_type))
        for name, field_type in into_type.fields
        if isinstance(self._unwrap_aliases(field_type), types.ListType)
    ]
    if len(list_fields) != 1:
      raise ValueError("window codegen currently supports exactly one list")
 
    list_field_name, list_type = list_fields[0]
    assert isinstance(list_type, types.ListType)
 
    header_frame = None
    header_field = None
    data_frame = None
    data_field = None
    for frame in window_type.frames:
      for field in frame.fields:
        if field.name != list_field_name:
          continue
        if field.bulk_count_width > 0:
          header_frame = frame
          header_field = field
        elif field.num_items > 0:
          data_frame = frame
          data_field = field
 
    if header_frame is None or header_field is None:
      raise ValueError("window codegen requires a bulk-count header frame")
    if data_frame is None or data_field is None:
      raise ValueError("window codegen requires a data frame for the list")
    if data_field.num_items != 1:
      raise ValueError("window codegen currently supports numItems == 1")
 
    ctor_params = [(name, field_type)
                   for name, field_type in into_type.fields
                   if name != list_field_name]
 
    header_fields = []
    header_bytes = 0
    count_field_name = f"{list_field_name}_count"
    count_width = header_field.bulk_count_width
    count_cpp = self._storage_type(count_width, signed=False)
    count_bytes = (count_width + 7) // 8
    for field in reversed(header_frame.fields):
      if field.name == list_field_name:
        header_fields.append((count_field_name, None))
        header_bytes += count_bytes
      else:
        field_type = field_map[field.name]
        header_fields.append((field.name, field_type))
        header_bytes += self._type_byte_width(field_type)
 
    data_fields = []
    data_bytes = 0
    for field in reversed(data_frame.fields):
      if field.name == list_field_name:
        data_fields.append((list_field_name, list_type.element_type))
        data_bytes += self._type_byte_width(list_type.element_type)
      else:
        field_type = field_map[field.name]
        data_fields.append((field.name, field_type))
        data_bytes += self._type_byte_width(field_type)
 
    frame_bytes = max(header_bytes, data_bytes)
 
    return {
        "ctor_params": ctor_params,
        "count_cpp": count_cpp,
        "count_field_name": count_field_name,
        "count_width": count_width,
        "data_fields": data_fields,
        "data_pad_bytes": frame_bytes - data_bytes,
        "element_cpp": self._cpp_type(list_type.element_type),
        "header_fields": header_fields,
        "header_pad_bytes": frame_bytes - header_bytes,
        "list_field_name": list_field_name,
        "window_name": self.type_id_map[window_type],
    }
 
  def _emit_struct(self, hdr: TextIO, struct_type: types.StructType) -> None:
    """Emit a packed struct declaration plus its type id string."""
    fields = list(struct_type.fields)
    if struct_type.cpp_type.reverse:
      fields = list(reversed(fields))
    field_decls: List[str] = []
    for field_name, field_type in fields:
      if isinstance(field_type, types.ArrayType):
        field_decls.append(self._format_array_decl(field_type, field_name))
      else:
        field_cpp = self._cpp_type(field_type)
        wrapped = self._unwrap_aliases(field_type)
        if isinstance(wrapped, (types.BitsType, types.IntType)):
          # TODO: Bitfield layout is implementation-defined; consider
          # byte-aligned storage with explicit pack/unpack helpers.
          bitfield_width = wrapped.bit_width
          if bitfield_width >= 0:
            field_decls.append(f"{field_cpp} {field_name} : {bitfield_width};")
          else:
            field_decls.append(f"{field_cpp} {field_name};")
        else:
          field_decls.append(f"{field_cpp} {field_name};")
    hdr.write(f"struct {self.type_id_map[struct_type]} {{\n")
    for decl in field_decls:
      hdr.write(f"  {decl}\n")
    hdr.write("\n")
    hdr.write(
        f"  static constexpr std::string_view _ESI_ID = {self._cpp_string_literal(struct_type.id)};\n"
    )
    hdr.write("};\n\n")
 
  def _emit_union(self, hdr: TextIO, union_type: types.UnionType) -> None:
    """Emit a packed union declaration plus its type id string.
 
    Fields narrower than the union width get wrapper structs with a `_pad`
    byte array so the data sits at the MSB end, matching SV packed union
    layout where padding occupies the LSBs / lower addresses.
    """
    union_name = self.type_id_map[union_type]
    union_bytes = self._field_byte_width(union_type)
    fields = list(union_type.fields)
 
    # First pass: emit wrapper structs for fields that need padding.
    wrapper_names: Dict[str, str] = {}
    for field_name, field_type in fields:
      field_bytes = self._field_byte_width(field_type)
      pad_bytes = union_bytes - field_bytes
      if pad_bytes > 0:
        wrapper = f"{union_name}_{field_name}"
        wrapper_names[field_name] = wrapper
        hdr.write(f"struct {wrapper} {{\n")
        hdr.write(f"  uint8_t _pad[{pad_bytes}];\n")
        field_cpp = self._cpp_type(field_type)
        hdr.write(f"  {field_cpp} {field_name};\n")
        hdr.write("};\n")
 
    # Second pass: emit the union itself.
    union_field_decls: List[str] = []
    for field_name, field_type in fields:
      if field_name in wrapper_names:
        union_field_decls.append(f"{wrapper_names[field_name]} {field_name};")
      else:
        field_cpp = self._cpp_type(field_type)
        union_field_decls.append(f"{field_cpp} {field_name};")
    hdr.write(f"union {union_name} {{\n")
    for decl in union_field_decls:
      hdr.write(f"  {decl}\n")
    hdr.write("\n")
    hdr.write(
        f"  static constexpr std::string_view _ESI_ID = {self._cpp_string_literal(union_type.id)};\n"
    )
    hdr.write("};\n\n")
 
  def _emit_window(self, hdr: TextIO, window_type: types.WindowType) -> None:
    """Emit a SegmentedMessageData helper for a serial list window."""
    info = self._analyze_window(window_type)
    ctor_params = [
        self._format_window_ctor_param(name, field_type)
        for name, field_type in info["ctor_params"]
    ]
    value_ctor_params = list(ctor_params)
    value_ctor_params.append(
        f"const std::vector<value_type> &{info['list_field_name']}")
    value_ctor_signature = ", ".join(value_ctor_params)
    frame_ctor_params = list(ctor_params)
    frame_ctor_params.append("std::vector<data_frame> frames")
    frame_ctor_signature = ", ".join(frame_ctor_params)
    helper_args = ", ".join(name for name, _ in info["ctor_params"])
    helper_call = f"{helper_args}, std::move(frames)" if helper_args else "std::move(frames)"
 
    hdr.write(
        f"struct {info['window_name']} : public esi::SegmentedMessageData {{\n")
    hdr.write("public:\n")
    hdr.write(f"  using value_type = {info['element_cpp']};\n")
    hdr.write(f"  using count_type = {info['count_cpp']};\n\n")
    hdr.write("  struct data_frame {\n")
    if info["data_pad_bytes"] > 0:
      hdr.write(f"    uint8_t _pad[{info['data_pad_bytes']}];\n")
    for field_name, field_type in info["data_fields"]:
      decl = self._format_window_field_decl(field_name, field_type)
      hdr.write(f"    {decl}\n")
    hdr.write("  };\n\n")
    hdr.write("private:\n")
    hdr.write("  struct header_frame {\n")
    if info["header_pad_bytes"] > 0:
      hdr.write(f"    uint8_t _pad[{info['header_pad_bytes']}];\n")
    for field_name, field_type in info["header_fields"]:
      if field_type is None:
        if info["count_width"] % 8 == 0:
          decl = f"count_type {field_name};"
        else:
          decl = f"count_type {field_name} : {info['count_width']};"
      else:
        decl = self._format_window_field_decl(field_name, field_type)
      hdr.write(f"    {decl}\n")
    hdr.write("  };\n\n")
    hdr.write("  header_frame header{};\n")
    hdr.write("  std::vector<data_frame> data_frames;\n")
    hdr.write("  header_frame footer{};\n\n")
    hdr.write(f"  void construct({frame_ctor_signature}) {{\n")
    hdr.write("    if (frames.empty())\n")
    hdr.write(
        f"      throw std::invalid_argument(\"{info['window_name']}: bulk windowed lists cannot be empty\");\n"
    )
    hdr.write(
        "    if (frames.size() > std::numeric_limits<count_type>::max())\n")
    hdr.write(
        f"      throw std::invalid_argument(\"{info['window_name']}: list too large for encoded count\");\n"
    )
    hdr.write(
        f"    header.{info['count_field_name']} = static_cast<count_type>(frames.size());\n"
    )
    for name, _ in info["ctor_params"]:
      field_type = next(
          field_type for field_name, field_type in info["ctor_params"]
          if field_name == name)
      self._emit_window_field_copy(hdr, f"header.{name}", name, field_type)
    hdr.write(f"    footer.{info['count_field_name']} = 0;\n")
    hdr.write("    data_frames = std::move(frames);\n")
    hdr.write("  }\n\n")
    hdr.write("public:\n")
    hdr.write(f"  {info['window_name']}({frame_ctor_signature}) {{\n")
    hdr.write(f"    construct({helper_call});\n")
    hdr.write("  }\n\n")
    hdr.write(f"  {info['window_name']}({value_ctor_signature}) {{\n")
    hdr.write("    std::vector<data_frame> frames;\n")
    hdr.write(f"    frames.reserve({info['list_field_name']}.size());\n")
    hdr.write(f"    for (const auto &element : {info['list_field_name']}) {{\n")
    hdr.write("      auto &frame = frames.emplace_back();\n")
    list_field_type = next(
        field_type for field_name, field_type in info["data_fields"]
        if field_name == info["list_field_name"])
    if isinstance(list_field_type, types.ArrayType):
      hdr.write(
          f"      std::memcpy(&frame.{info['list_field_name']}, &element, sizeof(frame.{info['list_field_name']}));\n"
      )
    else:
      hdr.write(f"      frame.{info['list_field_name']} = element;\n")
    hdr.write("    }\n")
    hdr.write(f"    construct({helper_call});\n")
    hdr.write("  }\n\n")
    hdr.write("  size_t numSegments() const override { return 3; }\n")
    hdr.write("  esi::Segment segment(size_t idx) const override {\n")
    hdr.write("    if (idx == 0)\n")
    hdr.write(
        "      return {reinterpret_cast<const uint8_t *>(&header), sizeof(header)};\n"
    )
    hdr.write("    if (idx == 1)\n")
    hdr.write(
        "      return {reinterpret_cast<const uint8_t *>(data_frames.data()),\n"
    )
    hdr.write("              data_frames.size() * sizeof(data_frame)};\n")
    hdr.write("    if (idx == 2)\n")
    hdr.write(
        "      return {reinterpret_cast<const uint8_t *>(&footer), sizeof(footer)};\n"
    )
    hdr.write(
        f"    throw std::out_of_range(\"{info['window_name']}: invalid segment index\");\n"
    )
    hdr.write("  }\n\n")
    hdr.write(
        f"  static constexpr std::string_view _ESI_ID = {self._cpp_string_literal(window_type.id)};\n"
    )
    hdr.write("};\n\n")
 
  def _emit_alias(self, hdr: TextIO, alias_type: types.TypeAlias) -> None:
    """Emit a using alias when the alias targets a different C++ type."""
    inner_wrapped = alias_type.inner_type
    alias_name = self.type_id_map[alias_type]
    inner_cpp = None
    if inner_wrapped is not None:
      inner_cpp = self._cpp_type(inner_wrapped)
    if inner_cpp is None:
      inner_cpp = self.type_id_map[alias_type]
    if inner_cpp != alias_name:
      hdr.write(f"using {alias_name} = {inner_cpp};\n\n")
 
  def write_header(self, output_dir: Path, system_name: str) -> None:
    """Emit the fully ordered types.h header into the output directory."""
    hdr_file = output_dir / "types.h"
    with open(hdr_file, "w") as hdr:
      hdr.write(
          textwrap.dedent(f"""
        // Generated header for {system_name} types.
        #pragma once
 
        #include <cstdint>
        #include <cstddef>
        #include <any>
        #include <cstring>
        #include <limits>
        #include <stdexcept>
        #include <string_view>
        #include <utility>
        #include <vector>
 
        #include "esi/Common.h"
 
        namespace {system_name} {{
        #pragma pack(push, 1)
 
      """))
      if self.has_cycle:
        sys.stderr.write("Warning: cyclic type dependencies detected.\n")
        sys.stderr.write("  Logically this should not be possible.\n")
        sys.stderr.write(
            "  Emitted code may fail to compile due to ordering issues.\n")
 
      for emit_type in self.ordered_types:
        try:
          if isinstance(emit_type, types.StructType):
            self._emit_struct(hdr, emit_type)
          elif isinstance(emit_type, types.UnionType):
            self._emit_union(hdr, emit_type)
          elif isinstance(emit_type, types.WindowType):
            self._emit_window(hdr, emit_type)
          elif isinstance(emit_type, types.TypeAlias):
            self._emit_alias(hdr, emit_type)
        except ValueError as e:
          sys.stderr.write(f"Error emitting type '{emit_type}': {e}\n")
          hdr.write(f"// Unsupported type '{emit_type}': {e}\n\n")
 
      hdr.write(
          textwrap.dedent(f"""
    #pragma pack(pop)
    }} // namespace {system_name}
    """))
 
 
def run(generator: Type[Generator] = CppGenerator,
        cmdline_args=sys.argv) -> int:
  """Create and run a generator reading options from the command line."""
 
  argparser = argparse.ArgumentParser(
      description=f"Generate {generator.language} headers from an ESI manifest",
      formatter_class=argparse.RawDescriptionHelpFormatter,
      epilog=textwrap.dedent("""
        Can read the manifest from either a file OR a running accelerator.
 
        Usage examples:
          # To read the manifest from a file:
          esi-cppgen --file /path/to/manifest.json
 
          # To read the manifest from a running accelerator:
          esi-cppgen --platform cosim --connection localhost:1234
      """))
 
  argparser.add_argument("--file",
                         type=str,
                         default=None,
                         help="Path to the manifest file.")
  argparser.add_argument(
      "--platform",
      type=str,
      help="Name of platform for live accelerator connection.")
  argparser.add_argument(
      "--connection",
      type=str,
      help="Connection string for live accelerator connection.")
  argparser.add_argument(
      "--output-dir",
      type=str,
      default="esi",
      help="Output directory for generated files. Recommend adding either `esi`"
      " or the system name to the end of the path so as to avoid header name"
      "conflicts. Defaults to `esi`")
  argparser.add_argument(
      "--system-name",
      type=str,
      default="esi_system",
      help="Name of the ESI system. For C++, this will be the namespace.")
 
  if (len(cmdline_args) <= 1):
    argparser.print_help()
    return 1
  args = argparser.parse_args(cmdline_args[1:])
 
  if args.file is not None and args.platform is not None:
    print("Cannot specify both --file and --platform")
    return 1
 
  conn: AcceleratorConnection
  if args.file is not None:
    # Use os.pathsep (';' on Windows, ':' on Unix) to avoid conflicts with
    # drive letters.
    conn = Context.default().connect("trace", f"-{os.pathsep}{args.file}")
  elif args.platform is not None:
    if args.connection is None:
      print("Must specify --connection with --platform")
      return 1
    conn = Context.default().connect(args.platform, args.connection)
  else:
    print("Must specify either --file or --platform")
    return 1
 
  output_dir = Path(args.output_dir)
  if output_dir.exists() and not output_dir.is_dir():
    print(f"Output directory {output_dir} is not a directory")
    return 1
  if not output_dir.exists():
    output_dir.mkdir(parents=True)
 
  gen = generator(conn)
  gen.generate(output_dir, args.system_name)
  return 0
 
 
if __name__ == '__main__':
  sys.exit(run())