feat: add workload-tohex functionality with API and event steps for converting ELF files to hex

Author: shirohasuki

api/steps/kernel/01_build_event.step.py

@@ -20,7 +20,7 @@
 
 config = {
     "name": "kernel-build",
-    "description": "build RISC-V kernel + rootfs for Pegasus via bb-tests/workloads/lib/kernel",
+    "description": "build RISC-V kernel + rootfs for image via bb-tests/workloads/lib/kernel",
     "flows": ["kernel"],
     "triggers": [queue("kernel.build")],
     "enqueues": [],
@@ -66,8 +66,8 @@ async def handler(input_data: dict, ctx: FlowContext) -> None:
         return
 
     # Convert bin to hex for P2E memory backdoor
-    bin_file = os.path.join(output_dir, "pegasus-bin")
-    hex_file = os.path.join(output_dir, "pegasus.hex")
+    bin_file = os.path.join(output_dir, "image.bin")
+    hex_file = os.path.join(output_dir, "image.hex")
 
     if os.path.exists(bin_file):
         ctx.logger.info(f"Converting {bin_file} to Verilog hex format...")

api/steps/workload/02_tohex_api.step.py

@@ -0,0 +1,19 @@
+from motia import ApiRequest, ApiResponse, FlowContext, api
+
+from utils.path import get_buckyball_path
+
+config = {
+    "name": "workload-tohex-api",
+    "description": "convert elf to hex",
+    "flows": ["workload"],
+    "triggers": [api("POST", "/workload/tohex")],
+    "enqueues": ["workload.tohex"],
+}
+
+
+async def handler(request: ApiRequest, ctx: FlowContext) -> ApiResponse:
+    bbdir = get_buckyball_path()
+    body = request.body or {}
+    data = {}
+    await ctx.enqueue({"topic": "workload.tohex", "data": {**data, "_trace_id": ctx.trace_id}})
+    return ApiResponse(status=202, body={"trace_id": ctx.trace_id})

api/steps/workload/02_tohex_event.step.py

@@ -0,0 +1,89 @@
+import os
+import subprocess
+import sys
+from pathlib import Path
+
+from motia import FlowContext, queue
+
+# Add the utils directory to the Python path
+utils_path = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
+if utils_path not in sys.path:
+    sys.path.insert(0, utils_path)
+
+from utils.path import get_buckyball_path
+from utils.stream_run import stream_run_logger
+from utils.event_common import check_result, get_origin_trace_id
+
+config = {
+    "name": "workload-tohex",
+    "description": "convert all -baremetal ELF files under bb-tests/output/workloads/src to hex",
+    "flows": ["workload"],
+    "triggers": [queue("workload.tohex")],
+    "enqueues": [],
+}
+
+
+async def handler(input_data: dict, ctx: FlowContext) -> None:
+    origin_tid = get_origin_trace_id(input_data, ctx)
+    bbdir = get_buckyball_path()
+    script = f"{bbdir}/bbdev/api/steps/workload/scripts/elf2hex.py"
+    search_root = Path(f"{bbdir}/bb-tests/output/workloads/src")
+
+    ctx.logger.info("Search root", {"path": str(search_root)})
+
+    if not search_root.is_dir():
+        ctx.logger.error(
+            "Workload output directory not found",
+            {"path": str(search_root)},
+        )
+        await check_result(ctx, 1, continue_run=False, trace_id=origin_tid)
+        return
+
+    ctx.logger.info("Search root is a directory", {"path": str(search_root)})
+
+    elf_files = sorted(p for p in search_root.rglob("*-baremetal") if p.is_file())
+
+    ctx.logger.info("ELF files found", {"count": len(elf_files)})
+
+    if not elf_files:
+        ctx.logger.warn(
+            "No -baremetal ELF files found",
+            {"search_root": str(search_root)},
+        )
+        await check_result(ctx, 0, continue_run=False, trace_id=origin_tid)
+        return
+
+    ctx.logger.info(
+        "Converting ELF files to hex",
+        {"count": len(elf_files), "search_root": str(search_root)},
+    )
+
+    overall_rc = 0
+    for elf in elf_files:
+        command = f"python3 {script} {elf}"
+        ctx.logger.info("Executing to-hex command", {"command": command})
+        result = stream_run_logger(
+            cmd=command,
+            logger=ctx.logger,
+            cwd=str(elf.parent),
+            executable="bash",
+            stdout_prefix=f"to-hex {elf.name}",
+            stderr_prefix=f"to-hex {elf.name}",
+        )
+        if result.returncode != 0:
+            overall_rc = result.returncode
+            ctx.logger.error(
+                "elf2hex failed",
+                {"elf": str(elf), "returncode": result.returncode},
+            )
+
+    # ==================================================================================
+    # Return conversion result
+    # ==================================================================================
+    success_result, failure_result = await check_result(
+        ctx, overall_rc, continue_run=False, trace_id=origin_tid)
+
+    # ==================================================================================
+    #  finish workflow
+    # ==================================================================================
+    return

api/steps/workload/scripts/elf2hex.py

@@ -0,0 +1,78 @@
+#!/usr/bin/env python3
+"""ELF -> hex converter for Buckyball P2E DDR loading.
+
+Output format:
+    @0
+    XX
+    XX
+    ...
+where XX is one byte per line (uppercase hex).
+
+The hex file represents bytes starting at DDR offset 0, which maps to
+CPU address 0x80000000. All LOAD segments are concatenated by physical
+address, with gaps padded by zeros.
+"""
+
+import argparse
+import subprocess
+import sys
+from pathlib import Path
+
+
+def elf_to_bin(elf_path: Path, bin_path: Path, objcopy: str) -> None:
+    subprocess.run(
+        [objcopy, "-O", "binary", str(elf_path), str(bin_path)],
+        check=True,
+    )
+
+
+def bin_to_hex(bin_path: Path, hex_path: Path) -> int:
+    data = bin_path.read_bytes()
+    lines = ["@0"]
+    lines.extend(f"{b:02X}" for b in data)
+    hex_path.write_text("\n".join(lines) + "\n")
+    return len(data)
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    parser.add_argument("elf", type=Path, help="Input ELF file")
+    parser.add_argument(
+        "hex",
+        type=Path,
+        nargs="?",
+        help="Output hex file (default: <elf>.hex)",
+    )
+    parser.add_argument(
+        "--objcopy",
+        default="riscv64-unknown-elf-objcopy",
+        help="objcopy binary (default: riscv64-unknown-elf-objcopy)",
+    )
+    args = parser.parse_args()
+
+    if not args.elf.is_file():
+        print(f"error: ELF not found: {args.elf}", file=sys.stderr)
+        return 1
+
+    hex_path = args.hex or args.elf.with_suffix(".hex")
+    bin_path = hex_path.with_suffix(".bin")
+
+    try:
+        elf_to_bin(args.elf, bin_path, args.objcopy)
+    except FileNotFoundError:
+        print(f"error: objcopy not found: {args.objcopy}", file=sys.stderr)
+        print("hint: run inside `nix develop` shell", file=sys.stderr)
+        return 1
+    except subprocess.CalledProcessError as e:
+        print(f"error: objcopy failed: {e}", file=sys.stderr)
+        return 1
+
+    n = bin_to_hex(bin_path, hex_path)
+    bin_path.unlink()
+
+    print(f"wrote {hex_path} ({n} bytes, {n + 1} lines)")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

api/tests/test_api_workload_tohex.py

@@ -0,0 +1,3 @@
+from _api_test_helper import run_bbdev_case
+
+run_bbdev_case("bbdev workload --tohex")

bbdev

@@ -42,6 +42,7 @@ WORKFLOW_COMMANDS = {
     },
     "workload": {
         "build": "Build workload.",
+        "tohex": "Convert all -baremetal ELF files under bb-tests/output/workloads/src to hex.",
     },
     "kernel": {
         "build": "Build RISC-V kernel + rootfs for Pegasus (cmake kernel-build).",