bmspy/tools/generate_ecoflow_river3_schema.py

#!/usr/bin/env python3
"""Generate diagnostic descriptors from the Apache-2.0 ha-ef-ble pr705 schema.

Run manually from a checkout. No network access or generator dependency is needed
at runtime. Only message/enum declarations reachable from the two upload messages
are retained. The generated file contains data, not upstream decoder code.
"""

import argparse
import base64
import json
from pathlib import Path
import re
from urllib.request import Request, urlopen

REPOSITORY = "rabits/ha-ef-ble"
SCALARS = {
    "bool", "bytes", "string", "float", "double", "int32", "int64", "uint32",
    "uint64", "sint32", "sint64", "fixed32", "fixed64", "sfixed32", "sfixed64",
}


def api(path):
    request = Request(f"https://api.github.com/repos/{REPOSITORY}/{path}",
                      headers={"User-Agent": "bmspy-schema-generator"})
    with urlopen(request, timeout=30) as response:
        return json.load(response)


def source_file(path, ref):
    data = api(f"contents/{path}?ref={ref}")
    if data["encoding"] != "base64":
        raise ValueError("Expected a base64-encoded source file")
    return base64.b64decode(data["content"]).decode("utf-8")


def declarations(text):
    """Yield top-level declarations without treating nested names as global."""
    position = 0
    while match := re.search(r"\b(message|enum)\s+(\w+)\s*\{", text[position:]):
        start = position + match.end()
        end, depth = start, 1
        while depth and end < len(text):
            depth += (text[end] == "{") - (text[end] == "}")
            end += 1
        if depth:
            raise ValueError("Unterminated schema declaration")
        yield *match.groups(), text[start:end - 1]
        position = end


def parse_schema(text):
    """Parse reachable flat declarations. Reject unsupported reachable types."""
    text = re.sub(r"/\*.*?\*/|//[^\n]*", "", text, flags=re.S)
    messages, enums = {}, {}
    for kind, name, body in declarations(text):
        if "{" in body:
            # EventPush has nested definitions but neither upload depends on it.
            # If a reachable type needs nesting, resolution below must fail.
            continue
        fields = {}
        for statement in body.split(";"):
            statement = statement.strip()
            if not statement:
                continue
            if kind == "message":
                match = re.fullmatch(r"(optional|repeated|required)\s+(\w+)\s+(\w+)\s*=\s*(\d+)", statement)
                if not match:
                    raise ValueError(f"Unsupported field declaration in {name}: {statement}")
                cardinality, data_type, field_name, number = match.groups()
                value = (field_name, data_type, cardinality == "repeated")
            else:
                match = re.fullmatch(r"(\w+)\s*=\s*(-?\d+)", statement)
                if not match:
                    raise ValueError(f"Unsupported enum declaration in {name}: {statement}")
                value, number = match.groups()
            number = int(number)
            if number in fields:
                raise ValueError(f"Duplicate field/enum value in {name}: {number}")
            fields[number] = value
        target = messages if kind == "message" else enums
        if name in target:
            raise ValueError(f"Duplicate declaration: {name}")
        target[name] = fields
    wanted_messages, wanted_enums = {}, {}
    pending = ["DisplayPropertyUpload", "RuntimePropertyUpload"]
    while pending:
        name = pending.pop()
        if name in SCALARS or name in wanted_messages or name in wanted_enums:
            continue
        if name in enums:
            wanted_enums[name] = enums[name]
        elif name in messages:
            wanted_messages[name] = messages[name]
            pending.extend(data_type for _, data_type, _ in messages[name].values())
        else:
            raise ValueError(f"Unresolved schema type: {name}")
    return wanted_messages, wanted_enums


def generate(messages, enums, ref):
    lines = [
        '# SPDX-License-Identifier: Apache-2.0',
        f'# Source: https://github.com/{REPOSITORY}/blob/{ref}/proto/pr705.proto',
        '# Generated by tools/generate_ecoflow_river3_schema.py. Do not edit by hand.',
        '', f'SCHEMA_COMMIT = {ref!r}', '',
        '# Field number: (name, protobuf type, repeated). Units are curated separately.',
        'MESSAGES = {',
    ]
    for name, fields in sorted(messages.items()):
        lines.append(f'    {name!r}: {{')
        lines.extend(f'        {number}: {field!r},' for number, field in sorted(fields.items()))
        lines.append('    },')
    lines.extend(['}', '', 'ENUMS = {'])
    for name, values in sorted(enums.items()):
        lines.append(f'    {name!r}: {{')
        lines.extend(f'        {number}: {value!r},' for number, value in sorted(values.items()))
        lines.append('    },')
    lines.extend(['}', ''])
    return '\n'.join(lines)


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--ref', help='Pinned upstream commit (defaults to current main)')
    args = parser.parse_args()
    ref = args.ref or api('commits/main')['sha']
    if not re.fullmatch(r'[a-f0-9]{40}', ref):
        parser.error('--ref must be a full commit SHA')
    messages, enums = parse_schema(source_file('proto/pr705.proto', ref))
    root = Path(__file__).resolve().parents[1]
    (root / 'bmspy/ecoflow_river3_schema.py').write_text(generate(messages, enums, ref), encoding='utf-8')
    print(f'Generated {len(messages)} messages and {len(enums)} enums from {ref}')
    for name in ('DisplayPropertyUpload', 'RuntimePropertyUpload'):
        print(f'{name}: {len(messages[name])} fields')


if __name__ == '__main__':
    main()