secs-gem/tools/gen_messages.py

#!/usr/bin/env python3
"""Generate include/secsgem/gem/messages.hpp from data/messages.yaml.

Reads a SECS-II message catalog and emits an inline-only C++ header with
builders + parsers for each message. The catalog is the single source of
truth for SxFy body shapes.

Shape grammar (used recursively for `body` and `element` and field types):

  none                       -> message has no body (header-only)

  {kind: scalar,
   item_type: <T>,
   enum: <E>?,               # optional C++ enum type
   param: <name>?,           # builder parameter name (default 'value')
   parser_struct_field: ...} # only valid inside a list's `fields`

  {kind: list,
   fields: [{name, shape}, ...],
   struct_name: <Name>?,     # if set, parser returns std::optional<Name>
   external_struct: bool?}   # if true, the struct is assumed to already exist

  {kind: list_of,
   element: <shape>,
   name: <param-name>?}      # builder param + parser value name

`item_type` is one of:
    ASCII, BINARY, BINARY_BYTE, BOOLEAN, ITEM,
    U1, U2, U4, U8, I1, I2, I4, I8, F4, F8

The generated file is overwrite-safe and entirely inline (no .cpp).
"""

import argparse
import sys
from pathlib import Path

import yaml


# --- Type table -----------------------------------------------------------

# Maps item_type -> (cpp_param_type, item_factory, parse_helper)
#   parse_helper(item_expr) -> C++ expression evaluating to std::optional<T>
SCALAR_TABLE = {
    "ASCII":        ("std::string", "secs2::Item::ascii({v})",
                     "as_ascii({i})"),
    "BINARY_BYTE":  ("uint8_t",     "secs2::Item::binary({{ {v} }})",
                     "as_binary_first({i})"),
    "BINARY":       ("std::string", "secs2::Item::binary(std::vector<uint8_t>({v}.begin(), {v}.end()))",
                     "as_binary_string({i})"),
    "BOOLEAN":      ("bool",        "secs2::Item::boolean({v})",
                     "as_boolean({i})"),
    "ITEM":         ("secs2::Item", "{v}",
                     "std::optional<secs2::Item>({i})"),
    "U1":           ("uint8_t",     "secs2::Item::u1({v})",
                     "as_u1_scalar({i})"),
    "U2":           ("uint16_t",    "secs2::Item::u2({v})",
                     "as_u2_scalar({i})"),
    "U4":           ("uint32_t",    "secs2::Item::u4({v})",
                     "as_u4_scalar({i})"),
    "U8":           ("uint64_t",    "secs2::Item::u8({v})",
                     "as_u8_scalar({i})"),
    "I1":           ("int8_t",      "secs2::Item::i1({v})",
                     "as_i1_scalar({i})"),
    "I2":           ("int16_t",     "secs2::Item::i2({v})",
                     "as_i2_scalar({i})"),
    "I4":           ("int32_t",     "secs2::Item::i4({v})",
                     "as_i4_scalar({i})"),
    "I8":           ("int64_t",     "secs2::Item::i8({v})",
                     "as_i8_scalar({i})"),
    "F4":           ("float",       "secs2::Item::f4({v})",
                     "as_f4_scalar({i})"),
    "F8":           ("double",      "secs2::Item::f8({v})",
                     "as_f8_scalar({i})"),
}


# --- Shape inspection -----------------------------------------------------

def is_scalar(shape):    return shape and shape.get("kind") == "scalar"
def is_list(shape):      return shape and shape.get("kind") == "list"
def is_list_of(shape):   return shape and shape.get("kind") == "list_of"


def cpp_type(shape):
    """C++ type used for a parameter / parser-return / struct field."""
    if is_scalar(shape):
        base, _, _ = SCALAR_TABLE[shape["item_type"]]
        return shape.get("enum", base)
    if is_list(shape):
        name = shape.get("struct_name")
        if not name:
            raise ValueError(f"list shape needs `struct_name`: {shape}")
        return name
    if is_list_of(shape):
        return f"std::vector<{cpp_type(shape['element'])}>"
    raise ValueError(f"unknown shape: {shape}")


def field_struct_name(field, message_name):
    return field.get("struct_name") or message_name


# --- Code emission --------------------------------------------------------

class Emit:
    def __init__(self):
        self.lines = []
        self.indent = 0
        self.emitted_structs = set()

    def line(self, s=""):
        self.lines.append("  " * self.indent + s if s else "")

    def block(self, header, body_fn, footer="}"):
        self.line(header)
        self.indent += 1
        body_fn()
        self.indent -= 1
        self.line(footer)


