#!/usr/bin/env python ''' AP_FLAKE8_CLEAN ''' from __future__ import print_function import argparse import copy import os import re import sys import emit_html import emit_rst import emit_xml import emit_md import enum_parse from enum_parse import EnumDocco topdir = os.path.join(os.path.dirname(os.path.realpath(__file__)), '../../../') topdir = os.path.realpath(topdir) # Regular expressions for finding message information in code comments re_loggermessage = re.compile(r"@LoggerMessage\s*:\s*([\w,]+)", re.MULTILINE) re_commentline = re.compile(r"\s*//") re_description = re.compile(r"\s*//\s*@Description\s*:\s*(.*)") re_url = re.compile(r"\s*//\s*@URL\s*:\s*(.*)") re_field = re.compile(r"\s*//\s*@Field\s*:\s*(\w+):\s*(.*)") re_fieldbits = re.compile(r"\s*//\s*@FieldBits\s*:\s*(\w+):\s*(.*)") re_fieldbitmaskenum = re.compile(r"\s*//\s*@FieldBitmaskEnum\s*:\s*(\w+):\s*(.*)") re_fieldvalueenum = re.compile(r"\s*//\s*@FieldValueEnum\s*:\s*(\w+):\s*(.*)") re_vehicles = re.compile(r"\s*//\s*@Vehicles\s*:\s*(.*)") # Regular expressions for finding message definitions in structure format re_start_messagedef = re.compile(r"^\s*{?\s*LOG_[A-Z0-9_]+_[MSGTA]+[A-Z0-9_]*\s*,") re_deffield = r'[\s\\]*"?([\w\-#?%]+)"?\s*' re_full_messagedef = re.compile(r'\s*LOG_\w+\s*,\s*\w+\([^)]+\)[\s\\]*,' + f'{re_deffield},{re_deffield},' + r'[\s\\]*"?([\w,]+)"?[\s\\]*,' + f'{re_deffield},{re_deffield}', re.MULTILINE) re_fmt_define = re.compile(r'#define\s+(\w+_FMT)\s+"([\w\-#?%]+)"') re_units_define = re.compile(r'#define\s+(\w+_UNITS)\s+"([\w\-#?%]+)"') re_mults_define = re.compile(r'#define\s+(\w+_MULTS)\s+"([\w\-#?%]+)"') # Regular expressions for finding message definitions in Write calls re_start_writecall = re.compile(r"\s*[AP:]*logger[\(\)]*.Write[StreamingCrcl]*\(") re_writefield = r'\s*"([\w\-#?%,]+)"\s*' re_full_writecall = re.compile(r'\s*[AP:]*logger[\(\)]*.Write[StreamingCrcl]*\(' + f'{re_writefield},{re_writefield},{re_writefield},({re_writefield},{re_writefield})?', re.MULTILINE) # Regular expression for extracting unit and multipliers from structure re_units_mults_struct = re.compile(r"^\s*{\s*'([\w\-#?%!/])',"+r'\s*"?([\w\-#?%./]*)"?\s*}') # TODO: validate URLS actually return 200 # Lookup tables are populated by reading LogStructure.h log_fmt_lookup = {} log_units_lookup = {} log_mult_lookup = {} # Lookup table to convert multiplier to prefix mult_prefix_lookup = { 0: "", 1: "", 1e-1: "d", # deci- 1e-2: "c", # centi- 1e-3: "m", # milli- 1e-6: "μ", # micro- 1e-9: "n" # nano- } class LoggerDocco(object): vehicle_map = { "Rover": "Rover", "Sub": "ArduSub", "Copter": "ArduCopter", "Plane": "ArduPlane", "Tracker": "AntennaTracker", "Blimp": "Blimp", } def __init__(self, vehicle): self.vehicle = vehicle self.doccos = [] self.emitters = [ emit_html.HTMLEmitter(), emit_rst.RSTEmitter(), emit_xml.XMLEmitter(), emit_md.MDEmitter(), ] self.msg_fmts_list = {} self.msg_units_list = {} self.msg_mults_list = {} class Docco(object): def __init__(self, name): self.name = name self.url = None if isinstance(name, list): self.description = [None] * len(name) else: self.description = None self.fields = {} self.fields_order = [] self.vehicles = None self.bits_enums = [] def add_name(self, name): # If self.name/description aren't lists, convert them if isinstance(self.name, str): self.name = [self.name] self.description = [self.description] # Replace any existing empty descriptions with empty strings for i in range(0, len(self.description)): if self.description[i] is None: self.description[i] = "" # Extend the name and description lists if isinstance(name, list): self.name.extend(name) self.description.extend([None] * len(name)) else: self.name.append(name) self.description.append(None) def set_description(self, desc): if isinstance(self.description, list): for i in range(0, len(self.description)): if self.description[i] is None: self.description[i] = desc else: self.description = desc def set_url(self, url): self.url = url def ensure_field(self, field): if field not in self.fields: self.fields[field] = {} self.fields_order.append(field) def set_field_description(self, field, description): if field in self.fields: raise ValueError("Already have field %s in %s" % (field, self.name)) self.ensure_field(field) self.fields[field]["description"] = description def set_field_bits(self, field, bits): bits = bits.split(",") count = 0 entries = [] for bit in bits: entries.append(EnumDocco.EnumEntry(bit, 1 << count, None)) count += 1 bitmask_name = self.name + field self.bits_enums.append(EnumDocco.Enumeration(bitmask_name, entries)) self.ensure_field(field) self.fields[field]["bitmaskenum"] = bitmask_name def set_fieldbitmaskenum(self, field, bits): self.ensure_field(field) self.fields[field]["bitmaskenum"] = bits def set_fieldvalueenum(self, field, bits): self.ensure_field(field) self.fields[field]["valueenum"] = bits def set_vehicles(self, vehicles): self.vehicles = vehicles def set_fmts(self, fmts): # If no fields are defined, do nothing if len(self.fields_order) == 0: return # Make sure lengths match up if len(fmts) != len(self.fields_order): print(f"Number of fmts don't match fields: msg={self.name} fmts={fmts} num_fields={len(self.fields_order)}") return # Loop through the list for idx in range(0, len(fmts)): if fmts[idx] in log_fmt_lookup: self.fields[self.fields_order[idx]]["fmt"] = log_fmt_lookup[fmts[idx]] else: print(f"Unrecognised format character: {fmts[idx]} in message {self.name}") def set_units(self, units, mults): # If no fields are defined, do nothing if len(self.fields_order) == 0: return # Make sure lengths match up if len(units) != len(self.fields_order) or len(units) != len(mults): print(f"Number of units/mults/fields don't match: msg={self.name} units={units} mults={mults} num_fields={len(self.fields_order)}") # noqa:E501 return # Loop through the list for idx in range(0, len(units)): # Get the index into fields from field_order f = self.fields_order[idx] # Convert unit char to base unit if units[idx] in log_units_lookup: baseunit = log_units_lookup[units[idx]] else: print(f"Unrecognised units character: {units[idx]} in message {self.name}") continue # Do nothing if this field has no unit defined if baseunit == "": continue # Convert mult char to value if mults[idx] in log_mult_lookup: mult = log_mult_lookup[mults[idx]] mult_num = float(mult) else: print(f"Unrecognised multiplier character: {mults[idx]} in message {self.name}") continue # Check if the defined format for this field contains its own multiplier # If so, the presented value will be the base-unit directly if 'fmt' in self.fields[f] and self.fields[f]['fmt'].endswith("* 100"): self.fields[f]["units"] = baseunit elif 'fmt' in self.fields[f] and "latitude/longitude" in self.fields[f]['fmt']: self.fields[f]["units"] = baseunit # Check if we have a defined prefix for this multiplier elif mult_num in mult_prefix_lookup: self.fields[f]["units"] = f"{mult_prefix_lookup[mult_num]}{baseunit}" # If all else fails, set the unit as the multipler and base unit together else: self.fields[f]["units"] = f"{mult} {baseunit}" def populate_lookups(self): # Initialise the lookup tables # Read the contents of the LogStructure.h file structfile = os.path.join(topdir, "libraries", "AP_Logger", "LogStructure.h") with open(structfile) as f: lines = f.readlines() f.close() # Initialise current section to none section = "none" # Loop through the lines in the file for line in lines: # Look for the start of fmt/unit/mult info if line.startswith("Format characters"): section = "fmt" elif line.startswith("const struct UnitStructure"): section = "units" elif line.startswith("const struct MultiplierStructure"): section = "mult" # Read formats from code comment, e.g.: # b : int8_t elif section == "fmt": if "*/" in line: section = "none" else: parts = line.split(":") log_fmt_lookup[parts[0].strip()] = parts[1].strip() # Read units or multipliers from C struct definition, e.g.: # { '2', 1e2 }, or { 'J', "W.s" }, elif section != "none": if "};" in line: section = "none" else: u = re_units_mults_struct.search(line) if u is not None and section == "units": log_units_lookup[u.group(1)] = u.group(2) if u is not None and section == "mult": log_mult_lookup[u.group(1)] = u.group(2) def search_for_files(self, dirs_to_search): _next = [] for _dir in dirs_to_search: _dir = os.path.join(topdir, _dir) for entry in os.listdir(_dir): filepath = os.path.join(_dir, entry) if os.path.isdir(filepath): _next.append(filepath) continue (name, extension) = os.path.splitext(filepath) if extension not in [".cpp", ".h"]: continue self.files.append(filepath) if len(_next): self.search_for_files(_next) def parse_messagedef(self, messagedef): # Merge concatinated strings and remove comments messagedef = re.sub(r'"\s+"', '', messagedef) messagedef = re.sub(r'//[^\n]*', '', messagedef) # Extract details from a structure definition d = re_full_messagedef.search(messagedef) if d is not None: self.msg_fmts_list[d.group(1)] = d.group(2) self.msg_units_list[d.group(1)] = d.group(4) self.msg_mults_list[d.group(1)] = d.group(5) return # Extract details from a WriteStreaming call d = re_full_writecall.search(messagedef) if d is not None: if d.group(1) in self.msg_fmts_list: return if d.group(5) is None: self.msg_fmts_list[d.group(1)] = d.group(3) else: self.msg_fmts_list[d.group(1)] = d.group(6) self.msg_units_list[d.group(1)] = d.group(3) self.msg_mults_list[d.group(1)] = d.group(5) return # Didn't parse # print(f"Unable to parse: {messagedef}") def search_messagedef_start(self, line, prevmessagedef=""): # Look for the start of a structure definition d = re_start_messagedef.search(line) if d is not None: messagedef = line if "}" in line: self.parse_messagedef(messagedef) return "" else: return messagedef # Look for a new call to WriteStreaming d = re_start_writecall.search(line) if d is not None: messagedef = line if ";" in line: self.parse_messagedef(messagedef) return "" else: return messagedef # If we didn't find a new one, continue with any previous state return prevmessagedef def parse_file(self, filepath): with open(filepath) as f: # print("Opened (%s)" % filepath) lines = f.readlines() f.close() def debug(x): pass # if filepath == "/home/pbarker/rc/ardupilot/libraries/AP_HAL/AnalogIn.h": # debug = print state_outside = "outside" state_inside = "inside" messagedef = "" state = state_outside docco = None for line in lines: debug(f"{state}: {line}") if messagedef: messagedef = messagedef + line if "}" in line or ";" in line: self.parse_messagedef(messagedef) messagedef = "" if state == state_outside: # Check for start of a message definition messagedef = self.search_messagedef_start(line, messagedef) # Check for fmt/unit/mult #define u = re_fmt_define.search(line) if u is not None: self.msg_fmts_list[u.group(1)] = u.group(2) u = re_units_define.search(line) if u is not None: self.msg_units_list[u.group(1)] = u.group(2) u = re_mults_define.search(line) if u is not None: self.msg_mults_list[u.group(1)] = u.group(2) # Check for the @LoggerMessage tag indicating the start of the docco block m = re_loggermessage.search(line) if m is None: continue name = m.group(1) if "," in name: name = name.split(",") state = state_inside docco = LoggerDocco.Docco(name) elif state == state_inside: # If this line is not a comment, then this is the end of the docco block if not re_commentline.match(line): state = state_outside if docco.vehicles is None or self.vehicle in docco.vehicles: self.finalise_docco(docco) messagedef = self.search_messagedef_start(line) continue # Check for an multiple @LoggerMessage lines in this docco block m = re_loggermessage.search(line) if m is not None: name = m.group(1) if "," in name: name = name.split(",") docco.add_name(name) continue # Find and extract data from the various docco fields m = re_description.match(line) if m is not None: docco.set_description(m.group(1)) continue m = re_url.match(line) if m is not None: docco.set_url(m.group(1)) continue m = re_field.match(line) if m is not None: docco.set_field_description(m.group(1), m.group(2)) continue m = re_fieldbits.match(line) if m is not None: docco.set_field_bits(m.group(1), m.group(2)) continue m = re_fieldbitmaskenum.match(line) if m is not None: docco.set_fieldbitmaskenum(m.group(1), m.group(2)) continue m = re_fieldvalueenum.match(line) if m is not None: docco.set_fieldvalueenum(m.group(1), m.group(2)) continue m = re_vehicles.match(line) if m is not None: docco.set_vehicles([x.strip() for x in m.group(1).split(',')]) continue print("Unknown field (%s)" % str(line)) sys.exit(1) def parse_files(self): for _file in self.files: self.parse_file(_file) def emit_output(self): # expand things like PIDR,PIDQ,PIDA into multiple doccos new_doccos = [] for docco in self.doccos: if isinstance(docco.name, list): for name, desc in zip(docco.name, docco.description): tmpdocco = copy.copy(docco) tmpdocco.name = name tmpdocco.description = desc new_doccos.append(tmpdocco) else: new_doccos.append(docco) new_doccos = sorted(new_doccos, key=lambda x : x.name) # Try to attach the formats/units/multipliers for docco in new_doccos: # Apply the Formats to the docco if docco.name in self.msg_fmts_list: if "FMT" in self.msg_fmts_list[docco.name]: if self.msg_fmts_list[docco.name] in self.msg_fmts_list: docco.set_fmts(self.msg_fmts_list[self.msg_fmts_list[docco.name]]) else: docco.set_fmts(self.msg_fmts_list[docco.name]) else: print(f"No formats found for message {docco.name}") # Get the Units units = None if docco.name in self.msg_units_list: if "UNITS" in self.msg_units_list[docco.name]: if self.msg_units_list[docco.name] in self.msg_units_list: units = self.msg_units_list[self.msg_units_list[docco.name]] else: units = self.msg_units_list[docco.name] # Get the Multipliers mults = None if docco.name in self.msg_mults_list: if "MULTS" in self.msg_mults_list[docco.name]: if self.msg_mults_list[docco.name] in self.msg_mults_list: mults = self.msg_mults_list[self.msg_mults_list[docco.name]] else: mults = self.msg_mults_list[docco.name] # Apply the units/mults to the docco if units is not None and mults is not None: docco.set_units(units, mults) elif units is not None or mults is not None: print(f"Cannot find matching units/mults for message {docco.name}") enums_by_name = {} for enum in self.enumerations: enums_by_name[enum.name] = enum for emitter in self.emitters: emitter.emit(new_doccos, enums_by_name) def run(self): self.populate_lookups() self.enumerations = enum_parse.EnumDocco(self.vehicle).get_enumerations() self.files = [] self.search_for_files([self.vehicle_map[self.vehicle], "libraries"]) self.parse_files() self.emit_output() def finalise_docco(self, docco): self.doccos.append(docco) self.enumerations += docco.bits_enums if __name__ == '__main__': parser = argparse.ArgumentParser(description="Parse parameters.") parser.add_argument("-v", "--verbose", dest='verbose', action='store_true', default=False, help="show debugging output") parser.add_argument("--vehicle", required=True, help="Vehicle type to generate for") args = parser.parse_args() s = LoggerDocco(args.vehicle) if args.vehicle not in s.vehicle_map: print("Invalid vehicle (choose from: %s)" % str(s.vehicle_map.keys())) sys.exit(1) s.run()