diff --git a/src/struct_frame/__init__.py b/src/struct_frame/__init__.py index 29926149..31a5750e 100644 --- a/src/struct_frame/__init__.py +++ b/src/struct_frame/__init__.py @@ -7,7 +7,18 @@ from .gql_gen import FileGqlGen from .cpp_gen import FileCppGen +from .frame_format import FrameFormat, FrameFormatCollection, parse_frame_formats +from .frame_parser_c_gen import generate_c_frame_parsers, FrameParserCGen +from .frame_parser_py_gen import generate_py_frame_parsers, FrameParserPyGen +from .frame_parser_ts_gen import generate_ts_frame_parsers, generate_js_frame_parsers, FrameParserTsGen, FrameParserJsGen +from .frame_parser_cpp_gen import generate_cpp_frame_parsers, FrameParserCppGen + from .generate import main __all__ = ["main", "FileCGen", "FileTsGen", "FileJsGen", "FilePyGen", "FileGqlGen", "FileCppGen", "version", - "NamingStyleC", "CamelToSnakeCase", "pascalCase"] + "NamingStyleC", "CamelToSnakeCase", "pascalCase", + "FrameFormat", "FrameFormatCollection", "parse_frame_formats", + "generate_c_frame_parsers", "generate_py_frame_parsers", + "generate_ts_frame_parsers", "generate_js_frame_parsers", + "generate_cpp_frame_parsers", + "FrameParserCGen", "FrameParserPyGen", "FrameParserTsGen", "FrameParserJsGen", "FrameParserCppGen"] diff --git a/src/struct_frame/frame_format.py b/src/struct_frame/frame_format.py new file mode 100644 index 00000000..22944472 --- /dev/null +++ b/src/struct_frame/frame_format.py @@ -0,0 +1,206 @@ +#!/usr/bin/env python3 +# kate: replace-tabs on; indent-width 4; + +""" +Frame Format Parser and Generator + +This module parses frame format definitions from .proto files and generates +frame parser code for multiple target languages. + +Frame formats define how messages are framed for communication, including: +- Start bytes for synchronization +- Header structure (message ID, optional length) +- Footer structure (CRC/checksum) +""" + +from proto_schema_parser.parser import Parser +from proto_schema_parser import ast + + +class FrameFormatField: + """Represents a field in a frame format definition""" + + def __init__(self, name, field_type, hex_value=None): + self.name = name + self.field_type = field_type + self.hex_value = hex_value # For start bytes with [(hex) = 0xNN] + self.is_start_byte = name.startswith('start_byte') or name in ['sync1', 'sync2', 'stx'] + self.is_crc = 'crc' in name.lower() or name.startswith('ck_') + self.is_length = 'length' in name.lower() or 'len' in name.lower() + self.is_msg_id = 'msg_id' in name.lower() or name == 'msg_id' + self.is_payload = name == 'payload' + + def __repr__(self): + return f"FrameFormatField({self.name}, {self.field_type}, hex={self.hex_value})" + + +class FrameFormat: + """ + Represents a parsed frame format definition. + + Frame formats describe how messages are framed for communication: + - Start bytes: Synchronization markers + - Header: Message ID, optional length field + - Payload: The actual message data + - Footer: CRC/checksum bytes + """ + + def __init__(self, name, comments=None): + self.name = name + self.comments = comments or [] + self.fields = [] + self.start_bytes = [] # List of (name, hex_value) tuples + self.has_crc = False + self.crc_bytes = 0 # Number of CRC bytes (usually 2) + self.has_length = False + self.length_bytes = 0 # 1 for uint8, 2 for uint16 + self.header_size = 0 # Total header size + self.footer_size = 0 # Total footer size + + def parse(self, message): + """Parse a proto message definition into a frame format""" + self.name = message.name + + for element in message.elements: + if isinstance(element, ast.Field): + hex_value = None + + # Check for [(hex) = 0xNN] option + if hasattr(element, 'options') and element.options: + for opt in element.options: + opt_name = getattr(opt, 'name', None) + opt_value = getattr(opt, 'value', None) + if opt_name and '(hex)' in str(opt_name): + # Parse hex value + try: + hex_value = int(str(opt_value), 16) + except (ValueError, TypeError): + hex_value = opt_value + + field = FrameFormatField(element.name, element.type, hex_value) + self.fields.append(field) + + # Track start bytes + if field.is_start_byte and hex_value is not None: + self.start_bytes.append((element.name, hex_value)) + self.header_size += 1 + + # Track CRC bytes + if field.is_crc: + self.has_crc = True + self.crc_bytes += 1 + self.footer_size += 1 + + # Track length field + if field.is_length: + self.has_length = True + if element.type == 'uint16': + self.length_bytes = 2 + else: + self.length_bytes = 1 + self.header_size += self.length_bytes + + # Track payload/msg_id (1 byte for msg_id) + if field.is_payload or field.is_msg_id: + self.header_size += 1 + + return True + + def get_enum_value(self): + """Get the enum value name for this frame format""" + # Convert CamelCase to UPPER_SNAKE_CASE + import re + name = re.sub('(.)([A-Z][a-z]+)', r'\1_\2', self.name) + name = re.sub('([a-z0-9])([A-Z])', r'\1_\2', name) + return name.upper() + + def __repr__(self): + return (f"FrameFormat({self.name}, start_bytes={self.start_bytes}, " + f"has_crc={self.has_crc}, has_length={self.has_length})") + + +class FrameFormatCollection: + """Collection of frame formats parsed from a proto file""" + + def __init__(self): + self.formats = {} + self.format_enum = None # The FrameFormatType enum if present + + def parse_file(self, filename): + """Parse frame formats from a proto file""" + with open(filename, 'r') as f: + result = Parser().parse(f.read()) + + for element in result.file_elements: + if isinstance(element, ast.Enum): + if element.name == 'FrameFormatType': + self.format_enum = element + + elif isinstance(element, ast.Message): + # Skip non-frame messages like BasicMessage, FrameFormatConfig + if self._is_frame_format_message(element): + frame = FrameFormat(element.name) + if frame.parse(element): + self.formats[element.name] = frame + + def _is_frame_format_message(self, message): + """Check if a message defines a frame format""" + # Frame format messages typically have start bytes or are named *Frame* + has_start = False + has_payload_or_id = False + + for element in message.elements: + if isinstance(element, ast.Field): + name = element.name.lower() + if 'start' in name or 'sync' in name or 'stx' in name: + has_start = True + if name == 'payload' or 'msg_id' in name: + has_payload_or_id = True + + # Consider it a frame format if it has start bytes and payload/msg_id, + # or if the name contains 'Frame' and has multiple fields + name_indicates_frame = 'Frame' in message.name or message.name.endswith('Frame') + + return (has_start and has_payload_or_id) or (name_indicates_frame and len(message.elements) > 1) + + def get_format_by_start_byte(self, start_byte): + """Find frame format(s) that match a given start byte""" + matches = [] + for name, fmt in self.formats.items(): + if fmt.start_bytes: + if fmt.start_bytes[0][1] == start_byte: + matches.append(fmt) + return matches + + def __iter__(self): + return iter(self.formats.values()) + + def __len__(self): + return len(self.formats) + + +def parse_frame_formats(filename): + """Parse frame formats from a proto file""" + collection = FrameFormatCollection() + collection.parse_file(filename) + return collection + + +if __name__ == '__main__': + # Test with frame_formats.proto + import sys + if len(sys.argv) > 1: + filename = sys.argv[1] + else: + filename = 'examples/frame_formats.proto' + + collection = parse_frame_formats(filename) + + print(f"Found {len(collection)} frame formats:") + for fmt in collection: + print(f"\n{fmt.name}:") + print(f" Start bytes: {fmt.start_bytes}") + print(f" Has CRC: {fmt.has_crc} ({fmt.crc_bytes} bytes)") + print(f" Has Length: {fmt.has_length} ({fmt.length_bytes} bytes)") + print(f" Header size: {fmt.header_size}") + print(f" Footer size: {fmt.footer_size}") diff --git a/src/struct_frame/frame_parser_c_gen.py b/src/struct_frame/frame_parser_c_gen.py new file mode 100644 index 00000000..a81e3ef2 --- /dev/null +++ b/src/struct_frame/frame_parser_c_gen.py @@ -0,0 +1,444 @@ +#!/usr/bin/env python3 +# kate: replace-tabs on; indent-width 4; + +""" +C Frame Parser Generator + +Generates C code for frame parsers based on frame format definitions. +""" + +from struct_frame import version +import time + + +def camel_to_snake(name): + """Convert CamelCase to snake_case""" + import re + name = re.sub('(.)([A-Z][a-z]+)', r'\1_\2', name) + name = re.sub('([a-z0-9])([A-Z])', r'\1_\2', name) + return name.lower() + + +def camel_to_upper_snake(name): + """Convert CamelCase to UPPER_SNAKE_CASE""" + return camel_to_snake(name).upper() + + +class FrameParserCGen: + """Generates C code for frame parsers""" + + @staticmethod + def generate_header(formats): + """Generate the complete frame parser header file""" + yield '/* Automatically generated frame parser header */\n' + yield '/* Generated by %s at %s. */\n\n' % (version, time.asctime()) + yield '#pragma once\n\n' + yield '#include \n' + yield '#include \n' + yield '#include \n' + yield '#include \n\n' + + # Generate frame format enum + yield '/* Frame format type enumeration */\n' + yield 'typedef enum FrameFormatType {\n' + for i, fmt in enumerate(formats): + enum_name = camel_to_upper_snake(fmt.name) + yield f' FRAME_FORMAT_{enum_name} = {i},\n' + yield '} FrameFormatType;\n\n' + + # Generate checksum function + yield '/*===========================================================================\n' + yield ' * Checksum Calculation\n' + yield ' *===========================================================================*/\n\n' + yield '''typedef struct frame_checksum { + uint8_t byte1; + uint8_t byte2; +} frame_checksum_t; + +/** + * Calculate Fletcher-16 checksum over the given data + */ +static inline frame_checksum_t frame_fletcher_checksum(const uint8_t* data, size_t length) { + frame_checksum_t ck = {0, 0}; + for (size_t i = 0; i < length; i++) { + ck.byte1 = (uint8_t)(ck.byte1 + data[i]); + ck.byte2 = (uint8_t)(ck.byte2 + ck.byte1); + } + return ck; +} + +''' + + # Generate common result type + yield '''/* Parse result */ +typedef struct frame_msg_info { + bool valid; + uint8_t msg_id; + size_t msg_len; + uint8_t* msg_data; +} frame_msg_info_t; + +''' + + # Generate individual frame format code + for fmt in formats: + yield from FrameParserCGen.generate_format(fmt) + yield '\n' + + @staticmethod + def generate_format(fmt): + """Generate C code for a single frame format""" + name = fmt.name + prefix = camel_to_snake(name) + PREFIX = camel_to_upper_snake(name) + + # Skip formats without start bytes (they need external sync) + # But still generate encode/decode helpers + + yield f'/*===========================================================================\n' + yield f' * {name} Frame Format\n' + yield f' *===========================================================================*/\n\n' + + # Generate constants + yield f'/* {name} constants */\n' + + # Start bytes + for i, (sb_name, sb_value) in enumerate(fmt.start_bytes): + yield f'#define {PREFIX}_START_BYTE{i + 1 if len(fmt.start_bytes) > 1 else ""} 0x{sb_value:02X}\n' + + yield f'#define {PREFIX}_HEADER_SIZE {fmt.header_size} /* ' + parts = [] + for sb_name, sb_value in fmt.start_bytes: + parts.append(f'{sb_name}') + parts.append('msg_id') + if fmt.has_length: + parts.append(f'length({fmt.length_bytes})') + yield ' + '.join(parts) + ' */\n' + + yield f'#define {PREFIX}_FOOTER_SIZE {fmt.footer_size} /* ' + if fmt.has_crc: + yield f'crc({fmt.crc_bytes} bytes)' + else: + yield 'no footer' + yield ' */\n' + + yield f'#define {PREFIX}_OVERHEAD ({PREFIX}_HEADER_SIZE + {PREFIX}_FOOTER_SIZE)\n\n' + + # Generate parser state enum + yield f'/* {name} parser states */\n' + yield f'typedef enum {prefix}_parser_state {{\n' + + # State machine states + if fmt.start_bytes: + for i in range(len(fmt.start_bytes)): + yield f' {PREFIX}_LOOKING_FOR_START{i + 1 if len(fmt.start_bytes) > 1 else ""} = {i},\n' + state_idx = len(fmt.start_bytes) + else: + state_idx = 0 + + yield f' {PREFIX}_GETTING_MSG_ID = {state_idx},\n' + state_idx += 1 + + if fmt.has_length: + yield f' {PREFIX}_GETTING_LENGTH = {state_idx},\n' + state_idx += 1 + + yield f' {PREFIX}_GETTING_PAYLOAD = {state_idx}\n' + yield f'}} {prefix}_parser_state_t;\n\n' + + # Generate parser structure + yield f'/* {name} parser state structure */\n' + yield f'typedef struct {prefix}_parser {{\n' + yield f' {prefix}_parser_state_t state;\n' + yield f' uint8_t* buffer;\n' + yield f' size_t buffer_max_size;\n' + yield f' size_t buffer_index;\n' + yield f' size_t packet_size;\n' + yield f' uint8_t msg_id;\n' + if fmt.has_length: + yield f' size_t msg_length; /* From length field */\n' + if fmt.length_bytes == 2: + yield f' uint8_t length_lo; /* Low byte for 16-bit length */\n' + yield f' /* User-provided function to get message length from msg_id (for non-length frames) */\n' + yield f' bool (*get_msg_length)(uint8_t msg_id, size_t* length);\n' + yield f'}} {prefix}_parser_t;\n\n' + + # Generate encode buffer structure + yield f'/* {name} encode buffer structure */\n' + yield f'typedef struct {prefix}_encode_buffer {{\n' + yield f' uint8_t* data;\n' + yield f' size_t max_size;\n' + yield f' size_t size;\n' + yield f' bool in_progress;\n' + yield f' size_t reserved_msg_size;\n' + yield f'}} {prefix}_encode_buffer_t;\n\n' + + # Generate init function + yield f'/**\n' + yield f' * Initialize a {name} parser\n' + yield f' */\n' + yield f'static inline void {prefix}_parser_init({prefix}_parser_t* parser,\n' + yield f' uint8_t* buffer, size_t buffer_size,\n' + yield f' bool (*get_msg_length)(uint8_t msg_id, size_t* length)) {{\n' + if fmt.start_bytes: + yield f' parser->state = {PREFIX}_LOOKING_FOR_START{"1" if len(fmt.start_bytes) > 1 else ""};\n' + else: + yield f' parser->state = {PREFIX}_GETTING_MSG_ID;\n' + yield f' parser->buffer = buffer;\n' + yield f' parser->buffer_max_size = buffer_size;\n' + yield f' parser->buffer_index = 0;\n' + yield f' parser->packet_size = 0;\n' + yield f' parser->msg_id = 0;\n' + if fmt.has_length: + yield f' parser->msg_length = 0;\n' + if fmt.length_bytes == 2: + yield f' parser->length_lo = 0;\n' + yield f' parser->get_msg_length = get_msg_length;\n' + yield f'}}\n\n' + + # Generate reset function + yield f'/**\n' + yield f' * Reset {name} parser state\n' + yield f' */\n' + yield f'static inline void {prefix}_parser_reset({prefix}_parser_t* parser) {{\n' + if fmt.start_bytes: + yield f' parser->state = {PREFIX}_LOOKING_FOR_START{"1" if len(fmt.start_bytes) > 1 else ""};\n' + else: + yield f' parser->state = {PREFIX}_GETTING_MSG_ID;\n' + yield f' parser->buffer_index = 0;\n' + yield f' parser->packet_size = 0;\n' + yield f' parser->msg_id = 0;\n' + if fmt.has_length: + yield f' parser->msg_length = 0;\n' + yield f'}}\n\n' + + # Generate parse_byte function + yield f'/**\n' + yield f' * Parse a single byte with {name} format\n' + yield f' * Returns frame_msg_info_t with valid=true when a complete valid message is received\n' + yield f' */\n' + yield f'static inline frame_msg_info_t {prefix}_parse_byte({prefix}_parser_t* parser, uint8_t byte) {{\n' + yield f' frame_msg_info_t result = {{false, 0, 0, NULL}};\n\n' + yield f' switch (parser->state) {{\n' + + # Generate state machine cases + if fmt.start_bytes: + for i, (sb_name, sb_value) in enumerate(fmt.start_bytes): + state_name = f'{PREFIX}_LOOKING_FOR_START{i + 1 if len(fmt.start_bytes) > 1 else ""}' + yield f' case {state_name}:\n' + yield f' if (byte == {PREFIX}_START_BYTE{i + 1 if len(fmt.start_bytes) > 1 else ""}) {{\n' + yield f' parser->buffer[{i}] = byte;\n' + yield f' parser->buffer_index = {i + 1};\n' + + if i + 1 < len(fmt.start_bytes): + next_state = f'{PREFIX}_LOOKING_FOR_START{i + 2}' + else: + next_state = f'{PREFIX}_GETTING_MSG_ID' + yield f' parser->state = {next_state};\n' + + yield f' }}' + + # Handle case where we see start byte 1 while looking for start byte 2 + # This code only runs when len(fmt.start_bytes) > 1 (i.e., i > 0 is only possible with 2+ start bytes) + if i > 0: + yield f' else if (byte == {PREFIX}_START_BYTE1) {{\n' + yield f' parser->buffer[0] = byte;\n' + yield f' parser->buffer_index = 1;\n' + yield f' parser->state = {PREFIX}_LOOKING_FOR_START2;\n' + yield f' }} else {{\n' + yield f' parser->state = {PREFIX}_LOOKING_FOR_START1;\n' + yield f' }}\n' + else: + yield '\n' + yield f' break;\n\n' + + # MSG_ID state + yield f' case {PREFIX}_GETTING_MSG_ID:\n' + yield f' parser->buffer[parser->buffer_index++] = byte;\n' + yield f' parser->msg_id = byte;\n' + + if fmt.has_length: + yield f' parser->state = {PREFIX}_GETTING_LENGTH;\n' + else: + yield f' {{\n' + yield f' size_t msg_length = 0;\n' + yield f' if (parser->get_msg_length && parser->get_msg_length(byte, &msg_length)) {{\n' + yield f' parser->packet_size = {PREFIX}_OVERHEAD + msg_length;\n' + yield f' if (parser->packet_size <= parser->buffer_max_size) {{\n' + yield f' parser->state = {PREFIX}_GETTING_PAYLOAD;\n' + yield f' }} else {{\n' + if fmt.start_bytes: + yield f' parser->state = {PREFIX}_LOOKING_FOR_START{"1" if len(fmt.start_bytes) > 1 else ""};\n' + else: + yield f' parser->state = {PREFIX}_GETTING_MSG_ID;\n' + yield f' }}\n' + yield f' }} else {{\n' + if fmt.start_bytes: + yield f' parser->state = {PREFIX}_LOOKING_FOR_START{"1" if len(fmt.start_bytes) > 1 else ""};\n' + else: + yield f' parser->state = {PREFIX}_GETTING_MSG_ID;\n' + yield f' }}\n' + yield f' }}\n' + yield f' break;\n\n' + + # LENGTH state (if applicable) + if fmt.has_length: + yield f' case {PREFIX}_GETTING_LENGTH:\n' + yield f' parser->buffer[parser->buffer_index++] = byte;\n' + if fmt.length_bytes == 1: + yield f' parser->msg_length = byte;\n' + yield f' parser->packet_size = {PREFIX}_OVERHEAD + parser->msg_length;\n' + yield f' if (parser->packet_size <= parser->buffer_max_size) {{\n' + yield f' parser->state = {PREFIX}_GETTING_PAYLOAD;\n' + yield f' }} else {{\n' + if fmt.start_bytes: + yield f' parser->state = {PREFIX}_LOOKING_FOR_START{"1" if len(fmt.start_bytes) > 1 else ""};\n' + else: + yield f' parser->state = {PREFIX}_GETTING_MSG_ID;\n' + yield f' }}\n' + else: + # 16-bit length + yield f' if (parser->buffer_index == {len(fmt.start_bytes) + 2}) {{\n' + yield f' parser->length_lo = byte;\n' + yield f' }} else {{\n' + yield f' parser->msg_length = parser->length_lo | ((size_t)byte << 8);\n' + yield f' parser->packet_size = {PREFIX}_OVERHEAD + parser->msg_length;\n' + yield f' if (parser->packet_size <= parser->buffer_max_size) {{\n' + yield f' parser->state = {PREFIX}_GETTING_PAYLOAD;\n' + yield f' }} else {{\n' + if fmt.start_bytes: + yield f' parser->state = {PREFIX}_LOOKING_FOR_START{"1" if len(fmt.start_bytes) > 1 else ""};\n' + else: + yield f' parser->state = {PREFIX}_GETTING_MSG_ID;\n' + yield f' }}\n' + yield f' }}\n' + yield f' break;\n\n' + + # PAYLOAD state + yield f' case {PREFIX}_GETTING_PAYLOAD:\n' + yield f' if (parser->buffer_index < parser->buffer_max_size) {{\n' + yield f' parser->buffer[parser->buffer_index++] = byte;\n' + yield f' }}\n\n' + yield f' if (parser->buffer_index >= parser->packet_size) {{\n' + + if fmt.has_crc: + yield f' /* Validate checksum */\n' + yield f' size_t msg_length = parser->packet_size - {PREFIX}_OVERHEAD;\n' + yield f' frame_checksum_t ck = frame_fletcher_checksum(\n' + yield f' parser->buffer + {len(fmt.start_bytes)}, msg_length + 1' + if fmt.has_length: + yield f' + {fmt.length_bytes}' + yield f');\n\n' + yield f' if (ck.byte1 == parser->buffer[parser->packet_size - 2] &&\n' + yield f' ck.byte2 == parser->buffer[parser->packet_size - 1]) {{\n' + yield f' result.valid = true;\n' + yield f' result.msg_id = parser->msg_id;\n' + yield f' result.msg_len = msg_length;\n' + yield f' result.msg_data = parser->buffer + {PREFIX}_HEADER_SIZE;\n' + yield f' }}\n' + else: + yield f' result.valid = true;\n' + yield f' result.msg_id = parser->msg_id;\n' + yield f' result.msg_len = parser->packet_size - {PREFIX}_OVERHEAD;\n' + yield f' result.msg_data = parser->buffer + {PREFIX}_HEADER_SIZE;\n' + + if fmt.start_bytes: + yield f' parser->state = {PREFIX}_LOOKING_FOR_START{"1" if len(fmt.start_bytes) > 1 else ""};\n' + else: + yield f' parser->state = {PREFIX}_GETTING_MSG_ID;\n' + yield f' }}\n' + yield f' break;\n' + + yield f' }}\n\n' + yield f' return result;\n' + yield f'}}\n\n' + + # Generate encode function + yield f'/**\n' + yield f' * Encode a message with {name} format\n' + yield f' * Returns the number of bytes written, or 0 on failure\n' + yield f' */\n' + yield f'static inline size_t {prefix}_encode(uint8_t* buffer, size_t buffer_size,\n' + yield f' uint8_t msg_id, const uint8_t* msg, size_t msg_size) {{\n' + yield f' size_t total_size = {PREFIX}_OVERHEAD + msg_size;\n' + yield f' if (buffer_size < total_size) {{\n' + yield f' return 0;\n' + yield f' }}\n\n' + + # Write header + idx = 0 + for i, (sb_name, sb_value) in enumerate(fmt.start_bytes): + yield f' buffer[{idx}] = {PREFIX}_START_BYTE{i + 1 if len(fmt.start_bytes) > 1 else ""};\n' + idx += 1 + yield f' buffer[{idx}] = msg_id;\n' + idx += 1 + + if fmt.has_length: + if fmt.length_bytes == 1: + yield f' buffer[{idx}] = (uint8_t)msg_size;\n' + else: + yield f' buffer[{idx}] = (uint8_t)(msg_size & 0xFF);\n' + yield f' buffer[{idx + 1}] = (uint8_t)((msg_size >> 8) & 0xFF);\n' + idx += fmt.length_bytes + + yield f'\n /* Write message data */\n' + yield f' if (msg_size > 0 && msg != NULL) {{\n' + yield f' memcpy(buffer + {PREFIX}_HEADER_SIZE, msg, msg_size);\n' + yield f' }}\n\n' + + if fmt.has_crc: + yield f' /* Calculate checksum */\n' + yield f' frame_checksum_t ck = frame_fletcher_checksum(buffer + {len(fmt.start_bytes)}, msg_size + 1' + if fmt.has_length: + yield f' + {fmt.length_bytes}' + yield f');\n' + yield f' buffer[{PREFIX}_HEADER_SIZE + msg_size] = ck.byte1;\n' + yield f' buffer[{PREFIX}_HEADER_SIZE + msg_size + 1] = ck.byte2;\n' + + yield f'\n return total_size;\n' + yield f'}}\n\n' + + # Generate validate_packet function + yield f'/**\n' + yield f' * Validate a complete {name} packet in a buffer\n' + yield f' */\n' + yield f'static inline frame_msg_info_t {prefix}_validate_packet(const uint8_t* buffer, size_t length) {{\n' + yield f' frame_msg_info_t result = {{false, 0, 0, NULL}};\n\n' + yield f' if (length < {PREFIX}_OVERHEAD) {{\n' + yield f' return result;\n' + yield f' }}\n\n' + + # Check start bytes + if fmt.start_bytes: + for i, (sb_name, sb_value) in enumerate(fmt.start_bytes): + yield f' if (buffer[{i}] != {PREFIX}_START_BYTE{i + 1 if len(fmt.start_bytes) > 1 else ""}) {{\n' + yield f' return result;\n' + yield f' }}\n' + + yield f'\n size_t msg_length = length - {PREFIX}_OVERHEAD;\n\n' + + if fmt.has_crc: + yield f' /* Validate checksum */\n' + yield f' frame_checksum_t ck = frame_fletcher_checksum(buffer + {len(fmt.start_bytes)}, msg_length + 1' + if fmt.has_length: + yield f' + {fmt.length_bytes}' + yield f');\n' + yield f' if (ck.byte1 == buffer[length - 2] && ck.byte2 == buffer[length - 1]) {{\n' + yield f' result.valid = true;\n' + yield f' result.msg_id = buffer[{len(fmt.start_bytes)}];\n' + yield f' result.msg_len = msg_length;\n' + yield f' result.msg_data = (uint8_t*)(buffer + {PREFIX}_HEADER_SIZE);\n' + yield f' }}\n' + else: + yield f' result.valid = true;\n' + yield f' result.msg_id = buffer[{len(fmt.start_bytes)}];\n' + yield f' result.msg_len = msg_length;\n' + yield f' result.msg_data = (uint8_t*)(buffer + {PREFIX}_HEADER_SIZE);\n' + + yield f'\n return result;\n' + yield f'}}\n\n' + + +def generate_c_frame_parsers(formats): + """Generate C frame parser code from frame format definitions""" + return ''.join(FrameParserCGen.generate_header(formats)) diff --git a/src/struct_frame/frame_parser_cpp_gen.py b/src/struct_frame/frame_parser_cpp_gen.py new file mode 100644 index 00000000..7ece1a65 --- /dev/null +++ b/src/struct_frame/frame_parser_cpp_gen.py @@ -0,0 +1,459 @@ +#!/usr/bin/env python3 +# kate: replace-tabs on; indent-width 4; + +""" +C++ Frame Parser Generator + +Generates C++ code for frame parsers based on frame format definitions. +""" + +from struct_frame import version +import time + + +def camel_to_snake(name): + """Convert CamelCase to snake_case""" + import re + name = re.sub('(.)([A-Z][a-z]+)', r'\1_\2', name) + name = re.sub('([a-z0-9])([A-Z])', r'\1_\2', name) + return name.lower() + + +def camel_to_upper_snake(name): + """Convert CamelCase to UPPER_SNAKE_CASE""" + return camel_to_snake(name).upper() + + +class FrameParserCppGen: + """Generates C++ code for frame parsers""" + + @staticmethod + def generate_header(formats): + """Generate the complete frame parser header file""" + yield '/* Automatically generated frame parser header */\n' + yield f'/* Generated by {version} at {time.asctime()}. */\n\n' + yield '#pragma once\n\n' + yield '#include \n' + yield '#include \n' + yield '#include \n' + yield '#include \n\n' + yield 'namespace FrameParsers {\n\n' + + # Generate frame format enum + yield '// Frame format type enumeration\n' + yield 'enum class FrameFormatType {\n' + for i, fmt in enumerate(formats): + enum_name = camel_to_upper_snake(fmt.name) + yield f' {enum_name} = {i},\n' + yield '};\n\n' + + # Generate checksum struct and function + yield '''// Checksum result +struct FrameChecksum { + uint8_t byte1; + uint8_t byte2; +}; + +// Fletcher-16 checksum calculation +inline FrameChecksum fletcher_checksum(const uint8_t* data, size_t length) { + FrameChecksum ck{0, 0}; + for (size_t i = 0; i < length; i++) { + ck.byte1 = static_cast(ck.byte1 + data[i]); + ck.byte2 = static_cast(ck.byte2 + ck.byte1); + } + return ck; +} + +''' + + # Generate common result struct + yield '''// Parse result +struct FrameMsgInfo { + bool valid; + uint8_t msg_id; + size_t msg_len; + uint8_t* msg_data; + + FrameMsgInfo() : valid(false), msg_id(0), msg_len(0), msg_data(nullptr) {} + FrameMsgInfo(bool v, uint8_t id, size_t len, uint8_t* data) + : valid(v), msg_id(id), msg_len(len), msg_data(data) {} +}; + +''' + + # Generate individual frame format classes + for fmt in formats: + yield from FrameParserCppGen.generate_format(fmt) + yield '\n' + + yield '} // namespace FrameParsers\n' + + @staticmethod + def generate_format(fmt): + """Generate C++ code for a single frame format""" + name = fmt.name + class_name = name + PREFIX = camel_to_upper_snake(name) + + yield f'// =============================================================================\n' + yield f'// {name} Frame Format\n' + yield f'// =============================================================================\n\n' + + # Generate parser state enum + yield f'enum class {class_name}ParserState : uint8_t {{\n' + state_idx = 0 + if fmt.start_bytes: + for i in range(len(fmt.start_bytes)): + state_name = f'LookingForStart{i + 1}' if len(fmt.start_bytes) > 1 else 'LookingForStart' + yield f' {state_name} = {state_idx},\n' + state_idx += 1 + yield f' GettingMsgId = {state_idx},\n' + state_idx += 1 + if fmt.has_length: + yield f' GettingLength = {state_idx},\n' + state_idx += 1 + yield f' GettingPayload = {state_idx}\n' + yield '};\n\n' + + # Generate constants + yield f'// {name} constants\n' + for i, (sb_name, sb_value) in enumerate(fmt.start_bytes): + const_name = f'{PREFIX}_START_BYTE{i + 1}' if len(fmt.start_bytes) > 1 else f'{PREFIX}_START_BYTE' + yield f'constexpr uint8_t {const_name} = 0x{sb_value:02X};\n' + yield f'constexpr size_t {PREFIX}_HEADER_SIZE = {fmt.header_size};\n' + yield f'constexpr size_t {PREFIX}_FOOTER_SIZE = {fmt.footer_size};\n' + yield f'constexpr size_t {PREFIX}_OVERHEAD = {PREFIX}_HEADER_SIZE + {PREFIX}_FOOTER_SIZE;\n' + if fmt.has_length: + yield f'constexpr size_t {PREFIX}_LENGTH_BYTES = {fmt.length_bytes};\n' + yield '\n' + + # Generate encode buffer class + yield f'/**\n' + yield f' * {name} Encode Buffer\n' + yield f' */\n' + yield f'class {class_name}EncodeBuffer {{\n' + yield f'public:\n' + yield f' {class_name}EncodeBuffer(uint8_t* data, size_t max_size)\n' + yield f' : data_(data), max_size_(max_size), size_(0), in_progress_(false) {{}}\n\n' + yield f' void reset() {{\n' + yield f' size_ = 0;\n' + yield f' in_progress_ = false;\n' + yield f' }}\n\n' + yield f' uint8_t* data() {{ return data_; }}\n' + yield f' const uint8_t* data() const {{ return data_; }}\n' + yield f' size_t size() const {{ return size_; }}\n' + yield f' size_t max_size() const {{ return max_size_; }}\n' + yield f' bool in_progress() const {{ return in_progress_; }}\n\n' + + # encode method + yield f' /**\n' + yield f' * Encode a message into the buffer\n' + yield f' */\n' + yield f' bool encode(uint8_t msg_id, const void* msg, size_t msg_size) {{\n' + yield f' if (in_progress_) return false;\n\n' + yield f' size_t total_size = {PREFIX}_OVERHEAD + msg_size;\n' + yield f' if (size_ + total_size > max_size_) return false;\n\n' + yield f' uint8_t* packet_start = data_ + size_;\n\n' + yield f' // Write header\n' + idx = 0 + for i, (sb_name, sb_value) in enumerate(fmt.start_bytes): + const_name = f'{PREFIX}_START_BYTE{i + 1}' if len(fmt.start_bytes) > 1 else f'{PREFIX}_START_BYTE' + yield f' packet_start[{idx}] = {const_name};\n' + idx += 1 + yield f' packet_start[{idx}] = msg_id;\n' + idx += 1 + if fmt.has_length: + if fmt.length_bytes == 1: + yield f' packet_start[{idx}] = static_cast(msg_size);\n' + else: + yield f' packet_start[{idx}] = static_cast(msg_size & 0xFF);\n' + yield f' packet_start[{idx + 1}] = static_cast((msg_size >> 8) & 0xFF);\n' + idx += fmt.length_bytes + + yield f'\n // Write message data\n' + yield f' if (msg_size > 0 && msg != nullptr) {{\n' + yield f' std::memcpy(packet_start + {PREFIX}_HEADER_SIZE, msg, msg_size);\n' + yield f' }}\n\n' + + if fmt.has_crc: + yield f' // Calculate checksum\n' + crc_data_start = len(fmt.start_bytes) + crc_data_len = f'msg_size + 1' + if fmt.has_length: + crc_data_len += f' + {fmt.length_bytes}' + yield f' FrameChecksum ck = fletcher_checksum(packet_start + {crc_data_start}, {crc_data_len});\n' + yield f' packet_start[{PREFIX}_HEADER_SIZE + msg_size] = ck.byte1;\n' + yield f' packet_start[{PREFIX}_HEADER_SIZE + msg_size + 1] = ck.byte2;\n' + + yield f'\n size_ += total_size;\n' + yield f' return true;\n' + yield f' }}\n\n' + + yield f'private:\n' + yield f' uint8_t* data_;\n' + yield f' size_t max_size_;\n' + yield f' size_t size_;\n' + yield f' bool in_progress_;\n' + yield f'}};\n\n' + + # Generate parser class + yield f'/**\n' + yield f' * {name} Frame Parser\n' + yield f' */\n' + yield f'class {class_name}Parser {{\n' + yield f'public:\n' + yield f' using MsgLengthCallback = std::function;\n\n' + yield f' {class_name}Parser(uint8_t* buffer, size_t buffer_size, MsgLengthCallback msg_length_cb = nullptr)\n' + yield f' : state_({class_name}ParserState::' + if fmt.start_bytes: + yield f'LookingForStart{"1" if len(fmt.start_bytes) > 1 else ""}' + else: + yield 'GettingMsgId' + yield f'),\n' + yield f' buffer_(buffer),\n' + yield f' buffer_max_size_(buffer_size),\n' + yield f' buffer_index_(0),\n' + yield f' packet_size_(0),\n' + yield f' msg_id_(0),\n' + if fmt.has_length: + yield f' msg_length_(0),\n' + if fmt.length_bytes == 2: + yield f' length_lo_(0),\n' + yield f' get_msg_length_(std::move(msg_length_cb)) {{}}\n\n' + + # reset method + yield f' void reset() {{\n' + if fmt.start_bytes: + yield f' state_ = {class_name}ParserState::LookingForStart{"1" if len(fmt.start_bytes) > 1 else ""};\n' + else: + yield f' state_ = {class_name}ParserState::GettingMsgId;\n' + yield f' buffer_index_ = 0;\n' + yield f' packet_size_ = 0;\n' + yield f' msg_id_ = 0;\n' + if fmt.has_length: + yield f' msg_length_ = 0;\n' + yield f' }}\n\n' + + # parse_byte method + yield f' /**\n' + yield f' * Parse a single byte\n' + yield f' * Returns FrameMsgInfo with valid=true when a complete valid message is received\n' + yield f' */\n' + yield f' FrameMsgInfo parse_byte(uint8_t byte) {{\n' + yield f' FrameMsgInfo result;\n\n' + yield f' switch (state_) {{\n' + + # State machine cases + if fmt.start_bytes: + for i, (sb_name, sb_value) in enumerate(fmt.start_bytes): + state_name = f'LookingForStart{i + 1}' if len(fmt.start_bytes) > 1 else 'LookingForStart' + const_name = f'{PREFIX}_START_BYTE{i + 1}' if len(fmt.start_bytes) > 1 else f'{PREFIX}_START_BYTE' + + yield f' case {class_name}ParserState::{state_name}:\n' + yield f' if (byte == {const_name}) {{\n' + yield f' buffer_[{i}] = byte;\n' + yield f' buffer_index_ = {i + 1};\n' + + if i + 1 < len(fmt.start_bytes): + next_state = f'LookingForStart{i + 2}' + else: + next_state = 'GettingMsgId' + yield f' state_ = {class_name}ParserState::{next_state};\n' + yield f' }}' + + # This code only runs when len(fmt.start_bytes) > 1 (i.e., i > 0 is only possible with 2+ start bytes) + if i > 0: + yield f' else if (byte == {PREFIX}_START_BYTE1) {{\n' + yield f' buffer_[0] = byte;\n' + yield f' buffer_index_ = 1;\n' + yield f' state_ = {class_name}ParserState::LookingForStart2;\n' + yield f' }} else {{\n' + yield f' state_ = {class_name}ParserState::LookingForStart1;\n' + yield f' }}\n' + else: + yield '\n' + yield f' break;\n\n' + + # GettingMsgId state + yield f' case {class_name}ParserState::GettingMsgId: {{\n' + yield f' buffer_[buffer_index_++] = byte;\n' + yield f' msg_id_ = byte;\n\n' + + if fmt.has_length: + yield f' state_ = {class_name}ParserState::GettingLength;\n' + else: + yield f' size_t msg_length = 0;\n' + yield f' if (get_msg_length_ && get_msg_length_(byte, &msg_length)) {{\n' + yield f' packet_size_ = {PREFIX}_OVERHEAD + msg_length;\n' + yield f' if (packet_size_ <= buffer_max_size_) {{\n' + yield f' state_ = {class_name}ParserState::GettingPayload;\n' + yield f' }} else {{\n' + reset_state = 'LookingForStart1' if len(fmt.start_bytes) > 1 else ('LookingForStart' if fmt.start_bytes else 'GettingMsgId') + yield f' state_ = {class_name}ParserState::{reset_state};\n' + yield f' }}\n' + yield f' }} else {{\n' + yield f' state_ = {class_name}ParserState::{reset_state};\n' + yield f' }}\n' + yield f' break;\n' + yield f' }}\n\n' + + # GettingLength state (if applicable) + if fmt.has_length: + yield f' case {class_name}ParserState::GettingLength:\n' + yield f' buffer_[buffer_index_++] = byte;\n' + if fmt.length_bytes == 1: + yield f' msg_length_ = byte;\n' + yield f' packet_size_ = {PREFIX}_OVERHEAD + msg_length_;\n' + yield f' if (packet_size_ <= buffer_max_size_) {{\n' + yield f' state_ = {class_name}ParserState::GettingPayload;\n' + yield f' }} else {{\n' + reset_state = 'LookingForStart1' if len(fmt.start_bytes) > 1 else ('LookingForStart' if fmt.start_bytes else 'GettingMsgId') + yield f' state_ = {class_name}ParserState::{reset_state};\n' + yield f' }}\n' + else: + yield f' if (buffer_index_ == {len(fmt.start_bytes) + 2}) {{\n' + yield f' length_lo_ = byte;\n' + yield f' }} else {{\n' + yield f' msg_length_ = length_lo_ | (static_cast(byte) << 8);\n' + yield f' packet_size_ = {PREFIX}_OVERHEAD + msg_length_;\n' + yield f' if (packet_size_ <= buffer_max_size_) {{\n' + yield f' state_ = {class_name}ParserState::GettingPayload;\n' + yield f' }} else {{\n' + reset_state = 'LookingForStart1' if len(fmt.start_bytes) > 1 else ('LookingForStart' if fmt.start_bytes else 'GettingMsgId') + yield f' state_ = {class_name}ParserState::{reset_state};\n' + yield f' }}\n' + yield f' }}\n' + yield f' break;\n\n' + + # GettingPayload state + yield f' case {class_name}ParserState::GettingPayload:\n' + yield f' if (buffer_index_ < buffer_max_size_) {{\n' + yield f' buffer_[buffer_index_++] = byte;\n' + yield f' }}\n\n' + yield f' if (buffer_index_ >= packet_size_) {{\n' + + if fmt.has_crc: + yield f' // Validate checksum\n' + yield f' size_t msg_length = packet_size_ - {PREFIX}_OVERHEAD;\n' + crc_data_start = len(fmt.start_bytes) + crc_data_len = f'msg_length + 1' + if fmt.has_length: + crc_data_len += f' + {fmt.length_bytes}' + yield f' FrameChecksum ck = fletcher_checksum(buffer_ + {crc_data_start}, {crc_data_len});\n\n' + yield f' if (ck.byte1 == buffer_[packet_size_ - 2] &&\n' + yield f' ck.byte2 == buffer_[packet_size_ - 1]) {{\n' + yield f' result.valid = true;\n' + yield f' result.msg_id = msg_id_;\n' + yield f' result.msg_len = msg_length;\n' + yield f' result.msg_data = buffer_ + {PREFIX}_HEADER_SIZE;\n' + yield f' }}\n' + else: + yield f' result.valid = true;\n' + yield f' result.msg_id = msg_id_;\n' + yield f' result.msg_len = packet_size_ - {PREFIX}_OVERHEAD;\n' + yield f' result.msg_data = buffer_ + {PREFIX}_HEADER_SIZE;\n' + + reset_state = 'LookingForStart1' if len(fmt.start_bytes) > 1 else ('LookingForStart' if fmt.start_bytes else 'GettingMsgId') + yield f' state_ = {class_name}ParserState::{reset_state};\n' + yield f' }}\n' + yield f' break;\n' + + yield f' }}\n\n' + yield f' return result;\n' + yield f' }}\n\n' + + yield f'private:\n' + yield f' {class_name}ParserState state_;\n' + yield f' uint8_t* buffer_;\n' + yield f' size_t buffer_max_size_;\n' + yield f' size_t buffer_index_;\n' + yield f' size_t packet_size_;\n' + yield f' uint8_t msg_id_;\n' + if fmt.has_length: + yield f' size_t msg_length_;\n' + if fmt.length_bytes == 2: + yield f' uint8_t length_lo_;\n' + yield f' MsgLengthCallback get_msg_length_;\n' + yield f'}};\n\n' + + # Static encode function + yield f'/**\n' + yield f' * Encode a message with {name} format\n' + yield f' * Returns the number of bytes written, or 0 on failure\n' + yield f' */\n' + yield f'inline size_t {camel_to_snake(name)}_encode(uint8_t* buffer, size_t buffer_size,\n' + yield f' uint8_t msg_id, const uint8_t* msg, size_t msg_size) {{\n' + yield f' size_t total_size = {PREFIX}_OVERHEAD + msg_size;\n' + yield f' if (buffer_size < total_size) return 0;\n\n' + + idx = 0 + for i, (sb_name, sb_value) in enumerate(fmt.start_bytes): + const_name = f'{PREFIX}_START_BYTE{i + 1}' if len(fmt.start_bytes) > 1 else f'{PREFIX}_START_BYTE' + yield f' buffer[{idx}] = {const_name};\n' + idx += 1 + yield f' buffer[{idx}] = msg_id;\n' + idx += 1 + + if fmt.has_length: + if fmt.length_bytes == 1: + yield f' buffer[{idx}] = static_cast(msg_size);\n' + else: + yield f' buffer[{idx}] = static_cast(msg_size & 0xFF);\n' + yield f' buffer[{idx + 1}] = static_cast((msg_size >> 8) & 0xFF);\n' + idx += fmt.length_bytes + + yield f'\n if (msg_size > 0 && msg != nullptr) {{\n' + yield f' std::memcpy(buffer + {PREFIX}_HEADER_SIZE, msg, msg_size);\n' + yield f' }}\n\n' + + if fmt.has_crc: + crc_data_start = len(fmt.start_bytes) + crc_data_len = f'msg_size + 1' + if fmt.has_length: + crc_data_len += f' + {fmt.length_bytes}' + yield f' FrameChecksum ck = fletcher_checksum(buffer + {crc_data_start}, {crc_data_len});\n' + yield f' buffer[{PREFIX}_HEADER_SIZE + msg_size] = ck.byte1;\n' + yield f' buffer[{PREFIX}_HEADER_SIZE + msg_size + 1] = ck.byte2;\n' + + yield f'\n return total_size;\n' + yield f'}}\n\n' + + # Static validate_packet function + yield f'/**\n' + yield f' * Validate a complete {name} packet in a buffer\n' + yield f' */\n' + yield f'inline FrameMsgInfo {camel_to_snake(name)}_validate_packet(const uint8_t* buffer, size_t length) {{\n' + yield f' FrameMsgInfo result;\n\n' + yield f' if (length < {PREFIX}_OVERHEAD) return result;\n\n' + + for i, (sb_name, sb_value) in enumerate(fmt.start_bytes): + const_name = f'{PREFIX}_START_BYTE{i + 1}' if len(fmt.start_bytes) > 1 else f'{PREFIX}_START_BYTE' + yield f' if (buffer[{i}] != {const_name}) return result;\n' + + yield f'\n size_t msg_length = length - {PREFIX}_OVERHEAD;\n\n' + + if fmt.has_crc: + crc_data_start = len(fmt.start_bytes) + crc_data_len = f'msg_length + 1' + if fmt.has_length: + crc_data_len += f' + {fmt.length_bytes}' + yield f' FrameChecksum ck = fletcher_checksum(buffer + {crc_data_start}, {crc_data_len});\n' + yield f' if (ck.byte1 == buffer[length - 2] && ck.byte2 == buffer[length - 1]) {{\n' + yield f' result.valid = true;\n' + yield f' result.msg_id = buffer[{len(fmt.start_bytes)}];\n' + yield f' result.msg_len = msg_length;\n' + yield f' result.msg_data = const_cast(buffer + {PREFIX}_HEADER_SIZE);\n' + yield f' }}\n' + else: + yield f' result.valid = true;\n' + yield f' result.msg_id = buffer[{len(fmt.start_bytes)}];\n' + yield f' result.msg_len = msg_length;\n' + yield f' result.msg_data = const_cast(buffer + {PREFIX}_HEADER_SIZE);\n' + + yield f'\n return result;\n' + yield f'}}\n\n' + + +def generate_cpp_frame_parsers(formats): + """Generate C++ frame parser code from frame format definitions""" + return ''.join(FrameParserCppGen.generate_header(formats)) diff --git a/src/struct_frame/frame_parser_py_gen.py b/src/struct_frame/frame_parser_py_gen.py new file mode 100644 index 00000000..8d089c00 --- /dev/null +++ b/src/struct_frame/frame_parser_py_gen.py @@ -0,0 +1,368 @@ +#!/usr/bin/env python3 +# kate: replace-tabs on; indent-width 4; + +""" +Python Frame Parser Generator + +Generates Python code for frame parsers based on frame format definitions. +""" + +from struct_frame import version +import time + + +def camel_to_snake(name): + """Convert CamelCase to snake_case""" + import re + name = re.sub('(.)([A-Z][a-z]+)', r'\1_\2', name) + name = re.sub('([a-z0-9])([A-Z])', r'\1_\2', name) + return name.lower() + + +def camel_to_upper_snake(name): + """Convert CamelCase to UPPER_SNAKE_CASE""" + return camel_to_snake(name).upper() + + +class FrameParserPyGen: + """Generates Python code for frame parsers""" + + @staticmethod + def generate(formats): + """Generate the complete frame parser Python file""" + yield '# Automatically generated frame parser\n' + yield f'# Generated by {version} at {time.asctime()}.\n\n' + yield 'from enum import Enum\n' + yield 'from typing import Optional, Callable, Dict, List, Tuple, Union\n' + yield 'from dataclasses import dataclass\n\n' + + # Generate frame format enum + yield '# Frame format type enumeration\n' + yield 'class FrameFormatType(Enum):\n' + for i, fmt in enumerate(formats): + enum_name = camel_to_upper_snake(fmt.name) + yield f' {enum_name} = {i}\n' + yield '\n\n' + + # Generate checksum function + yield '''def fletcher_checksum(buffer: Union[bytes, List[int]], start: int = 0, end: int = None) -> Tuple[int, int]: + """Calculate Fletcher-16 checksum over the given data""" + if end is None: + end = len(buffer) + + byte1 = 0 + byte2 = 0 + + for x in range(start, end): + byte1 = (byte1 + buffer[x]) % 256 + byte2 = (byte2 + byte1) % 256 + + return (byte1, byte2) + + +''' + + # Generate common result class + yield '''@dataclass +class FrameMsgInfo: + """Parse result containing message information""" + valid: bool = False + msg_id: int = 0 + msg_len: int = 0 + msg_data: bytes = b'' + + +''' + + # Generate individual frame format classes + for fmt in formats: + yield from FrameParserPyGen.generate_format(fmt) + yield '\n\n' + + @staticmethod + def generate_format(fmt): + """Generate Python code for a single frame format""" + name = fmt.name + class_name = name + prefix = camel_to_snake(name) + PREFIX = camel_to_upper_snake(name) + + yield f'# =============================================================================\n' + yield f'# {name} Frame Format\n' + yield f'# =============================================================================\n\n' + + # Generate parser state enum + yield f'class {class_name}ParserState(Enum):\n' + state_idx = 0 + if fmt.start_bytes: + for i in range(len(fmt.start_bytes)): + state_name = f'LOOKING_FOR_START{i + 1}' if len(fmt.start_bytes) > 1 else 'LOOKING_FOR_START' + yield f' {state_name} = {state_idx}\n' + state_idx += 1 + yield f' GETTING_MSG_ID = {state_idx}\n' + state_idx += 1 + if fmt.has_length: + yield f' GETTING_LENGTH = {state_idx}\n' + state_idx += 1 + yield f' GETTING_PAYLOAD = {state_idx}\n\n\n' + + # Generate frame format class + yield f'class {class_name}:\n' + yield f' """\n' + yield f' {name} - Frame format parser and encoder\n' + yield f' \n' + yield f' Format: ' + parts = [] + for sb_name, sb_value in fmt.start_bytes: + parts.append(f'[{sb_name.upper()}=0x{sb_value:02X}]') + parts.append('[MSG_ID]') + if fmt.has_length: + parts.append(f'[LEN{"16" if fmt.length_bytes == 2 else ""}]') + parts.append('[MSG...]') + if fmt.has_crc: + parts.append('[CRC1] [CRC2]') + yield ' '.join(parts) + yield '\n """\n\n' + + # Constants + for i, (sb_name, sb_value) in enumerate(fmt.start_bytes): + const_name = f'START_BYTE{i + 1}' if len(fmt.start_bytes) > 1 else 'START_BYTE' + yield f' {const_name} = 0x{sb_value:02X}\n' + yield f' HEADER_SIZE = {fmt.header_size}\n' + yield f' FOOTER_SIZE = {fmt.footer_size}\n' + yield f' OVERHEAD = {fmt.header_size + fmt.footer_size}\n' + if fmt.has_length: + yield f' LENGTH_BYTES = {fmt.length_bytes}\n' + yield '\n' + + # __init__ + yield f' def __init__(self, get_msg_length: Callable[[int], int] = None):\n' + yield f' """\n' + yield f' Initialize the {name} parser\n' + yield f' \n' + yield f' Args:\n' + yield f' get_msg_length: Callback function to get message length from msg_id\n' + yield f' Required for formats without a length field\n' + yield f' """\n' + yield f' self.get_msg_length = get_msg_length\n' + yield f' self.reset()\n\n' + + # reset method + yield f' def reset(self):\n' + yield f' """Reset parser state"""\n' + if fmt.start_bytes: + state_name = 'LOOKING_FOR_START1' if len(fmt.start_bytes) > 1 else 'LOOKING_FOR_START' + yield f' self.state = {class_name}ParserState.{state_name}\n' + else: + yield f' self.state = {class_name}ParserState.GETTING_MSG_ID\n' + yield f' self.buffer = []\n' + yield f' self.packet_size = 0\n' + yield f' self.msg_id = 0\n' + if fmt.has_length: + yield f' self.msg_length = 0\n' + if fmt.length_bytes == 2: + yield f' self.length_lo = 0\n' + yield '\n' + + # parse_byte method + yield f' def parse_byte(self, byte: int) -> FrameMsgInfo:\n' + yield f' """\n' + yield f' Parse a single byte\n' + yield f' \n' + yield f' Returns:\n' + yield f' FrameMsgInfo with valid=True when a complete valid message is received\n' + yield f' """\n' + yield f' result = FrameMsgInfo()\n\n' + + # State machine + if fmt.start_bytes: + for i, (sb_name, sb_value) in enumerate(fmt.start_bytes): + state_name = f'LOOKING_FOR_START{i + 1}' if len(fmt.start_bytes) > 1 else 'LOOKING_FOR_START' + const_name = f'self.START_BYTE{i + 1}' if len(fmt.start_bytes) > 1 else 'self.START_BYTE' + + yield f' if self.state == {class_name}ParserState.{state_name}:\n' + yield f' if byte == {const_name}:\n' + + if i == 0: + yield f' self.buffer = [byte]\n' + else: + yield f' self.buffer.append(byte)\n' + + if i + 1 < len(fmt.start_bytes): + next_state = f'LOOKING_FOR_START{i + 2}' + else: + next_state = 'GETTING_MSG_ID' + yield f' self.state = {class_name}ParserState.{next_state}\n' + + # Handle seeing start byte 1 while looking for later bytes + # This code only runs when len(fmt.start_bytes) > 1 (i.e., i > 0 is only possible with 2+ start bytes) + if i > 0: + yield f' elif byte == self.START_BYTE1:\n' + yield f' self.buffer = [byte]\n' + yield f' self.state = {class_name}ParserState.LOOKING_FOR_START2\n' + yield f' else:\n' + yield f' self.state = {class_name}ParserState.LOOKING_FOR_START1\n' + yield '\n' + + # GETTING_MSG_ID state + if fmt.start_bytes: + yield f' elif self.state == {class_name}ParserState.GETTING_MSG_ID:\n' + else: + yield f' if self.state == {class_name}ParserState.GETTING_MSG_ID:\n' + yield f' self.buffer.append(byte)\n' + yield f' self.msg_id = byte\n' + + if fmt.has_length: + yield f' self.state = {class_name}ParserState.GETTING_LENGTH\n' + else: + yield f' if self.get_msg_length:\n' + yield f' msg_length = self.get_msg_length(byte)\n' + yield f' if msg_length is not None:\n' + yield f' self.packet_size = self.OVERHEAD + msg_length\n' + yield f' self.state = {class_name}ParserState.GETTING_PAYLOAD\n' + yield f' else:\n' + reset_state = 'LOOKING_FOR_START1' if len(fmt.start_bytes) > 1 else ('LOOKING_FOR_START' if fmt.start_bytes else 'GETTING_MSG_ID') + yield f' self.state = {class_name}ParserState.{reset_state}\n' + yield f' else:\n' + yield f' self.state = {class_name}ParserState.{reset_state}\n' + yield '\n' + + # GETTING_LENGTH state (if applicable) + if fmt.has_length: + yield f' elif self.state == {class_name}ParserState.GETTING_LENGTH:\n' + yield f' self.buffer.append(byte)\n' + if fmt.length_bytes == 1: + yield f' self.msg_length = byte\n' + yield f' self.packet_size = self.OVERHEAD + self.msg_length\n' + yield f' self.state = {class_name}ParserState.GETTING_PAYLOAD\n' + else: + yield f' if len(self.buffer) == {len(fmt.start_bytes) + 2}:\n' + yield f' self.length_lo = byte\n' + yield f' else:\n' + yield f' self.msg_length = self.length_lo | (byte << 8)\n' + yield f' self.packet_size = self.OVERHEAD + self.msg_length\n' + yield f' self.state = {class_name}ParserState.GETTING_PAYLOAD\n' + yield '\n' + + # GETTING_PAYLOAD state + yield f' elif self.state == {class_name}ParserState.GETTING_PAYLOAD:\n' + yield f' self.buffer.append(byte)\n\n' + yield f' if len(self.buffer) >= self.packet_size:\n' + + if fmt.has_crc: + yield f' # Validate checksum\n' + yield f' msg_length = self.packet_size - self.OVERHEAD\n' + crc_start = len(fmt.start_bytes) + crc_len = f'msg_length + 1' + if fmt.has_length: + crc_len += f' + {fmt.length_bytes}' + yield f' ck = fletcher_checksum(self.buffer, {crc_start}, {crc_start} + {crc_len})\n' + yield f' if ck[0] == self.buffer[-2] and ck[1] == self.buffer[-1]:\n' + yield f' result.valid = True\n' + yield f' result.msg_id = self.msg_id\n' + yield f' result.msg_len = msg_length\n' + yield f' result.msg_data = bytes(self.buffer[self.HEADER_SIZE:self.packet_size - self.FOOTER_SIZE])\n' + else: + yield f' result.valid = True\n' + yield f' result.msg_id = self.msg_id\n' + yield f' result.msg_len = self.packet_size - self.OVERHEAD\n' + yield f' result.msg_data = bytes(self.buffer[self.HEADER_SIZE:])\n' + + reset_state = 'LOOKING_FOR_START1' if len(fmt.start_bytes) > 1 else ('LOOKING_FOR_START' if fmt.start_bytes else 'GETTING_MSG_ID') + yield f' self.state = {class_name}ParserState.{reset_state}\n' + yield '\n' + yield f' return result\n\n' + + # encode method + yield f' def encode(self, msg_id: int, msg: bytes) -> bytes:\n' + yield f' """\n' + yield f' Encode a message with {name} format\n' + yield f' \n' + yield f' Args:\n' + yield f' msg_id: Message ID\n' + yield f' msg: Message data bytes\n' + yield f' \n' + yield f' Returns:\n' + yield f' Encoded frame as bytes\n' + yield f' """\n' + yield f' output = []\n' + + # Write header + for i, (sb_name, sb_value) in enumerate(fmt.start_bytes): + const_name = f'self.START_BYTE{i + 1}' if len(fmt.start_bytes) > 1 else 'self.START_BYTE' + yield f' output.append({const_name})\n' + yield f' output.append(msg_id)\n' + + if fmt.has_length: + if fmt.length_bytes == 1: + yield f' output.append(len(msg) & 0xFF)\n' + else: + yield f' output.append(len(msg) & 0xFF)\n' + yield f' output.append((len(msg) >> 8) & 0xFF)\n' + + yield f' output.extend(msg)\n' + + if fmt.has_crc: + crc_start = len(fmt.start_bytes) + crc_len = 'len(msg) + 1' + if fmt.has_length: + crc_len += f' + {fmt.length_bytes}' + yield f' ck = fletcher_checksum(output, {crc_start}, {crc_start} + {crc_len})\n' + yield f' output.append(ck[0])\n' + yield f' output.append(ck[1])\n' + + yield f' return bytes(output)\n\n' + + # encode_msg method (for message objects with msg_id and pack()) + yield f' def encode_msg(self, msg) -> bytes:\n' + yield f' """Encode a message object (must have msg_id and pack() method)"""\n' + yield f' return self.encode(msg.msg_id, msg.pack())\n\n' + + # validate_packet method + yield f' @staticmethod\n' + yield f' def validate_packet(buffer: Union[bytes, List[int]]) -> FrameMsgInfo:\n' + yield f' """\n' + yield f' Validate a complete {name} packet in a buffer\n' + yield f' \n' + yield f' Args:\n' + yield f' buffer: Buffer containing the complete packet\n' + yield f' \n' + yield f' Returns:\n' + yield f' FrameMsgInfo with valid=True if packet is valid\n' + yield f' """\n' + yield f' result = FrameMsgInfo()\n\n' + yield f' if len(buffer) < {class_name}.OVERHEAD:\n' + yield f' return result\n\n' + + # Check start bytes + for i, (sb_name, sb_value) in enumerate(fmt.start_bytes): + const_name = f'{class_name}.START_BYTE{i + 1}' if len(fmt.start_bytes) > 1 else f'{class_name}.START_BYTE' + yield f' if buffer[{i}] != {const_name}:\n' + yield f' return result\n' + + yield f'\n msg_length = len(buffer) - {class_name}.OVERHEAD\n\n' + + if fmt.has_crc: + crc_start = len(fmt.start_bytes) + crc_len = 'msg_length + 1' + if fmt.has_length: + crc_len += f' + {fmt.length_bytes}' + yield f' # Validate checksum\n' + yield f' ck = fletcher_checksum(buffer, {crc_start}, {crc_start} + {crc_len})\n' + yield f' if ck[0] == buffer[-2] and ck[1] == buffer[-1]:\n' + yield f' result.valid = True\n' + yield f' result.msg_id = buffer[{len(fmt.start_bytes)}]\n' + yield f' result.msg_len = msg_length\n' + yield f' result.msg_data = bytes(buffer[{class_name}.HEADER_SIZE:len(buffer) - {class_name}.FOOTER_SIZE])\n' + else: + yield f' result.valid = True\n' + yield f' result.msg_id = buffer[{len(fmt.start_bytes)}]\n' + yield f' result.msg_len = msg_length\n' + yield f' result.msg_data = bytes(buffer[{class_name}.HEADER_SIZE:])\n' + + yield f'\n return result\n' + + +def generate_py_frame_parsers(formats): + """Generate Python frame parser code from frame format definitions""" + return ''.join(FrameParserPyGen.generate(formats)) diff --git a/src/struct_frame/frame_parser_ts_gen.py b/src/struct_frame/frame_parser_ts_gen.py new file mode 100644 index 00000000..573c27eb --- /dev/null +++ b/src/struct_frame/frame_parser_ts_gen.py @@ -0,0 +1,721 @@ +#!/usr/bin/env python3 +# kate: replace-tabs on; indent-width 4; + +""" +TypeScript Frame Parser Generator + +Generates TypeScript code for frame parsers based on frame format definitions. +""" + +from struct_frame import version +import time + + +def camel_to_snake(name): + """Convert CamelCase to snake_case""" + import re + name = re.sub('(.)([A-Z][a-z]+)', r'\1_\2', name) + name = re.sub('([a-z0-9])([A-Z])', r'\1_\2', name) + return name.lower() + + +def camel_to_upper_snake(name): + """Convert CamelCase to UPPER_SNAKE_CASE""" + return camel_to_snake(name).upper() + + +class FrameParserTsGen: + """Generates TypeScript code for frame parsers""" + + @staticmethod + def generate(formats): + """Generate the complete frame parser TypeScript file""" + yield '// Automatically generated frame parser\n' + yield f'// Generated by {version} at {time.asctime()}.\n\n' + + # Generate frame format enum + yield '// Frame format type enumeration\n' + yield 'export enum FrameFormatType {\n' + for i, fmt in enumerate(formats): + enum_name = camel_to_upper_snake(fmt.name) + yield f' {enum_name} = {i},\n' + yield '}\n\n' + + # Generate checksum function + yield '''// Fletcher-16 checksum calculation +export function fletcher_checksum(buffer: Uint8Array | number[], start: number = 0, end?: number): [number, number] { + if (end === undefined) { + end = buffer.length; + } + + let byte1 = 0; + let byte2 = 0; + + for (let i = start; i < end; i++) { + byte1 = (byte1 + buffer[i]) % 256; + byte2 = (byte2 + byte1) % 256; + } + + return [byte1, byte2]; +} + +''' + + # Generate common result interface + yield '''// Parse result interface +export interface FrameMsgInfo { + valid: boolean; + msg_id: number; + msg_len: number; + msg_data: Uint8Array; +} + +// Create default FrameMsgInfo +export function createFrameMsgInfo(): FrameMsgInfo { + return { + valid: false, + msg_id: 0, + msg_len: 0, + msg_data: new Uint8Array(0) + }; +} + +''' + + # Generate individual frame format classes + for fmt in formats: + yield from FrameParserTsGen.generate_format(fmt) + yield '\n\n' + + @staticmethod + def generate_format(fmt): + """Generate TypeScript code for a single frame format""" + name = fmt.name + class_name = name + PREFIX = camel_to_upper_snake(name) + + yield f'// =============================================================================\n' + yield f'// {name} Frame Format\n' + yield f'// =============================================================================\n\n' + + # Generate parser state enum + yield f'export enum {class_name}ParserState {{\n' + state_idx = 0 + if fmt.start_bytes: + for i in range(len(fmt.start_bytes)): + state_name = f'LOOKING_FOR_START{i + 1}' if len(fmt.start_bytes) > 1 else 'LOOKING_FOR_START' + yield f' {state_name} = {state_idx},\n' + state_idx += 1 + yield f' GETTING_MSG_ID = {state_idx},\n' + state_idx += 1 + if fmt.has_length: + yield f' GETTING_LENGTH = {state_idx},\n' + state_idx += 1 + yield f' GETTING_PAYLOAD = {state_idx}\n' + yield '}\n\n' + + # Generate frame format class + yield f'/**\n' + yield f' * {name} - Frame format parser and encoder\n' + yield f' * \n' + yield f' * Format: ' + parts = [] + for sb_name, sb_value in fmt.start_bytes: + parts.append(f'[{sb_name.upper()}=0x{sb_value:02X}]') + parts.append('[MSG_ID]') + if fmt.has_length: + parts.append(f'[LEN{"16" if fmt.length_bytes == 2 else ""}]') + parts.append('[MSG...]') + if fmt.has_crc: + parts.append('[CRC1] [CRC2]') + yield ' '.join(parts) + yield '\n */\n' + yield f'export class {class_name} {{\n' + + # Constants + for i, (sb_name, sb_value) in enumerate(fmt.start_bytes): + const_name = f'START_BYTE{i + 1}' if len(fmt.start_bytes) > 1 else 'START_BYTE' + yield f' static readonly {const_name} = 0x{sb_value:02X};\n' + yield f' static readonly HEADER_SIZE = {fmt.header_size};\n' + yield f' static readonly FOOTER_SIZE = {fmt.footer_size};\n' + yield f' static readonly OVERHEAD = {fmt.header_size + fmt.footer_size};\n' + if fmt.has_length: + yield f' static readonly LENGTH_BYTES = {fmt.length_bytes};\n' + yield '\n' + + # Instance properties + yield f' private state: {class_name}ParserState;\n' + yield f' private buffer: number[];\n' + yield f' private packet_size: number;\n' + yield f' private msg_id: number;\n' + if fmt.has_length: + yield f' private msg_length: number;\n' + if fmt.length_bytes == 2: + yield f' private length_lo: number;\n' + yield f' private get_msg_length?: (msg_id: number) => number | undefined;\n\n' + + # Constructor + yield f' /**\n' + yield f' * Create a new {name} parser\n' + yield f' * @param get_msg_length Callback to get message length from msg_id (required for non-length frames)\n' + yield f' */\n' + yield f' constructor(get_msg_length?: (msg_id: number) => number | undefined) {{\n' + yield f' this.get_msg_length = get_msg_length;\n' + if fmt.start_bytes: + state_name = 'LOOKING_FOR_START1' if len(fmt.start_bytes) > 1 else 'LOOKING_FOR_START' + yield f' this.state = {class_name}ParserState.{state_name};\n' + else: + yield f' this.state = {class_name}ParserState.GETTING_MSG_ID;\n' + yield f' this.buffer = [];\n' + yield f' this.packet_size = 0;\n' + yield f' this.msg_id = 0;\n' + if fmt.has_length: + yield f' this.msg_length = 0;\n' + if fmt.length_bytes == 2: + yield f' this.length_lo = 0;\n' + yield f' }}\n\n' + + # Reset method + yield f' /** Reset parser state */\n' + yield f' reset(): void {{\n' + if fmt.start_bytes: + state_name = 'LOOKING_FOR_START1' if len(fmt.start_bytes) > 1 else 'LOOKING_FOR_START' + yield f' this.state = {class_name}ParserState.{state_name};\n' + else: + yield f' this.state = {class_name}ParserState.GETTING_MSG_ID;\n' + yield f' this.buffer = [];\n' + yield f' this.packet_size = 0;\n' + yield f' this.msg_id = 0;\n' + if fmt.has_length: + yield f' this.msg_length = 0;\n' + yield f' }}\n\n' + + # parse_byte method + yield f' /**\n' + yield f' * Parse a single byte\n' + yield f' * @param byte The byte to parse\n' + yield f' * @returns FrameMsgInfo with valid=true when a complete valid message is received\n' + yield f' */\n' + yield f' parse_byte(byte: number): FrameMsgInfo {{\n' + yield f' const result = createFrameMsgInfo();\n\n' + yield f' switch (this.state) {{\n' + + # State machine cases + if fmt.start_bytes: + for i, (sb_name, sb_value) in enumerate(fmt.start_bytes): + state_name = f'LOOKING_FOR_START{i + 1}' if len(fmt.start_bytes) > 1 else 'LOOKING_FOR_START' + const_name = f'{class_name}.START_BYTE{i + 1}' if len(fmt.start_bytes) > 1 else f'{class_name}.START_BYTE' + + yield f' case {class_name}ParserState.{state_name}:\n' + yield f' if (byte === {const_name}) {{\n' + + if i == 0: + yield f' this.buffer = [byte];\n' + else: + yield f' this.buffer.push(byte);\n' + + if i + 1 < len(fmt.start_bytes): + next_state = f'LOOKING_FOR_START{i + 2}' + else: + next_state = 'GETTING_MSG_ID' + yield f' this.state = {class_name}ParserState.{next_state};\n' + yield f' }}' + + # Handle seeing start byte 1 while looking for later bytes + # This code only runs when len(fmt.start_bytes) > 1 (i.e., i > 0 is only possible with 2+ start bytes) + if i > 0: + yield f' else if (byte === {class_name}.START_BYTE1) {{\n' + yield f' this.buffer = [byte];\n' + yield f' this.state = {class_name}ParserState.LOOKING_FOR_START2;\n' + yield f' }} else {{\n' + yield f' this.state = {class_name}ParserState.LOOKING_FOR_START1;\n' + yield f' }}\n' + else: + yield '\n' + yield f' break;\n\n' + + # GETTING_MSG_ID state + yield f' case {class_name}ParserState.GETTING_MSG_ID:\n' + yield f' this.buffer.push(byte);\n' + yield f' this.msg_id = byte;\n' + + if fmt.has_length: + yield f' this.state = {class_name}ParserState.GETTING_LENGTH;\n' + else: + yield f' if (this.get_msg_length) {{\n' + yield f' const msg_length = this.get_msg_length(byte);\n' + yield f' if (msg_length !== undefined) {{\n' + yield f' this.packet_size = {class_name}.OVERHEAD + msg_length;\n' + yield f' this.state = {class_name}ParserState.GETTING_PAYLOAD;\n' + yield f' }} else {{\n' + reset_state = 'LOOKING_FOR_START1' if len(fmt.start_bytes) > 1 else ('LOOKING_FOR_START' if fmt.start_bytes else 'GETTING_MSG_ID') + yield f' this.state = {class_name}ParserState.{reset_state};\n' + yield f' }}\n' + yield f' }} else {{\n' + yield f' this.state = {class_name}ParserState.{reset_state};\n' + yield f' }}\n' + yield f' break;\n\n' + + # GETTING_LENGTH state (if applicable) + if fmt.has_length: + yield f' case {class_name}ParserState.GETTING_LENGTH:\n' + yield f' this.buffer.push(byte);\n' + if fmt.length_bytes == 1: + yield f' this.msg_length = byte;\n' + yield f' this.packet_size = {class_name}.OVERHEAD + this.msg_length;\n' + yield f' this.state = {class_name}ParserState.GETTING_PAYLOAD;\n' + else: + yield f' if (this.buffer.length === {len(fmt.start_bytes) + 2}) {{\n' + yield f' this.length_lo = byte;\n' + yield f' }} else {{\n' + yield f' this.msg_length = this.length_lo | (byte << 8);\n' + yield f' this.packet_size = {class_name}.OVERHEAD + this.msg_length;\n' + yield f' this.state = {class_name}ParserState.GETTING_PAYLOAD;\n' + yield f' }}\n' + yield f' break;\n\n' + + # GETTING_PAYLOAD state + yield f' case {class_name}ParserState.GETTING_PAYLOAD:\n' + yield f' this.buffer.push(byte);\n\n' + yield f' if (this.buffer.length >= this.packet_size) {{\n' + + if fmt.has_crc: + yield f' // Validate checksum\n' + yield f' const msg_length = this.packet_size - {class_name}.OVERHEAD;\n' + crc_start = len(fmt.start_bytes) + crc_len = f'msg_length + 1' + if fmt.has_length: + crc_len += f' + {fmt.length_bytes}' + yield f' const ck = fletcher_checksum(this.buffer, {crc_start}, {crc_start} + {crc_len});\n' + yield f' if (ck[0] === this.buffer[this.buffer.length - 2] && ck[1] === this.buffer[this.buffer.length - 1]) {{\n' + yield f' result.valid = true;\n' + yield f' result.msg_id = this.msg_id;\n' + yield f' result.msg_len = msg_length;\n' + yield f' result.msg_data = new Uint8Array(this.buffer.slice({class_name}.HEADER_SIZE, this.packet_size - {class_name}.FOOTER_SIZE));\n' + yield f' }}\n' + else: + yield f' result.valid = true;\n' + yield f' result.msg_id = this.msg_id;\n' + yield f' result.msg_len = this.packet_size - {class_name}.OVERHEAD;\n' + yield f' result.msg_data = new Uint8Array(this.buffer.slice({class_name}.HEADER_SIZE));\n' + + reset_state = 'LOOKING_FOR_START1' if len(fmt.start_bytes) > 1 else ('LOOKING_FOR_START' if fmt.start_bytes else 'GETTING_MSG_ID') + yield f' this.state = {class_name}ParserState.{reset_state};\n' + yield f' }}\n' + yield f' break;\n' + + yield f' }}\n\n' + yield f' return result;\n' + yield f' }}\n\n' + + # encode method + yield f' /**\n' + yield f' * Encode a message with {name} format\n' + yield f' * @param msg_id Message ID\n' + yield f' * @param msg Message data\n' + yield f' * @returns Encoded frame as Uint8Array\n' + yield f' */\n' + yield f' static encode(msg_id: number, msg: Uint8Array | number[]): Uint8Array {{\n' + yield f' const output: number[] = [];\n' + + # Write header + for i, (sb_name, sb_value) in enumerate(fmt.start_bytes): + const_name = f'{class_name}.START_BYTE{i + 1}' if len(fmt.start_bytes) > 1 else f'{class_name}.START_BYTE' + yield f' output.push({const_name});\n' + yield f' output.push(msg_id);\n' + + if fmt.has_length: + if fmt.length_bytes == 1: + yield f' output.push(msg.length & 0xFF);\n' + else: + yield f' output.push(msg.length & 0xFF);\n' + yield f' output.push((msg.length >> 8) & 0xFF);\n' + + yield f' for (let i = 0; i < msg.length; i++) {{\n' + yield f' output.push(msg[i]);\n' + yield f' }}\n' + + if fmt.has_crc: + crc_start = len(fmt.start_bytes) + crc_len = 'msg.length + 1' + if fmt.has_length: + crc_len += f' + {fmt.length_bytes}' + yield f' const ck = fletcher_checksum(output, {crc_start}, {crc_start} + {crc_len});\n' + yield f' output.push(ck[0]);\n' + yield f' output.push(ck[1]);\n' + + yield f' return new Uint8Array(output);\n' + yield f' }}\n\n' + + # validate_packet static method + yield f' /**\n' + yield f' * Validate a complete {name} packet in a buffer\n' + yield f' * @param buffer Buffer containing the complete packet\n' + yield f' * @returns FrameMsgInfo with valid=true if packet is valid\n' + yield f' */\n' + yield f' static validate_packet(buffer: Uint8Array | number[]): FrameMsgInfo {{\n' + yield f' const result = createFrameMsgInfo();\n\n' + yield f' if (buffer.length < {class_name}.OVERHEAD) {{\n' + yield f' return result;\n' + yield f' }}\n\n' + + # Check start bytes + for i, (sb_name, sb_value) in enumerate(fmt.start_bytes): + const_name = f'{class_name}.START_BYTE{i + 1}' if len(fmt.start_bytes) > 1 else f'{class_name}.START_BYTE' + yield f' if (buffer[{i}] !== {const_name}) {{\n' + yield f' return result;\n' + yield f' }}\n' + + yield f'\n const msg_length = buffer.length - {class_name}.OVERHEAD;\n\n' + + if fmt.has_crc: + crc_start = len(fmt.start_bytes) + crc_len = 'msg_length + 1' + if fmt.has_length: + crc_len += f' + {fmt.length_bytes}' + yield f' // Validate checksum\n' + yield f' const ck = fletcher_checksum(buffer, {crc_start}, {crc_start} + {crc_len});\n' + yield f' if (ck[0] === buffer[buffer.length - 2] && ck[1] === buffer[buffer.length - 1]) {{\n' + yield f' result.valid = true;\n' + yield f' result.msg_id = buffer[{len(fmt.start_bytes)}];\n' + yield f' result.msg_len = msg_length;\n' + yield f' result.msg_data = new Uint8Array(Array.prototype.slice.call(buffer, {class_name}.HEADER_SIZE, buffer.length - {class_name}.FOOTER_SIZE));\n' + yield f' }}\n' + else: + yield f' result.valid = true;\n' + yield f' result.msg_id = buffer[{len(fmt.start_bytes)}];\n' + yield f' result.msg_len = msg_length;\n' + yield f' result.msg_data = new Uint8Array(Array.prototype.slice.call(buffer, {class_name}.HEADER_SIZE));\n' + + yield f'\n return result;\n' + yield f' }}\n' + yield f'}}\n' + + +def generate_ts_frame_parsers(formats): + """Generate TypeScript frame parser code from frame format definitions""" + return ''.join(FrameParserTsGen.generate(formats)) + + +class FrameParserJsGen: + """Generates JavaScript code for frame parsers (no TypeScript annotations)""" + + @staticmethod + def generate(formats): + """Generate the complete frame parser JavaScript file""" + yield '// Automatically generated frame parser\n' + yield f'// Generated by {version} at {time.asctime()}.\n\n' + + # Generate frame format enum as object + yield '// Frame format type enumeration\n' + yield 'const FrameFormatType = {\n' + for i, fmt in enumerate(formats): + enum_name = camel_to_upper_snake(fmt.name) + yield f' {enum_name}: {i},\n' + yield '};\n\n' + + # Generate checksum function + yield '''// Fletcher-16 checksum calculation +function fletcher_checksum(buffer, start = 0, end = undefined) { + if (end === undefined) { + end = buffer.length; + } + + let byte1 = 0; + let byte2 = 0; + + for (let i = start; i < end; i++) { + byte1 = (byte1 + buffer[i]) % 256; + byte2 = (byte2 + byte1) % 256; + } + + return [byte1, byte2]; +} + +// Create default FrameMsgInfo +function createFrameMsgInfo() { + return { + valid: false, + msg_id: 0, + msg_len: 0, + msg_data: new Uint8Array(0) + }; +} + +''' + + # Generate individual frame format classes + for fmt in formats: + yield from FrameParserJsGen.generate_format(fmt) + yield '\n\n' + + # Generate exports + yield '// Exports\n' + yield 'module.exports = {\n' + yield ' FrameFormatType,\n' + yield ' fletcher_checksum,\n' + yield ' createFrameMsgInfo,\n' + for fmt in formats: + yield f' {fmt.name},\n' + yield f' {fmt.name}ParserState,\n' + yield '};\n' + + @staticmethod + def generate_format(fmt): + """Generate JavaScript code for a single frame format (same as TS but without types)""" + name = fmt.name + class_name = name + PREFIX = camel_to_upper_snake(name) + + yield f'// =============================================================================\n' + yield f'// {name} Frame Format\n' + yield f'// =============================================================================\n\n' + + # Generate parser state enum as object + yield f'const {class_name}ParserState = {{\n' + state_idx = 0 + if fmt.start_bytes: + for i in range(len(fmt.start_bytes)): + state_name = f'LOOKING_FOR_START{i + 1}' if len(fmt.start_bytes) > 1 else 'LOOKING_FOR_START' + yield f' {state_name}: {state_idx},\n' + state_idx += 1 + yield f' GETTING_MSG_ID: {state_idx},\n' + state_idx += 1 + if fmt.has_length: + yield f' GETTING_LENGTH: {state_idx},\n' + state_idx += 1 + yield f' GETTING_PAYLOAD: {state_idx}\n' + yield '};\n\n' + + # Generate frame format class + yield f'/**\n' + yield f' * {name} - Frame format parser and encoder\n' + yield f' */\n' + yield f'class {class_name} {{\n' + + # Static constants + for i, (sb_name, sb_value) in enumerate(fmt.start_bytes): + const_name = f'START_BYTE{i + 1}' if len(fmt.start_bytes) > 1 else 'START_BYTE' + yield f' static {const_name} = 0x{sb_value:02X};\n' + yield f' static HEADER_SIZE = {fmt.header_size};\n' + yield f' static FOOTER_SIZE = {fmt.footer_size};\n' + yield f' static OVERHEAD = {fmt.header_size + fmt.footer_size};\n' + if fmt.has_length: + yield f' static LENGTH_BYTES = {fmt.length_bytes};\n' + yield '\n' + + # Constructor + yield f' constructor(get_msg_length) {{\n' + yield f' this.get_msg_length = get_msg_length;\n' + if fmt.start_bytes: + state_name = 'LOOKING_FOR_START1' if len(fmt.start_bytes) > 1 else 'LOOKING_FOR_START' + yield f' this.state = {class_name}ParserState.{state_name};\n' + else: + yield f' this.state = {class_name}ParserState.GETTING_MSG_ID;\n' + yield f' this.buffer = [];\n' + yield f' this.packet_size = 0;\n' + yield f' this.msg_id = 0;\n' + if fmt.has_length: + yield f' this.msg_length = 0;\n' + if fmt.length_bytes == 2: + yield f' this.length_lo = 0;\n' + yield f' }}\n\n' + + # Reset method + yield f' reset() {{\n' + if fmt.start_bytes: + state_name = 'LOOKING_FOR_START1' if len(fmt.start_bytes) > 1 else 'LOOKING_FOR_START' + yield f' this.state = {class_name}ParserState.{state_name};\n' + else: + yield f' this.state = {class_name}ParserState.GETTING_MSG_ID;\n' + yield f' this.buffer = [];\n' + yield f' this.packet_size = 0;\n' + yield f' this.msg_id = 0;\n' + if fmt.has_length: + yield f' this.msg_length = 0;\n' + yield f' }}\n\n' + + # parse_byte method (simplified version - same logic as TypeScript) + yield f' parse_byte(byte) {{\n' + yield f' const result = createFrameMsgInfo();\n\n' + yield f' switch (this.state) {{\n' + + # State machine cases (same as TypeScript but without type annotations) + if fmt.start_bytes: + for i, (sb_name, sb_value) in enumerate(fmt.start_bytes): + state_name = f'LOOKING_FOR_START{i + 1}' if len(fmt.start_bytes) > 1 else 'LOOKING_FOR_START' + const_name = f'{class_name}.START_BYTE{i + 1}' if len(fmt.start_bytes) > 1 else f'{class_name}.START_BYTE' + + yield f' case {class_name}ParserState.{state_name}:\n' + yield f' if (byte === {const_name}) {{\n' + + if i == 0: + yield f' this.buffer = [byte];\n' + else: + yield f' this.buffer.push(byte);\n' + + if i + 1 < len(fmt.start_bytes): + next_state = f'LOOKING_FOR_START{i + 2}' + else: + next_state = 'GETTING_MSG_ID' + yield f' this.state = {class_name}ParserState.{next_state};\n' + yield f' }}' + + # This code only runs when len(fmt.start_bytes) > 1 (i.e., i > 0 is only possible with 2+ start bytes) + if i > 0: + yield f' else if (byte === {class_name}.START_BYTE1) {{\n' + yield f' this.buffer = [byte];\n' + yield f' this.state = {class_name}ParserState.LOOKING_FOR_START2;\n' + yield f' }} else {{\n' + yield f' this.state = {class_name}ParserState.LOOKING_FOR_START1;\n' + yield f' }}\n' + else: + yield '\n' + yield f' break;\n\n' + + # GETTING_MSG_ID state + yield f' case {class_name}ParserState.GETTING_MSG_ID:\n' + yield f' this.buffer.push(byte);\n' + yield f' this.msg_id = byte;\n' + + if fmt.has_length: + yield f' this.state = {class_name}ParserState.GETTING_LENGTH;\n' + else: + yield f' if (this.get_msg_length) {{\n' + yield f' const msg_length = this.get_msg_length(byte);\n' + yield f' if (msg_length !== undefined) {{\n' + yield f' this.packet_size = {class_name}.OVERHEAD + msg_length;\n' + yield f' this.state = {class_name}ParserState.GETTING_PAYLOAD;\n' + yield f' }} else {{\n' + reset_state = 'LOOKING_FOR_START1' if len(fmt.start_bytes) > 1 else ('LOOKING_FOR_START' if fmt.start_bytes else 'GETTING_MSG_ID') + yield f' this.state = {class_name}ParserState.{reset_state};\n' + yield f' }}\n' + yield f' }} else {{\n' + yield f' this.state = {class_name}ParserState.{reset_state};\n' + yield f' }}\n' + yield f' break;\n\n' + + # GETTING_LENGTH state (if applicable) + if fmt.has_length: + yield f' case {class_name}ParserState.GETTING_LENGTH:\n' + yield f' this.buffer.push(byte);\n' + if fmt.length_bytes == 1: + yield f' this.msg_length = byte;\n' + yield f' this.packet_size = {class_name}.OVERHEAD + this.msg_length;\n' + yield f' this.state = {class_name}ParserState.GETTING_PAYLOAD;\n' + else: + yield f' if (this.buffer.length === {len(fmt.start_bytes) + 2}) {{\n' + yield f' this.length_lo = byte;\n' + yield f' }} else {{\n' + yield f' this.msg_length = this.length_lo | (byte << 8);\n' + yield f' this.packet_size = {class_name}.OVERHEAD + this.msg_length;\n' + yield f' this.state = {class_name}ParserState.GETTING_PAYLOAD;\n' + yield f' }}\n' + yield f' break;\n\n' + + # GETTING_PAYLOAD state + yield f' case {class_name}ParserState.GETTING_PAYLOAD:\n' + yield f' this.buffer.push(byte);\n\n' + yield f' if (this.buffer.length >= this.packet_size) {{\n' + + if fmt.has_crc: + yield f' const msg_length = this.packet_size - {class_name}.OVERHEAD;\n' + crc_start = len(fmt.start_bytes) + crc_len = f'msg_length + 1' + if fmt.has_length: + crc_len += f' + {fmt.length_bytes}' + yield f' const ck = fletcher_checksum(this.buffer, {crc_start}, {crc_start} + {crc_len});\n' + yield f' if (ck[0] === this.buffer[this.buffer.length - 2] && ck[1] === this.buffer[this.buffer.length - 1]) {{\n' + yield f' result.valid = true;\n' + yield f' result.msg_id = this.msg_id;\n' + yield f' result.msg_len = msg_length;\n' + yield f' result.msg_data = new Uint8Array(this.buffer.slice({class_name}.HEADER_SIZE, this.packet_size - {class_name}.FOOTER_SIZE));\n' + yield f' }}\n' + else: + yield f' result.valid = true;\n' + yield f' result.msg_id = this.msg_id;\n' + yield f' result.msg_len = this.packet_size - {class_name}.OVERHEAD;\n' + yield f' result.msg_data = new Uint8Array(this.buffer.slice({class_name}.HEADER_SIZE));\n' + + reset_state = 'LOOKING_FOR_START1' if len(fmt.start_bytes) > 1 else ('LOOKING_FOR_START' if fmt.start_bytes else 'GETTING_MSG_ID') + yield f' this.state = {class_name}ParserState.{reset_state};\n' + yield f' }}\n' + yield f' break;\n' + + yield f' }}\n\n' + yield f' return result;\n' + yield f' }}\n\n' + + # Static encode method + yield f' static encode(msg_id, msg) {{\n' + yield f' const output = [];\n' + + for i, (sb_name, sb_value) in enumerate(fmt.start_bytes): + const_name = f'{class_name}.START_BYTE{i + 1}' if len(fmt.start_bytes) > 1 else f'{class_name}.START_BYTE' + yield f' output.push({const_name});\n' + yield f' output.push(msg_id);\n' + + if fmt.has_length: + if fmt.length_bytes == 1: + yield f' output.push(msg.length & 0xFF);\n' + else: + yield f' output.push(msg.length & 0xFF);\n' + yield f' output.push((msg.length >> 8) & 0xFF);\n' + + yield f' for (let i = 0; i < msg.length; i++) {{\n' + yield f' output.push(msg[i]);\n' + yield f' }}\n' + + if fmt.has_crc: + crc_start = len(fmt.start_bytes) + crc_len = 'msg.length + 1' + if fmt.has_length: + crc_len += f' + {fmt.length_bytes}' + yield f' const ck = fletcher_checksum(output, {crc_start}, {crc_start} + {crc_len});\n' + yield f' output.push(ck[0]);\n' + yield f' output.push(ck[1]);\n' + + yield f' return new Uint8Array(output);\n' + yield f' }}\n\n' + + # Static validate_packet method + yield f' static validate_packet(buffer) {{\n' + yield f' const result = createFrameMsgInfo();\n\n' + yield f' if (buffer.length < {class_name}.OVERHEAD) {{\n' + yield f' return result;\n' + yield f' }}\n\n' + + for i, (sb_name, sb_value) in enumerate(fmt.start_bytes): + const_name = f'{class_name}.START_BYTE{i + 1}' if len(fmt.start_bytes) > 1 else f'{class_name}.START_BYTE' + yield f' if (buffer[{i}] !== {const_name}) {{\n' + yield f' return result;\n' + yield f' }}\n' + + yield f'\n const msg_length = buffer.length - {class_name}.OVERHEAD;\n\n' + + if fmt.has_crc: + crc_start = len(fmt.start_bytes) + crc_len = 'msg_length + 1' + if fmt.has_length: + crc_len += f' + {fmt.length_bytes}' + yield f' const ck = fletcher_checksum(buffer, {crc_start}, {crc_start} + {crc_len});\n' + yield f' if (ck[0] === buffer[buffer.length - 2] && ck[1] === buffer[buffer.length - 1]) {{\n' + yield f' result.valid = true;\n' + yield f' result.msg_id = buffer[{len(fmt.start_bytes)}];\n' + yield f' result.msg_len = msg_length;\n' + yield f' result.msg_data = new Uint8Array(Array.prototype.slice.call(buffer, {class_name}.HEADER_SIZE, buffer.length - {class_name}.FOOTER_SIZE));\n' + yield f' }}\n' + else: + yield f' result.valid = true;\n' + yield f' result.msg_id = buffer[{len(fmt.start_bytes)}];\n' + yield f' result.msg_len = msg_length;\n' + yield f' result.msg_data = new Uint8Array(Array.prototype.slice.call(buffer, {class_name}.HEADER_SIZE));\n' + + yield f'\n return result;\n' + yield f' }}\n' + yield f'}}\n' + + +def generate_js_frame_parsers(formats): + """Generate JavaScript frame parser code from frame format definitions""" + return ''.join(FrameParserJsGen.generate(formats)) diff --git a/src/struct_frame/generate.py b/src/struct_frame/generate.py index 2b188062..e4bfbf0e 100644 --- a/src/struct_frame/generate.py +++ b/src/struct_frame/generate.py @@ -464,6 +464,8 @@ def __str__(self): parser.add_argument('--build_gql', action='store_true') parser.add_argument('--gql_path', nargs=1, type=str, default=['generated/gql/']) +parser.add_argument('--frame_formats', nargs=1, type=str, + help='Proto file containing frame format definitions to generate frame parsers') def parseFile(filename): @@ -565,6 +567,41 @@ def generateCppFileStrings(path): return out +def generateFrameParserFiles(frame_formats_file, c_path, ts_path, js_path, py_path, cpp_path, + build_c, build_ts, build_js, build_py, build_cpp): + """Generate frame parser files from frame format definitions""" + from struct_frame.frame_format import parse_frame_formats + from struct_frame.frame_parser_c_gen import generate_c_frame_parsers + from struct_frame.frame_parser_py_gen import generate_py_frame_parsers + from struct_frame.frame_parser_ts_gen import generate_ts_frame_parsers, generate_js_frame_parsers + from struct_frame.frame_parser_cpp_gen import generate_cpp_frame_parsers + + formats = parse_frame_formats(frame_formats_file) + files = {} + + if build_c: + name = os.path.join(c_path, "frame_parsers_gen.h") + files[name] = generate_c_frame_parsers(formats) + + if build_ts: + name = os.path.join(ts_path, "frame_parsers_gen.ts") + files[name] = generate_ts_frame_parsers(formats) + + if build_js: + name = os.path.join(js_path, "frame_parsers_gen.js") + files[name] = generate_js_frame_parsers(formats) + + if build_py: + name = os.path.join(py_path, "frame_parsers_gen.py") + files[name] = generate_py_frame_parsers(formats) + + if build_cpp: + name = os.path.join(cpp_path, "frame_parsers_gen.hpp") + files[name] = generate_cpp_frame_parsers(formats) + + return files + + def main(): args = parser.parse_args() parseFile(args.filename) @@ -618,6 +655,17 @@ def main(): data = ''.join(FileGqlGen.generate(value)) files[name] = data + # Generate frame parsers if frame_formats proto is provided + if args.frame_formats: + frame_parser_files = generateFrameParserFiles( + args.frame_formats[0], + args.c_path[0], args.ts_path[0], args.js_path[0], + args.py_path[0], args.cpp_path[0], + args.build_c, args.build_ts, args.build_js, + args.build_py, args.build_cpp + ) + files.update(frame_parser_files) + for filename, filedata in files.items(): dirname = os.path.dirname(filename) if dirname and not os.path.exists(dirname): @@ -628,25 +676,75 @@ def main(): dir_path = os.path.dirname(os.path.realpath(__file__)) - if (args.build_c): - shutil.copytree(os.path.join(dir_path, "boilerplate/c"), - args.c_path[0], dirs_exist_ok=True) - - if (args.build_ts): - shutil.copytree(os.path.join(dir_path, "boilerplate/ts"), - args.ts_path[0], dirs_exist_ok=True) - - if (args.build_js): - shutil.copytree(os.path.join(dir_path, "boilerplate/js"), - args.js_path[0], dirs_exist_ok=True) - - if (args.build_py): - shutil.copytree(os.path.join(dir_path, "boilerplate/py"), - args.py_path[0], dirs_exist_ok=True) - - if (args.build_cpp): - shutil.copytree(os.path.join(dir_path, "boilerplate/cpp"), - args.cpp_path[0], dirs_exist_ok=True) + # When --frame_formats is provided, the frame parser boilerplate files are + # replaced by the generated frame parsers, so we only copy utility files. + # Otherwise, copy all boilerplate files including hand-coded frame parsers. + if args.frame_formats: + # Frame parser files to exclude when generating frame parsers + frame_parser_files = { + 'c': ['basic_frame.h', 'basic_frame_with_len.h', 'struct_frame_parser.h'], + 'cpp': ['basic_frame.hpp', 'basic_frame_with_len.hpp', 'struct_frame_parser.hpp'], + 'ts': ['struct_frame_parser.ts'], + 'js': ['struct_frame_parser.js'], + 'py': ['struct_frame_parser.py'] + } + + def copy_boilerplate_selective(src_dir, dst_dir, exclude_files): + """Copy boilerplate files excluding frame parser files""" + if not os.path.exists(dst_dir): + os.makedirs(dst_dir) + for item in os.listdir(src_dir): + if item not in exclude_files: + src_path = os.path.join(src_dir, item) + dst_path = os.path.join(dst_dir, item) + if os.path.isfile(src_path): + shutil.copy2(src_path, dst_path) + + if args.build_c: + copy_boilerplate_selective( + os.path.join(dir_path, "boilerplate/c"), + args.c_path[0], frame_parser_files['c']) + + if args.build_ts: + copy_boilerplate_selective( + os.path.join(dir_path, "boilerplate/ts"), + args.ts_path[0], frame_parser_files['ts']) + + if args.build_js: + copy_boilerplate_selective( + os.path.join(dir_path, "boilerplate/js"), + args.js_path[0], frame_parser_files['js']) + + if args.build_py: + copy_boilerplate_selective( + os.path.join(dir_path, "boilerplate/py"), + args.py_path[0], frame_parser_files['py']) + + if args.build_cpp: + copy_boilerplate_selective( + os.path.join(dir_path, "boilerplate/cpp"), + args.cpp_path[0], frame_parser_files['cpp']) + else: + # Copy all boilerplate files (default behavior) + if (args.build_c): + shutil.copytree(os.path.join(dir_path, "boilerplate/c"), + args.c_path[0], dirs_exist_ok=True) + + if (args.build_ts): + shutil.copytree(os.path.join(dir_path, "boilerplate/ts"), + args.ts_path[0], dirs_exist_ok=True) + + if (args.build_js): + shutil.copytree(os.path.join(dir_path, "boilerplate/js"), + args.js_path[0], dirs_exist_ok=True) + + if (args.build_py): + shutil.copytree(os.path.join(dir_path, "boilerplate/py"), + args.py_path[0], dirs_exist_ok=True) + + if (args.build_cpp): + shutil.copytree(os.path.join(dir_path, "boilerplate/cpp"), + args.cpp_path[0], dirs_exist_ok=True) # No boilerplate for GraphQL currently