diff --git a/application/basic_demo/fatfs_image/scripts/random_demo_stress.lua b/application/basic_demo/fatfs_image/scripts/random_demo_stress.lua deleted file mode 100644 index d784e67..0000000 --- a/application/basic_demo/fatfs_image/scripts/random_demo_stress.lua +++ /dev/null @@ -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") diff --git a/components/claw_capabilities/cap_boards/tools/generate_board_skill.py b/components/claw_capabilities/cap_boards/tools/generate_board_skill.py index 49c76f5..364b787 100755 --- a/components/claw_capabilities/cap_boards/tools/generate_board_skill.py +++ b/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) diff --git a/components/claw_capabilities/cap_system/CMakeLists.txt b/components/claw_capabilities/cap_system/CMakeLists.txt index c19d4e1..ebd4713 100644 --- a/components/claw_capabilities/cap_system/CMakeLists.txt +++ b/components/claw_capabilities/cap_system/CMakeLists.txt @@ -8,4 +8,6 @@ idf_component_register( json esp_netif esp_wifi + PRIV_REQUIRES + esp_timer )