Merge branch 'fix/capabilities-board-skill-bmgr' into 'master'

refactor(capabilities): load board skill from bmgr code

See merge request ae_group/esp-clawgent!37

application/basic_demo/fatfs_image/scripts/random_demo_stress.lua

@@ -1,185 +0,0 @@
-local delay = require("delay")
-local esp_heap = require("esp_heap")
-local storage = require("storage")
-
-local LUA_ROOT = storage.join_path(storage.get_root_dir(), "scripts")
-local caps = esp_heap.caps
-
-local iterations = (args and args.iterations) or 20
-local pause_ms = (args and args.pause_ms) or 200
-local gc_each_round = (args and args.gc_each_round) ~= false
-local include_interactive = (args and args.include_interactive) == true
-local summary = {
-    ok = 0,
-    failed = 0,
-    failures = {},
-}
-
-local demos = {
-    "hello.lua",
-    "coroutine_demo.lua",
-    "gc_demo.lua",
-    "led_strip_demo.lua",
-    "display_demo.lua",
-    "audio_play_test_wav.lua",
-    "audio_demo.lua",
-    "camera_capture_demo.lua",
-}
-
-if include_interactive then
-    demos[#demos + 1] = "button_demo.lua"
-    demos[#demos + 1] = "button_play_test_wav.lua"
-    demos[#demos + 1] = "lcd_touch_demo.lua"
-    demos[#demos + 1] = "lcd_touch_paint.lua"
-end
-
-local function info(c)
-    return esp_heap.get_info(c)
-end
-
-local function print_heap_snapshot(tag)
-    local default_info = info(caps.DEFAULT)
-    local internal_info = info(caps.INTERNAL)
-    local spiram_info = info(caps.SPIRAM)
-
-    print(string.format(
-        "[random_demo_stress] heap %s default free=%d min=%d largest=%d | internal free=%d min=%d | spiram free=%d min=%d",
-        tag,
-        default_info.free_size,
-        default_info.minimum_free_size,
-        default_info.largest_free_block,
-        internal_info.free_size,
-        internal_info.minimum_free_size,
-        spiram_info.free_size,
-        spiram_info.minimum_free_size
-    ))
-end
-
-local function print_task_watermarks(tag)
-    local tasks = esp_heap.get_task_watermarks()
-
-    if tasks._warning then
-        print("[random_demo_stress] task watermark note: " .. tostring(tasks._warning))
-    end
-
-    table.sort(tasks, function(a, b)
-        if a.stack_high_water_mark_bytes == b.stack_high_water_mark_bytes then
-            return a.name < b.name
-        end
-        return a.stack_high_water_mark_bytes < b.stack_high_water_mark_bytes
-    end)
-
-    print("[random_demo_stress] task watermarks " .. tag)
-    for i = 1, math.min(#tasks, 6) do
-        local task = tasks[i]
-        print(string.format(
-            "[random_demo_stress]   #%d %s state=%s stack_hwm=%dB prio=%d",
-            i,
-            tostring(task.name),
-            tostring(task.state),
-            tonumber(task.stack_high_water_mark_bytes) or -1,
-            tonumber(task.current_priority) or -1
-        ))
-    end
-end
-
-local function run_demo(name)
-    local path = LUA_ROOT .. "/" .. name
-    local original_print = print
-    local saw_error_output = false
-    local result_ok = false
-    local result_reason = nil
-
-    local function wrapped_print(...)
-        local parts = {}
-        for i = 1, select("#", ...) do
-            parts[i] = tostring(select(i, ...))
-        end
-        local line = table.concat(parts, "\t")
-        if string.find(line, "ERROR:", 1, true) then
-            saw_error_output = true
-        end
-        original_print(...)
-    end
-
-    print(string.format("[random_demo_stress] running %s", name))
-    print = wrapped_print
-    local ok, err = xpcall(function()
-        dofile(path)
-    end, debug.traceback)
-    print = original_print
-
-    if ok and not saw_error_output then
-        print(string.format("[random_demo_stress] result %s ok", name))
-        result_ok = true
-    elseif ok then
-        print(string.format("[random_demo_stress] result %s failed: demo printed ERROR output", name))
-        result_reason = "demo printed ERROR output"
-    else
-        print(string.format("[random_demo_stress] result %s failed: %s", name, tostring(err)))
-        result_reason = tostring(err)
-    end
-
-    return result_ok, result_reason
-end
-
-math.randomseed((os.time() % 2147483647) + math.floor(collectgarbage("count")))
-
-print(string.format(
-    "[random_demo_stress] start iterations=%d pause_ms=%d gc_each_round=%s include_interactive=%s",
-    iterations,
-    pause_ms,
-    tostring(gc_each_round),
-    tostring(include_interactive)
-))
-
-print_heap_snapshot("initial")
-print_task_watermarks("initial")
-
-for i = 1, iterations do
-    local index = math.random(1, #demos)
-    local name = demos[index]
-
-    print(string.format("[random_demo_stress] round %d/%d pick=%s", i, iterations, name))
-    print_heap_snapshot("before")
-    local ok, reason = run_demo(name)
-    if ok then
-        summary.ok = summary.ok + 1
-    else
-        summary.failed = summary.failed + 1
-        summary.failures[#summary.failures + 1] = {
-            round = i,
-            name = name,
-            reason = reason or "unknown failure",
-        }
-    end
-
-    if gc_each_round then
-        collectgarbage("collect")
-    end
-
-    print_heap_snapshot("after")
-    print_task_watermarks("after round " .. tostring(i))
-    if pause_ms > 0 then
-        delay.delay_ms(pause_ms)
-    end
-end
-
-print_heap_snapshot("final")
-print_task_watermarks("final")
-print(string.format(
-    "[random_demo_stress] summary total=%d ok=%d failed=%d",
-    iterations,
-    summary.ok,
-    summary.failed
-))
-for i = 1, #summary.failures do
-    local item = summary.failures[i]
-    print(string.format(
-        "[random_demo_stress] summary failure round=%d demo=%s reason=%s",
-        item.round,
-        item.name,
-        tostring(item.reason)
-    ))
-end
-print("[random_demo_stress] done")

components/claw_capabilities/cap_boards/tools/generate_board_skill.py

@@ -12,8 +12,6 @@ import sys
 from pathlib import Path
 from typing import Any
 
-import yaml
-
 PIN_KEY_SUFFIXES = (
     '_io_num',
     '_gpio_num',
@@ -23,6 +21,9 @@ PIN_KEY_NAMES = {
     'gpio_num',
     'sda',
     'scl',
+    'mclk',
+    'bclk',
+    'ws',
     'mosi',
     'miso',
     'sclk',
@@ -52,19 +53,6 @@ def fail(message: str) -> None:
     raise RuntimeError(message)
 
 
-def load_yaml(path: Path) -> dict[str, Any]:
-    try:
-        with path.open('r', encoding='utf-8') as file:
-            data = yaml.safe_load(file)
-    except FileNotFoundError as exc:
-        raise RuntimeError(f'Missing YAML file: {path}') from exc
-    except yaml.YAMLError as exc:
-        raise RuntimeError(f'Invalid YAML file {path}: {exc}') from exc
-    if not isinstance(data, dict):
-        fail(f'Expected a YAML mapping in {path}')
-    return data
-
-
 def parse_board_path(gen_bmgr_dir: Path) -> Path:
     cmake_path = gen_bmgr_dir / 'CMakeLists.txt'
     text = cmake_path.read_text(encoding='utf-8')
@@ -74,18 +62,293 @@ def parse_board_path(gen_bmgr_dir: Path) -> Path:
     return Path(match.group(1)).resolve()
 
 
+def strip_c_comments(text: str) -> str:
+    text = re.sub(r'/\*.*?\*/', '', text, flags=re.DOTALL)
+    return re.sub(r'//.*', '', text)
+
+
+class CInitializerParser:
+    def __init__(self, text: str) -> None:
+        self.text = text
+        self.length = len(text)
+        self.index = 0
+
+    def parse(self) -> Any:
+        self._skip_ws()
+        value = self._parse_value()
+        self._skip_ws()
+        return value
+
+    def _skip_ws(self) -> None:
+        while self.index < self.length and self.text[self.index].isspace():
+            self.index += 1
+
+    def _peek(self) -> str | None:
+        if self.index >= self.length:
+            return None
+        return self.text[self.index]
+
+    def _consume(self, expected: str | None = None) -> str:
+        char = self._peek()
+        if char is None:
+            fail('Unexpected end of C initializer')
+        if expected is not None and char != expected:
+            fail(f'Expected {expected!r} in C initializer, got {char!r}')
+        self.index += 1
+        return char
+
+    def _parse_value(self) -> Any:
+        self._skip_ws()
+        char = self._peek()
+        if char == '{':
+            return self._parse_initializer()
+        if char == '"':
+            return self._parse_string()
+        if char == '-':
+            self._consume('-')
+            self._skip_ws()
+            number = self._parse_number()
+            if isinstance(number, int):
+                return -number
+            return f'-{number}'
+        if char is None:
+            fail('Missing C initializer value')
+        if char.isdigit():
+            return self._parse_number()
+        return self._parse_expression()
+
+    def _parse_initializer(self) -> Any:
+        self._consume('{')
+        self._skip_ws()
+        if self._peek() == '}':
+            self._consume('}')
+            return []
+
+        items: list[Any] = []
+        mapping: dict[str, Any] = {}
+        has_designators = False
+
+        while True:
+            self._skip_ws()
+            char = self._peek()
+            if char == '.':
+                has_designators = True
+                key = self._parse_designator()
+                self._skip_ws()
+                self._consume('=')
+                mapping[key] = self._parse_value()
+            else:
+                items.append(self._parse_value())
+
+            self._skip_ws()
+            char = self._peek()
+            if char == ',':
+                self._consume(',')
+                self._skip_ws()
+                if self._peek() == '}':
+                    self._consume('}')
+                    break
+                continue
+            if char == '}':
+                self._consume('}')
+                break
+            fail(f'Unexpected token {char!r} inside C initializer')
+
+        return mapping if has_designators else items
+
+    def _parse_designator(self) -> str:
+        self._consume('.')
+        start = self.index
+        while self.index < self.length and (self.text[self.index].isalnum() or self.text[self.index] == '_'):
+            self.index += 1
+        if start == self.index:
+            fail('Expected designator name in C initializer')
+        return self.text[start:self.index]
+
+    def _parse_string(self) -> str:
+        self._consume('"')
+        result: list[str] = []
+        while True:
+            char = self._consume()
+            if char == '"':
+                return ''.join(result)
+            if char == '\\':
+                escaped = self._consume()
+                translations = {'n': '\n', 'r': '\r', 't': '\t', '\\': '\\', '"': '"'}
+                result.append(translations.get(escaped, escaped))
+                continue
+            result.append(char)
+
+    def _parse_number(self) -> int | float | str:
+        start = self.index
+        while self.index < self.length and (self.text[self.index].isalnum() or self.text[self.index] in {'.', 'x', 'X'}):
+            self.index += 1
+        token = self.text[start:self.index]
+        if any(marker in token for marker in {'.', 'e', 'E'}):
+            try:
+                return float(token)
+            except ValueError:
+                return token
+        try:
+            return int(token, 0)
+        except ValueError:
+            return token
+
+    def _parse_expression(self) -> Any:
+        start = self.index
+        paren_depth = 0
+        while self.index < self.length:
+            char = self.text[self.index]
+            if char == '(':
+                paren_depth += 1
+            elif char == ')':
+                if paren_depth == 0:
+                    break
+                paren_depth -= 1
+            elif paren_depth == 0 and char in {',', '}'}:
+                break
+            self.index += 1
+        token = self.text[start:self.index].strip()
+        if token == 'true':
+            return True
+        if token == 'false':
+            return False
+        if token == 'NULL':
+            return None
+        return token
+
+
+def find_initializer_block(text: str, variable_name: str) -> str:
+    pattern = re.compile(rf'^\s*.*?\b{re.escape(variable_name)}\b\s*(?:\[[^\]]*\])?\s*=\s*\{{', re.MULTILINE)
+    match = pattern.search(text)
+    if not match:
+        fail(f'Could not find initializer for {variable_name}')
+
+    brace_start = text.find('{', match.start())
+    depth = 0
+    index = brace_start
+    while index < len(text):
+        char = text[index]
+        if char == '{':
+            depth += 1
+        elif char == '}':
+            depth -= 1
+            if depth == 0:
+                return text[brace_start:index + 1]
+        index += 1
+    fail(f'Unterminated initializer for {variable_name}')
+
+
+def parse_c_initializer_file(path: Path, variable_name: str) -> Any:
+    try:
+        text = path.read_text(encoding='utf-8')
+    except FileNotFoundError as exc:
+        raise RuntimeError(f'Missing generated source file: {path}') from exc
+    initializer = find_initializer_block(strip_c_comments(text), variable_name)
+    return CInitializerParser(initializer).parse()
+
+
+def load_board_info(gen_bmgr_dir: Path) -> dict[str, Any]:
+    data = parse_c_initializer_file(gen_bmgr_dir / 'gen_board_info.c', 'g_esp_board_info')
+    if not isinstance(data, dict):
+        fail('Expected g_esp_board_info to be a designated initializer')
+    return {
+        'board': data.get('name', 'unknown'),
+        'chip': data.get('chip', 'unknown'),
+        'version': data.get('version', 'unknown'),
+        'description': data.get('description', ''),
+        'manufacturer': data.get('manufacturer', 'unknown'),
+    }
+
+
+def load_peripheral_map(gen_bmgr_dir: Path) -> dict[str, dict[str, Any]]:
+    source = gen_bmgr_dir / 'gen_board_periph_config.c'
+    descriptors = parse_c_initializer_file(source, 'g_esp_board_peripherals')
+    if not isinstance(descriptors, list):
+        fail('Expected g_esp_board_peripherals to be an array initializer')
+
+    peripheral_map: dict[str, dict[str, Any]] = {}
+    for descriptor in descriptors:
+        if not isinstance(descriptor, dict):
+            continue
+        name = descriptor.get('name')
+        cfg_ref = descriptor.get('cfg')
+        if not isinstance(name, str) or not name or not isinstance(cfg_ref, str) or not cfg_ref.startswith('&'):
+            continue
+        cfg_var_name = cfg_ref[1:]
+        cfg = parse_c_initializer_file(source, cfg_var_name)
+        peripheral_map[name] = summarize_peripheral({'name': name, 'config': cfg})
+    return peripheral_map
+
+
+def collect_peripheral_refs(node: Any, parents: list[str] | None = None) -> list[str]:
+    parents = parents or []
+    refs: list[str] = []
+
+    if isinstance(node, dict):
+        for key, value in node.items():
+            if isinstance(value, str):
+                if key in {'gpio_name', 'ledc_name', 'i2c_name', 'spi_name', 'periph_name', 'peripheral_name'}:
+                    refs.append(value)
+                elif key == 'name' and parents and parents[-1] in {'pa_cfg', 'i2c_cfg', 'i2s_cfg'}:
+                    refs.append(value)
+            refs.extend(collect_peripheral_refs(value, parents + [str(key)]))
+    elif isinstance(node, list):
+        for item in node:
+            refs.extend(collect_peripheral_refs(item, parents))
+
+    return refs
+
+
+def load_devices(gen_bmgr_dir: Path, peripheral_map: dict[str, dict[str, Any]]) -> list[dict[str, Any]]:
+    source = gen_bmgr_dir / 'gen_board_device_config.c'
+    descriptors = parse_c_initializer_file(source, 'g_esp_board_devices')
+    if not isinstance(descriptors, list):
+        fail('Expected g_esp_board_devices to be an array initializer')
+
+    devices: list[dict[str, Any]] = []
+    for descriptor in descriptors:
+        if not isinstance(descriptor, dict):
+            continue
+        name = descriptor.get('name')
+        cfg_ref = descriptor.get('cfg')
+        if not isinstance(name, str) or not name or name.startswith('fake'):
+            continue
+        if not isinstance(cfg_ref, str) or not cfg_ref.startswith('&'):
+            continue
+
+        config = parse_c_initializer_file(source, cfg_ref[1:])
+        device = summarize_device({'name': name, 'config': config}, peripheral_map)
+
+        power_ctrl_device = descriptor.get('power_ctrl_device')
+        if isinstance(power_ctrl_device, str) and power_ctrl_device:
+            related_device = next((item for item in devices if item['name'] == power_ctrl_device), None)
+            if related_device:
+                device['peripherals'].append({'name': power_ctrl_device, 'io_lines': related_device['io_lines']})
+
+        devices.append(device)
+    return devices
+
+
 def normalize_label(path_parts: list[str]) -> str:
     filtered = [part for part in path_parts if part not in {'config', 'pins', 'flags', 'sub_cfg', 'lcd_panel_config', 'io_spi_config', 'spi_bus_config'}]
     if not filtered:
         filtered = path_parts
-    label = '.'.join(filtered)
-    label = label.replace('_io_num', '').replace('_gpio_num', '').replace('_io', '')
+    normalized_parts = []
+    for part in filtered:
+        normalized = part.replace('_io_num', '').replace('_gpio_num', '').replace('_io', '')
+        if normalized == 'gpio_num':
+            normalized = 'gpio'
+        normalized_parts.append(normalized)
+    label = '.'.join(part for part in normalized_parts if part)
     if label == 'doubt':
         return 'dout'
     return label
 
 
 def is_pin_key(key: str, parents: list[str]) -> bool:
+    if parents and parents[-1] == 'levels':
+        return False
     return key in PIN_KEY_NAMES or key.endswith(PIN_KEY_SUFFIXES) or (parents and parents[-1] == 'pins')
 
 
@@ -96,6 +359,11 @@ def collect_io_entries(node: Any, path_parts: list[str] | None = None) -> list[t
     if isinstance(node, dict):
         for key, value in node.items():
             current_path = path_parts + [str(key)]
+            if key == 'pin_bit_mask' and isinstance(value, str):
+                match = re.fullmatch(r'BIT(?:64)?\((\d+)\)', value)
+                if match:
+                    entries.append(('gpio', int(match.group(1))))
+                continue
             if isinstance(value, int) and value >= 0 and is_pin_key(str(key), path_parts):
                 entries.append((normalize_label(current_path), value))
                 continue
@@ -137,22 +405,15 @@ def summarize_peripheral(peripheral: dict[str, Any]) -> dict[str, Any]:
 def summarize_device(device: dict[str, Any], peripheral_map: dict[str, dict[str, Any]]) -> dict[str, Any]:
     name = str(device.get('name', ''))
     config = device.get('config')
-    peripheral_refs = device.get('peripherals')
     io_entries = collect_io_entries(config) if isinstance(config, dict) else []
     peripherals: list[dict[str, Any]] = []
 
-    if isinstance(peripheral_refs, list):
-        for item in peripheral_refs:
-            if not isinstance(item, dict):
-                continue
-            peripheral_name = item.get('name')
-            if not isinstance(peripheral_name, str) or not peripheral_name:
-                continue
-            periph_summary = peripheral_map.get(peripheral_name)
-            peripherals.append({
-                'name': peripheral_name,
-                'io_lines': periph_summary['io_lines'] if periph_summary else [],
-            })
+    for peripheral_name in dict.fromkeys(collect_peripheral_refs(config) if isinstance(config, dict) else []):
+        periph_summary = peripheral_map.get(peripheral_name)
+        peripherals.append({
+            'name': peripheral_name,
+            'io_lines': periph_summary['io_lines'] if periph_summary else [],
+        })
 
     return {
         'name': name,
@@ -220,32 +481,11 @@ def main() -> int:
     output_md = Path(args.output_md).resolve()
 
     board_dir = parse_board_path(gen_bmgr_dir)
-    print(f'[cap_boards] Loading board metadata from {board_dir}')
-    board_info = load_yaml(board_dir / 'board_info.yaml')
-    board_devices = load_yaml(board_dir / 'board_devices.yaml')
-    board_peripherals = load_yaml(board_dir / 'board_peripherals.yaml')
-
-    devices_raw = board_devices.get('devices')
-    peripherals_raw = board_peripherals.get('peripherals')
-    if not isinstance(devices_raw, list):
-        fail(f"Expected 'devices' array in {board_dir / 'board_devices.yaml'}")
-    if not isinstance(peripherals_raw, list):
-        fail(f"Expected 'peripherals' array in {board_dir / 'board_peripherals.yaml'}")
-
-    peripheral_map = {}
-    for peripheral in peripherals_raw:
-        if not isinstance(peripheral, dict):
-            continue
-        name = peripheral.get('name')
-        if isinstance(name, str) and name:
-            peripheral_map[name] = summarize_peripheral(peripheral)
-
-    devices = [
-        summarize_device(device, peripheral_map)
-        for device in devices_raw
-        if isinstance(device, dict) and isinstance(device.get('name'), str) and not str(device.get('name')).startswith('fake')
-    ]
-    board_version = str(board_info.get('version') or board_devices.get('version') or board_peripherals.get('version') or 'unknown')
+    print(f'[cap_boards] Loading generated board metadata from {gen_bmgr_dir} (board path: {board_dir})')
+    board_info = load_board_info(gen_bmgr_dir)
+    peripheral_map = load_peripheral_map(gen_bmgr_dir)
+    devices = load_devices(gen_bmgr_dir, peripheral_map)
+    board_version = str(board_info.get('version') or 'unknown')
     markdown = render_markdown(board_info, board_version, devices)
 
     output_md.parent.mkdir(parents=True, exist_ok=True)

components/claw_capabilities/cap_system/CMakeLists.txt

@@ -8,4 +8,6 @@ idf_component_register(
         json
         esp_netif
         esp_wifi
+    PRIV_REQUIRES
+        esp_timer
 )