def collect_structs(shape, out):
    """Post-order traversal of the shape tree, appending any (name, fields) for
    list-with-struct_name nodes that aren't marked external. Post-order ensures
    deeper structs are emitted before the structs that reference them."""
    if not shape:
        return
    if is_list(shape):
        for f in shape["fields"]:
            collect_structs(f["shape"], out)
        name = shape.get("struct_name")
        if name and not shape.get("external_struct"):
            out.append((name, shape["fields"]))
    elif is_list_of(shape):
        collect_structs(shape["element"], out)


def emit_struct(e, name, fields):
    if name in e.emitted_structs:
        return
    e.emitted_structs.add(name)
    e.line(f"struct {name} {{")
    e.indent += 1
    for f in fields:
        e.line(f"{cpp_type(f['shape'])} {f['name']};")
    e.indent -= 1
    e.line("};")
    e.line()


def emit_build_expr(shape, var, depth=0):
    """Return a C++ expression that constructs an Item from `var`."""
    if is_scalar(shape):
        _, factory, _ = SCALAR_TABLE[shape["item_type"]]
        if "enum" in shape:
            # Cast the typed enum to its underlying byte before constructing.
            base, _, _ = SCALAR_TABLE[shape["item_type"]]
            return factory.format(v=f"static_cast<{base}>({var})")
        return factory.format(v=var)
    if is_list(shape):
        parts = [emit_build_expr(f["shape"], f"{var}.{f['name']}", depth)
                 for f in shape["fields"]]
        return "secs2::Item::list({" + ", ".join(parts) + "})"
    if is_list_of(shape):
        c = f"__c_{depth}"
        ev = f"__e_{depth}"
        elem_expr = emit_build_expr(shape["element"], ev, depth + 1)
        return (f"[&]{{ secs2::Item::List {c}; "
                f"for (const auto& {ev} : {var}) {c}.push_back({elem_expr}); "
                f"return secs2::Item::list(std::move({c})); }}()")
    raise ValueError(f"unknown shape: {shape}")


def emit_parse_expr(shape, item_var, depth=0):
    """Return a C++ expression of type std::optional<T> that parses item_var."""
    if is_scalar(shape):
        _, _, helper = SCALAR_TABLE[shape["item_type"]]
        expr = helper.format(i=item_var)
        if "enum" in shape:
            enum_t = shape["enum"]
            t = f"__t_{depth}"
            return (f"[&]() -> std::optional<{enum_t}> {{ auto {t} = {expr}; "
                    f"if (!{t}) return std::nullopt; "
                    f"return static_cast<{enum_t}>(*{t}); }}()")
        return expr
    if is_list(shape):
        sname = shape["struct_name"]
        n = len(shape["fields"])
        L = f"__l_{depth}"
        R = f"__r_{depth}"
        V = f"__v_{depth}"
        parts = [
            f"if (!{item_var}.is_list() || {item_var}.as_list().size() != {n}) return std::nullopt;",
            f"const auto& {L} = {item_var}.as_list();",
            f"{sname} {R};",
        ]
        for i, f in enumerate(shape["fields"]):
            sub = emit_parse_expr(f["shape"], f"{L}[{i}]", depth + 1)
            parts.append(f"{{ auto {V} = {sub}; if (!{V}) return std::nullopt; {R}.{f['name']} = *{V}; }}")
        parts.append(f"return std::optional<{sname}>({R});")
        return "[&]() -> std::optional<" + sname + "> { " + " ".join(parts) + " }()"
    if is_list_of(shape):
        elem_t = cpp_type(shape["element"])
        c = f"__c_{depth}"
        v = f"__v_{depth}"
        out = f"__out_{depth}"
        sub = emit_parse_expr(shape["element"], c, depth + 1)
        return ("[&]() -> std::optional<std::vector<" + elem_t + ">> { "
                f"if (!{item_var}.is_list()) return std::nullopt; "
                f"std::vector<{elem_t}> {out}; "
                f"for (const auto& {c} : {item_var}.as_list()) {{ "
                f"  auto {v} = {sub}; if (!{v}) return std::nullopt; "
                f"  {out}.push_back(*{v}); }} "
                f"return {out}; }}()")
    raise ValueError(f"unknown shape: {shape}")


