| #!/usr/bin/env python3 |
| |
| import json |
| |
| |
| class ParseError(Exception): |
| pass |
| |
| |
| magic_prefix = "vl_api_" |
| magic_suffix = "_t" |
| |
| |
| def remove_magic(what): |
| if what.startswith(magic_prefix) and what.endswith(magic_suffix): |
| return what[len(magic_prefix) : -len(magic_suffix)] |
| return what |
| |
| |
| class Field(object): |
| def __init__(self, field_name, field_type, array_len=None, nelem_field=None): |
| self.name = field_name |
| self.type = field_type |
| self.len = array_len |
| self.nelem_field = nelem_field |
| |
| def __str__(self): |
| if self.len is None: |
| return "Field(name: %s, type: %s)" % (self.name, self.type) |
| elif type(self.len) == dict: |
| return "Field(name: %s, type: %s, length: %s)" % ( |
| self.name, |
| self.type, |
| self.len, |
| ) |
| elif self.len > 0: |
| return "Field(name: %s, type: %s, length: %s)" % ( |
| self.name, |
| self.type, |
| self.len, |
| ) |
| else: |
| return "Field(name: %s, type: %s, variable length stored in: %s)" % ( |
| self.name, |
| self.type, |
| self.nelem_field, |
| ) |
| |
| def is_vla(self): |
| return self.nelem_field is not None |
| |
| def has_vla(self): |
| return self.is_vla() or self.type.has_vla() |
| |
| |
| class Alias(Field): |
| pass |
| |
| |
| class Type(object): |
| def __init__(self, name): |
| self.name = name |
| |
| def __str__(self): |
| return self.name |
| |
| |
| class SimpleType(Type): |
| def has_vla(self): |
| return False |
| |
| |
| def get_msg_header_defs(struct_type_class, field_class, json_parser, logger): |
| return [ |
| struct_type_class( |
| [ |
| "msg_header1_t", |
| ["u16", "_vl_msg_id"], |
| ["u32", "context"], |
| ], |
| json_parser, |
| field_class, |
| logger, |
| ), |
| struct_type_class( |
| [ |
| "msg_header2_t", |
| ["u16", "_vl_msg_id"], |
| ["u32", "client_index"], |
| ["u32", "context"], |
| ], |
| json_parser, |
| field_class, |
| logger, |
| ), |
| ] |
| |
| |
| class Struct(object): |
| def __init__(self, name, fields): |
| self.name = name |
| self.fields = fields |
| self.field_names = [n.name for n in self.fields] |
| self.depends = [f.type for f in self.fields] |
| |
| def __str__(self): |
| return "[%s]" % "], [".join([str(f) for f in self.fields]) |
| |
| def has_vla(self): |
| for f in self.fields: |
| if f.has_vla(): |
| return True |
| return False |
| |
| |
| class Enum(SimpleType): |
| def __init__(self, name, value_pairs, enumtype): |
| super(Enum, self).__init__(name) |
| self.type = enumtype |
| self.value_pairs = value_pairs |
| |
| def __str__(self): |
| return "Enum(%s, [%s])" % ( |
| self.name, |
| "], [".join(["%s => %s" % (i, j) for i, j in self.value_pairs]), |
| ) |
| |
| |
| class Union(Type): |
| def __init__(self, name, type_pairs, crc): |
| Type.__init__(self, name) |
| self.crc = crc |
| self.type_pairs = type_pairs |
| self.depends = [t for t, _ in self.type_pairs] |
| |
| def __str__(self): |
| return "Union(%s, [%s])" % ( |
| self.name, |
| "], [".join(["%s %s" % (i, j) for i, j in self.type_pairs]), |
| ) |
| |
| def has_vla(self): |
| return False |
| |
| |
| class Message(object): |
| def __init__(self, logger, definition, json_parser): |
| struct_type_class = json_parser.struct_type_class |
| field_class = json_parser.field_class |
| self.request = None |
| self.logger = logger |
| m = definition |
| logger.debug("Parsing message definition `%s'" % m) |
| name = m[0] |
| self.name = name |
| logger.debug("Message name is `%s'" % name) |
| ignore = True |
| self.header = None |
| self.is_reply = json_parser.is_reply(self.name) |
| self.is_event = json_parser.is_event(self.name) |
| fields = [] |
| for header in get_msg_header_defs( |
| struct_type_class, field_class, json_parser, logger |
| ): |
| logger.debug("Probing header `%s'" % header.name) |
| if header.is_part_of_def(m[1:]): |
| self.header = header |
| logger.debug("Found header `%s'" % header.name) |
| fields.append(field_class(field_name="header", field_type=self.header)) |
| ignore = False |
| break |
| if ignore and not self.is_event and not self.is_reply: |
| raise ParseError( |
| "While parsing message `%s': could not find all " |
| "common header fields" % name |
| ) |
| for field in m[1:]: |
| if isinstance(field, dict) and "crc" in field: |
| self.crc = field["crc"] |
| logger.debug("Found CRC `%s'" % self.crc) |
| continue |
| else: |
| field_type = json_parser.lookup_type_like_id(field[0]) |
| logger.debug("Parsing message field `%s'" % field) |
| l = len(field) |
| if any(type(n) is dict for n in field): |
| l -= 1 |
| if l == 2: |
| if self.header is not None and self.header.has_field(field[1]): |
| continue |
| p = field_class(field_name=field[1], field_type=field_type) |
| elif l == 3: |
| if field[2] == 0 and field[0] != "string": |
| raise ParseError( |
| "While parsing message `%s': variable length " |
| "array `%s' doesn't have reference to member " |
| "containing the actual length" % (name, field[1]) |
| ) |
| if field[0] == "string" and field[2] > 0: |
| field_type = json_parser.lookup_type_like_id("u8") |
| |
| p = field_class( |
| field_name=field[1], field_type=field_type, array_len=field[2] |
| ) |
| elif l == 4: |
| nelem_field = None |
| for f in fields: |
| if f.name == field[3]: |
| nelem_field = f |
| if nelem_field is None: |
| raise ParseError( |
| "While parsing message `%s': couldn't find " |
| "variable length array `%s' member containing " |
| "the actual length `%s'" % (name, field[1], field[3]) |
| ) |
| p = field_class( |
| field_name=field[1], |
| field_type=field_type, |
| array_len=field[2], |
| nelem_field=nelem_field, |
| ) |
| else: |
| raise Exception( |
| "Don't know how to parse message " |
| "definition for message `%s': `%s'" % (m, m[1:]) |
| ) |
| logger.debug("Parsed field `%s'" % p) |
| fields.append(p) |
| self.fields = fields |
| self.depends = [f.type for f in self.fields] |
| logger.debug("Parsed message: %s" % self) |
| |
| def __str__(self): |
| return "Message(%s, [%s], {crc: %s}" % ( |
| self.name, |
| "], [".join([str(f) for f in self.fields]), |
| self.crc, |
| ) |
| |
| |
| class StructType(Type, Struct): |
| def __init__(self, definition, json_parser, field_class, logger): |
| t = definition |
| logger.debug("Parsing struct definition `%s'" % t) |
| name = t[0] |
| fields = [] |
| for field in t[1:]: |
| if len(field) == 1 and "crc" in field: |
| self.crc = field["crc"] |
| continue |
| field_type = json_parser.lookup_type_like_id(field[0]) |
| logger.debug("Parsing type field `%s'" % field) |
| if len(field) == 2: |
| p = field_class(field_name=field[1], field_type=field_type) |
| elif len(field) == 3: |
| if field[2] == 0: |
| raise ParseError( |
| "While parsing type `%s': array `%s' has " |
| "variable length" % (name, field[1]) |
| ) |
| p = field_class( |
| field_name=field[1], field_type=field_type, array_len=field[2] |
| ) |
| elif len(field) == 4: |
| nelem_field = None |
| for f in fields: |
| if f.name == field[3]: |
| nelem_field = f |
| if nelem_field is None: |
| raise ParseError( |
| "While parsing message `%s': couldn't find " |
| "variable length array `%s' member containing " |
| "the actual length `%s'" % (name, field[1], field[3]) |
| ) |
| p = field_class( |
| field_name=field[1], |
| field_type=field_type, |
| array_len=field[2], |
| nelem_field=nelem_field, |
| ) |
| else: |
| raise ParseError( |
| "Don't know how to parse field `%s' of type definition " |
| "for type `%s'" % (field, t) |
| ) |
| fields.append(p) |
| Type.__init__(self, name) |
| Struct.__init__(self, name, fields) |
| |
| def __str__(self): |
| return "StructType(%s, %s)" % (Type.__str__(self), Struct.__str__(self)) |
| |
| def has_field(self, name): |
| return name in self.field_names |
| |
| def is_part_of_def(self, definition): |
| for idx in range(len(self.fields)): |
| field = definition[idx] |
| p = self.fields[idx] |
| if field[1] != p.name: |
| return False |
| if field[0] != p.type.name: |
| raise ParseError( |
| "Unexpected field type `%s' (should be `%s'), " |
| "while parsing msg/def/field `%s/%s/%s'" |
| % (field[0], p.type, p.name, definition, field) |
| ) |
| return True |
| |
| |
| class JsonParser(object): |
| def __init__( |
| self, |
| logger, |
| files, |
| simple_type_class=SimpleType, |
| enum_class=Enum, |
| union_class=Union, |
| struct_type_class=StructType, |
| field_class=Field, |
| message_class=Message, |
| alias_class=Alias, |
| ): |
| self.services = {} |
| self.messages = {} |
| self.enums = {} |
| self.unions = {} |
| self.aliases = {} |
| self.types = { |
| x: simple_type_class(x) |
| for x in [ |
| "i8", |
| "i16", |
| "i32", |
| "i64", |
| "u8", |
| "u16", |
| "u32", |
| "u64", |
| "f64", |
| "bool", |
| ] |
| } |
| |
| self.types["string"] = simple_type_class("vl_api_string_t") |
| self.replies = set() |
| self.events = set() |
| self.simple_type_class = simple_type_class |
| self.enum_class = enum_class |
| self.union_class = union_class |
| self.struct_type_class = struct_type_class |
| self.field_class = field_class |
| self.alias_class = alias_class |
| self.message_class = message_class |
| |
| self.exceptions = [] |
| self.json_files = [] |
| self.types_by_json = {} |
| self.enums_by_json = {} |
| self.unions_by_json = {} |
| self.aliases_by_json = {} |
| self.messages_by_json = {} |
| self.logger = logger |
| for f in files: |
| self.parse_json_file(f) |
| self.finalize_parsing() |
| |
| def parse_json_file(self, path): |
| self.logger.info("Parsing json api file: `%s'" % path) |
| self.json_files.append(path) |
| self.types_by_json[path] = [] |
| self.enums_by_json[path] = [] |
| self.unions_by_json[path] = [] |
| self.aliases_by_json[path] = [] |
| self.messages_by_json[path] = {} |
| with open(path) as f: |
| j = json.load(f) |
| for k in j["services"]: |
| if k in self.services: |
| raise ParseError("Duplicate service `%s'" % k) |
| self.services[k] = j["services"][k] |
| self.replies.add(self.services[k]["reply"]) |
| if "events" in self.services[k]: |
| for x in self.services[k]["events"]: |
| self.events.add(x) |
| for e in j["enums"]: |
| name = e[0] |
| value_pairs = e[1:-1] |
| enumtype = self.types[e[-1]["enumtype"]] |
| enum = self.enum_class(name, value_pairs, enumtype) |
| self.enums[enum.name] = enum |
| self.logger.debug("Parsed enum: %s" % enum) |
| self.enums_by_json[path].append(enum) |
| exceptions = [] |
| progress = 0 |
| last_progress = 0 |
| while True: |
| for u in j["unions"]: |
| name = u[0] |
| if name in self.unions: |
| progress = progress + 1 |
| continue |
| try: |
| type_pairs = [ |
| [self.lookup_type_like_id(t), n] for t, n in u[1:] |
| ] |
| union = self.union_class(name, type_pairs, 0) |
| progress = progress + 1 |
| except ParseError as e: |
| exceptions.append(e) |
| continue |
| self.unions[union.name] = union |
| self.logger.debug("Parsed union: %s" % union) |
| self.unions_by_json[path].append(union) |
| for t in j["types"]: |
| if t[0] in self.types: |
| progress = progress + 1 |
| continue |
| try: |
| type_ = self.struct_type_class( |
| t, self, self.field_class, self.logger |
| ) |
| if type_.name in self.types: |
| raise ParseError("Duplicate type `%s'" % type_.name) |
| progress = progress + 1 |
| except ParseError as e: |
| exceptions.append(e) |
| continue |
| self.types[type_.name] = type_ |
| self.types_by_json[path].append(type_) |
| self.logger.debug("Parsed type: %s" % type_) |
| for name, body in j["aliases"].items(): |
| if name in self.aliases: |
| progress = progress + 1 |
| continue |
| if "length" in body: |
| array_len = body["length"] |
| else: |
| array_len = None |
| try: |
| t = self.lookup_type_like_id(body["type"]) |
| except ParseError as e: |
| exceptions.append(e) |
| continue |
| alias = self.alias_class(name, t, array_len) |
| self.aliases[name] = alias |
| self.logger.debug("Parsed alias: %s" % alias) |
| self.aliases_by_json[path].append(alias) |
| if not exceptions: |
| # finished parsing |
| break |
| if progress <= last_progress: |
| # cannot make forward progress |
| self.exceptions.extend(exceptions) |
| break |
| exceptions = [] |
| last_progress = progress |
| progress = 0 |
| prev_length = len(self.messages) |
| processed = [] |
| while True: |
| exceptions = [] |
| for m in j["messages"]: |
| if m in processed: |
| continue |
| try: |
| msg = self.message_class(self.logger, m, self) |
| if msg.name in self.messages: |
| raise ParseError("Duplicate message `%s'" % msg.name) |
| except ParseError as e: |
| exceptions.append(e) |
| continue |
| self.messages[msg.name] = msg |
| self.messages_by_json[path][msg.name] = msg |
| processed.append(m) |
| if prev_length == len(self.messages): |
| # cannot make forward progress ... |
| self.exceptions.extend(exceptions) |
| break |
| prev_length = len(self.messages) |
| |
| def lookup_type_like_id(self, name): |
| mundane_name = remove_magic(name) |
| if name in self.types: |
| return self.types[name] |
| elif name in self.enums: |
| return self.enums[name] |
| elif name in self.unions: |
| return self.unions[name] |
| elif name in self.aliases: |
| return self.aliases[name] |
| elif mundane_name in self.types: |
| return self.types[mundane_name] |
| elif mundane_name in self.enums: |
| return self.enums[mundane_name] |
| elif mundane_name in self.unions: |
| return self.unions[mundane_name] |
| elif mundane_name in self.aliases: |
| return self.aliases[mundane_name] |
| raise ParseError( |
| "Could not find type, enum or union by magic name `%s' nor by " |
| "mundane name `%s'" % (name, mundane_name) |
| ) |
| |
| def is_reply(self, message): |
| return message in self.replies |
| |
| def is_event(self, message): |
| return message in self.events |
| |
| def get_reply(self, message): |
| return self.messages[self.services[message]["reply"]] |
| |
| def finalize_parsing(self): |
| if len(self.messages) == 0: |
| for e in self.exceptions: |
| self.logger.warning(e) |
| for jn, j in self.messages_by_json.items(): |
| remove = [] |
| for n, m in j.items(): |
| try: |
| if not m.is_reply and not m.is_event: |
| try: |
| m.reply = self.get_reply(n) |
| if "stream" in self.services[m.name]: |
| m.reply_is_stream = self.services[m.name]["stream"] |
| else: |
| m.reply_is_stream = False |
| m.reply.request = m |
| except: |
| raise ParseError("Cannot find reply to message `%s'" % n) |
| except ParseError as e: |
| self.exceptions.append(e) |
| remove.append(n) |
| |
| self.messages_by_json[jn] = {k: v for k, v in j.items() if k not in remove} |