def emit_message(e, m):
    stream, fn = m["stream"], m["function"]
    w = "true" if m.get("w") else "false"
    builder = m["builder"]
    body = m.get("body")

    # Emit any structs referenced by the body.
    structs = []
    if body:
        collect_structs(body, structs)
    for name, fields in structs:
        emit_struct(e, name, fields)

    # ---- Builder -------------------------------------------------------
    if body is None:
        e.line(f"inline secs2::Message {builder}() {{")
        e.indent += 1
        e.line(f"return secs2::Message({stream}, {fn}, {w});")
        e.indent -= 1
        e.line("}")
        e.line()
    elif is_scalar(body):
        ptype = cpp_type(body)
        pname = body.get("param", "value")
        build = emit_build_expr(body, pname)
        e.line(f"inline secs2::Message {builder}({ptype} {pname}) {{")
        e.indent += 1
        e.line(f"return secs2::Message({stream}, {fn}, {w}, {build});")
        e.indent -= 1
        e.line("}")
        e.line()
    elif is_list(body):
        fields = body["fields"]
        sname = body.get("struct_name")
        # Always emit a positional-parameter builder (no struct needed).
        param_decl = ", ".join(
            f"const {cpp_type(f['shape'])}& {f['name']}" for f in fields)
        parts = [emit_build_expr(f["shape"], f["name"]) for f in fields]
        body_expr = "secs2::Item::list({" + ", ".join(parts) + "})"
        e.line(f"inline secs2::Message {builder}({param_decl}) {{")
        e.indent += 1
        e.line(f"return secs2::Message({stream}, {fn}, {w}, {body_expr});")
        e.indent -= 1
        e.line("}")
        # If a struct exists, also emit an overload that takes it.
        if sname:
            forward = ", ".join(f"v.{f['name']}" for f in fields)
            e.line(f"inline secs2::Message {builder}(const {sname}& v) {{")
            e.indent += 1
            e.line(f"return {builder}({forward});")
            e.indent -= 1
            e.line("}")
        e.line()
    elif is_list_of(body):
        elem_t = cpp_type(body["element"])
        pname = body.get("name", "values")
        e.line(f"inline secs2::Message {builder}(const std::vector<{elem_t}>& {pname}) {{")
        e.indent += 1
        e.line(f"return secs2::Message({stream}, {fn}, {w}, {emit_build_expr(body, pname)});")
        e.indent -= 1
        e.line("}")
        e.line()
    else:
        raise ValueError(f"unknown body kind for {m['id']}: {body}")

    # ---- Parser --------------------------------------------------------
    parser = m.get("parser")
    if not parser:
        return
    if body is None:
        # no body to parse
        return

    if is_scalar(body):
        ret_t = cpp_type(body)
        e.line(f"inline std::optional<{ret_t}> {parser}(const secs2::Message& m) {{")
        e.indent += 1
        e.line("if (!m.body) return std::nullopt;")
        e.line(f"return {emit_parse_expr(body, '(*m.body)')};")
        e.indent -= 1
        e.line("}")
        e.line()
    elif is_list(body):
        sname = body["struct_name"]
        e.line(f"inline std::optional<{sname}> {parser}(const secs2::Message& m) {{")
        e.indent += 1
        e.line("if (!m.body) return std::nullopt;")
        e.line(f"return {emit_parse_expr(body, '(*m.body)')};")
        e.indent -= 1
        e.line("}")
        e.line()
    elif is_list_of(body):
        elem_t = cpp_type(body["element"])
        e.line(f"inline std::optional<std::vector<{elem_t}>> {parser}(const secs2::Message& m) {{")
        e.indent += 1
        e.line("if (!m.body) return std::nullopt;")
        e.line(f"return {emit_parse_expr(body, '(*m.body)')};")
        e.indent -= 1
        e.line("}")
        e.line()


# --- Top level ------------------------------------------------------------

PREAMBLE = """\
// GENERATED by tools/gen_messages.py from data/messages.yaml.
// DO NOT EDIT — re-run the generator instead.
//
// Provides inline builders + parsers for every SECS-II message in the
// catalog. Hand-written helpers (scalar accessors, list-of-U4 helper,
// special-case messages) live in messages_helpers.hpp.

#pragma once

#include <cstdint>
#include <optional>
#include <string>
#include <utility>
#include <vector>

#include "secsgem/gem/control_state.hpp"
#include "secsgem/gem/data_model.hpp"
#include "secsgem/gem/messages_helpers.hpp"
#include "secsgem/secs2/item.hpp"
#include "secsgem/secs2/message.hpp"

namespace secsgem::gem {

namespace secs2 = secsgem::secs2;

"""

POSTAMBLE = """\
}  // namespace secsgem::gem
"""


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--catalog", required=True, help="path to data/messages.yaml")
    ap.add_argument("--out", required=True, help="output header path")
    args = ap.parse_args()

    catalog = yaml.safe_load(Path(args.catalog).read_text())
    messages = catalog["messages"]

    e = Emit()
    e.lines.append(PREAMBLE.rstrip())
    e.line()

    for m in messages:
        e.line(f"// ---- {m['id']} : {m['builder']} ----")
        emit_message(e, m)

    e.line(POSTAMBLE.rstrip())

    Path(args.out).parent.mkdir(parents=True, exist_ok=True)
    Path(args.out).write_text("\n".join(e.lines) + "\n")
    print(f"generated {len(messages)} messages -> {args.out}", file=sys.stderr)


if __name__ == "__main__":
    main()