Mirror
/
ardupilot
mirror of https://github.com/ArduPilot/ardupilot
5
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity
ardupilot/Tools/scripts/mavlink_parse.py

503 lines
21 KiB
Python
Executable File
Raw Blame History

#!/usr/bin/env python
import re
from enum import StrEnum # requires Python >= 3.11
from pathlib import Path
from itertools import chain
from dataclasses import dataclass, astuple
from pymavlink.dialects.v20 import (
common, icarous, cubepilot, uAvionix, ardupilotmega
)
class MAVLinkDialect(StrEnum):
# in subset, superset, unknown order, for correct links
# supported values must match imported dialect names
COMMON = 'common'
ICAROUS = 'icarous'
CUBEPILOT = 'cubepilot'
UAVIONIX = 'uAvionix'
ARDUPILOTMEGA = 'ardupilotmega'
UNKNOWN = 'UNKNOWN'
@dataclass(slots=True, order=True)
class MAVLinkMessage:
name: str
source: str
dialect: MAVLinkDialect = MAVLinkDialect.UNKNOWN
PREFIX = 'MAVLINK_MSG_ID_'
# Get message sets, for quick containment checks.
# Function is required because of Python's class scoping rules.
# See https://stackoverflow.com/questions/13905741.
def _get_known_messages(prefix):
''' Returns a dictionary of {dialect: {messages}} given 'prefix'. '''
return {
dialect: set(m for m in dir(globals()[dialect])
if m.startswith(prefix))
for dialect in MAVLinkDialect
if dialect != MAVLinkDialect.UNKNOWN
}
KNOWN_DIALECTS = _get_known_messages(PREFIX)
# Try to determine dialect if initialised without one specified.
def __post_init__(self):
if self.dialect == MAVLinkDialect.UNKNOWN:
self.determine_dialect()
@property
def id_name(self):
return self.PREFIX + self.name
def determine_dialect(self):
for dialect, message_set in self.KNOWN_DIALECTS.items():
if self.id_name in message_set:
self.dialect = dialect
break # dialect found, no need to continue searching
else:
self.dialect = MAVLinkDialect.UNKNOWN
def as_tuple(self):
return astuple(self)
def __str__(self):
return f'{self.name:<45}{self.source:<55}{self.dialect}'
@classmethod
def get_unsupported(cls, supported: set, remove_prefix=True):
''' Yields known messages that are not included in 'supported'. '''
offset = len(cls.PREFIX) if remove_prefix else 0
known_missing = set() # don't double-count for supersets
for dialect, message_set in cls.KNOWN_DIALECTS.items():
missing_names = message_set - supported - known_missing
for name in missing_names:
yield cls(name[offset:], 'UNSUPPORTED', dialect)
known_missing |= missing_names
class MAVLinkCommand(MAVLinkMessage):
PREFIX = 'MAV_CMD_'
KNOWN_DIALECTS = MAVLinkMessage._get_known_messages(PREFIX)
@property
def id_name(self):
return self.name # commands are registered with their prefix
@classmethod
def get_unsupported(cls, supported: set, remove_prefix=False):
''' Yields known commands that are not included in 'supported'. '''
# avoid accidentally treating enum values as commands
enums = [f'{e}_' for e in ardupilotmega.enums
if e.startswith(cls.PREFIX)]
for command in super().get_unsupported(supported, remove_prefix):
if not any(command.name.startswith(e) for e in enums):
yield command
class MAVLinkDetector:
# file paths
BASE_DIR = Path(__file__).parent / '../..'
COMMON_FILE = BASE_DIR / 'libraries/GCS_MAVLink/GCS_Common.cpp'
STREAM_GROUP_FILE = 'GCS_MAVLink.cpp'
# regex for messages handled by the autopilot
INCOMING_MESSAGES = re.compile(r'case MAVLINK_MSG_ID_([A-Z0-9_]*)')
# regex for commands handled by the autopilot
INCOMING_COMMANDS = re.compile(r'case (MAV_CMD_[A-Z0-9_]*)')
# regex for messages that can be requested from the autopilot
REQUESTABLE_REGION = re.compile(' map\[\]([^;]*);')
REQUESTABLE_MAP = re.compile(r'MAVLINK_MSG_ID_([A-Z0-9_]*),\s*MSG_([A-Z0-9_]*)')
# regex for messages the autopilot might send, but cannot be requested
OUTGOING_MESSAGES = re.compile(r'mavlink_msg_([a-z0-9_]*)_send\(')
# regex for extracting messages in stream groups
STREAM_GROUPS = re.compile(r'ap_message STREAM_([A-Z0-9_]*)_msgs\[\] = \{([^\}]*)')
AP_MESSAGE = re.compile(r'MSG_([A-Z0-9_]*)')
# regex for named values
NAMED_FLOAT = re.compile(r'send_named_float\("([\w]*)"')
NAMED_INT = re.compile(r'send_named_int\("([\w]*)"')
TYPE_DESCRIPTIONS = {
'incoming_messages':
'Messages the autopilot handles when received.',
'requestable_messages':
'Messages that can be requested/streamed from the autopilot.',
'outgoing_messages':
'Messages the autopilot will send automatically (unrequested).',
'named_floats':
'Breakout of named floating-point (numerical) values sent by the autopilot.',
'named_ints':
'Breakout of named integer values sent by the autopilot.',
}
EXTRA_DESCRIPTIONS = {
'stream_groups':
'Message groups with stream rates requestable by `SRn_*` parameters.'
' Messages in a group are only sent if the corresponding feature'
' is active.',
'missing_messages':
'Unsupported / unhandled messages.',
'incoming_commands':
TYPE_DESCRIPTIONS['incoming_messages'].replace('Messages','Commands'),
}
EXTRA_DESCRIPTIONS['missing_commands'] = \
EXTRA_DESCRIPTIONS['missing_messages'].replace('messages', 'commands')
TYPE_OPTIONS = {
'messages': MAVLinkMessage,
'commands': MAVLinkCommand,
}
MAVLINK_URL = 'https://mavlink.io/en/messages/{dialect}.html#{message_name}'
ARDUPILOT_URL = 'https://github.com/ArduPilot/ardupilot/tree/{branch}/{source}'
EXPORT_FILETYPES = {
'csv': 'csv',
'markdown': 'md'
}
MARKDOWN_INTRO = (
'The [MAVLink](https://mavlink.io/en/) protocol supports a variety'
' of features and functionalities, but not all'
' [messages](https://mavlink.io/en/messages/) or'
' [commands](https://mavlink.io/en/services/command.html)'
' are implemented by the ArduPilot ecosystem, or relevant to a'
' particular autopilot firmware.\n\n'
'This page is auto-generated from analysing the {vehicle} source'
' code, and provides an indication of which messages{commands} are'
' handled by, requestable from, and sent from the firmware. '
'A message being handled does not guarantee full support, but at'
' least shows that the autopilot is aware it exists, and will try'
' to do something meaningful with it.{unsupported}{stream_groups}'
)
VEHICLES = ('AntennaTracker', 'ArduCopter', 'ArduPlane', 'ArduSub', 'Rover')
def __init__(self, common_files, vehicle='ALL',
exclude_libraries=['SITL', 'AP_Scripting']):
self.vehicle = vehicle
vehicles = [vehicle] if vehicle != 'ALL' else self.VEHICLES
files = chain(*((self.BASE_DIR / vehicle).glob('**/*.cpp')
for vehicle in vehicles),
common_files)
self.incoming_messages = {}
self.incoming_commands = {}
self.outgoing_messages = {}
self.requestable_messages = {}
self._ap_to_mavlink = {
'NAMED_FLOAT': 'NAMED_VALUE_FLOAT', # manual inclusion
}
self.named_floats = {}
self.named_ints = {}
for file in files:
folder = file.parent.stem
if folder in exclude_libraries:
continue
text = file.read_text()
source = f'{folder}/{file.name}'
if file == self.COMMON_FILE:
for mavlink, ap_message in self.find_requestable_messages(text):
self.requestable_messages[mavlink] = \
MAVLinkMessage(mavlink, source)
if ap_message != mavlink:
self._ap_to_mavlink[ap_message] = mavlink
named_types = ('float', 'int') if folder in vehicles else ()
for type_ in named_types:
substring = f'named_{type_}s'
method = getattr(self, f'find_{substring}')
names = getattr(self, substring)
new_names = set(method(text)) - names.keys()
for name in new_names:
names[name] = MAVLinkMessage(f'NAMED_VALUE_{type_.upper()}:{name}',
source, MAVLinkDialect.COMMON)
for method, data, type_ in (
(self.find_incoming_messages, self.incoming_messages, 'messages'),
(self.find_incoming_commands, self.incoming_commands, 'commands'),
(self.find_outgoing_messages, self.outgoing_messages, 'messages'),
):
new_data = set(method(text)) - data.keys()
cls = self.TYPE_OPTIONS[type_]
for datum in new_data:
data[datum] = cls(datum, source)
self._supported_names = {'messages': None, 'commands': None}
self._unsupported = self._supported_names.copy()
self._stream_groups = self.get_stream_groups(vehicle) if len(vehicles) == 1 else []
@classmethod
def get_description(cls, query):
return cls.TYPE_DESCRIPTIONS.get(query,
cls.EXTRA_DESCRIPTIONS.get(query, '')
)
@classmethod
def find_incoming_messages(cls, text: str):
return cls.INCOMING_MESSAGES.findall(text)
@classmethod
def find_incoming_commands(cls, text: str):
return cls.INCOMING_COMMANDS.findall(text)
@classmethod
def find_outgoing_messages(cls, text: str):
return (msg.upper() for msg in
cls.OUTGOING_MESSAGES.findall(text))
@classmethod
def find_requestable_messages(cls, text: str):
region = cls.REQUESTABLE_REGION.search(text).group()
return cls.REQUESTABLE_MAP.findall(region)
@classmethod
def find_named_floats(cls, text: str):
return cls.NAMED_FLOAT.findall(text)
@classmethod
def find_named_ints(cls, text: str):
return cls.NAMED_INT.findall(text)
def get_stream_groups(self, vehicle):
stream_groups = ['stream_groups']
text = (self.BASE_DIR / vehicle / self.STREAM_GROUP_FILE).read_text()
for group_name, message_data in self.STREAM_GROUPS.findall(text):
stream_groups.extend(sorted(
MAVLinkMessage(self._ap_to_mavlink.get(ap_message, ap_message),
f'SRn_{group_name}')
for ap_message in self.AP_MESSAGE.findall(message_data)
))
return stream_groups
def get_supported(self, type: str, inject_commands=False):
if type == 'messages':
for message_type in self.TYPE_DESCRIPTIONS:
values = getattr(self, message_type).values()
if not values:
continue
yield message_type
yield from sorted(values)
# add in incoming_commands right after incoming_messages
if inject_commands and message_type == 'incoming_messages':
yield from self.get_supported('commands')
elif type == 'commands':
yield 'incoming_commands'
yield from sorted(self.incoming_commands.values())
def get_supported_names(self, type: str):
if self._supported_names[type] is None:
self._supported_names[type] = set(
m.id_name for m in self.get_supported(type)
if isinstance(m, MAVLinkMessage)
)
return self._supported_names[type]
def get_unsupported(self, type='messages'):
if self._unsupported[type] is None:
supported_messages = self.get_supported_names(type)
cls = self.TYPE_OPTIONS[type]
self._unsupported[type] = sorted(
cls.get_unsupported(supported_messages)
)
if self._unsupported[type]:
yield f'missing_{type}'
yield from self._unsupported[type]
def get_iterable(self, include_commands=False, include_stream_groups=False,
include_unsupported=False):
iterables = [self.get_supported('messages', include_commands)]
if include_stream_groups:
iterables.append(self._stream_groups)
if include_unsupported:
iterables.append(self.get_unsupported('messages'))
if include_commands:
iterables.append(self.get_unsupported('commands'))
return chain(*iterables)
def printout(self, **iter_options):
for data in self.get_iterable(**iter_options):
match data: # requires Python >= 3.10
case str() as type_:
print(f'\n{type_}:',
self.get_description(type_),
sep='\n')
case MAVLinkMessage() as message:
print(message)
def export(self, filename: Path, type='csv', include_commands=False,
include_stream_groups=False, include_unsupported=False,
**export_options):
export_method = getattr(self, f'export_{type}')
# ensure export_method and get_iterable have the same options specified
iter_options = dict(
include_commands = include_commands,
include_stream_groups = include_stream_groups,
include_unsupported = include_unsupported,
)
with open(filename, 'w') as file:
export_method(file, self.get_iterable(**iter_options),
**export_options, **iter_options)
def export_csv(self, file, iterable, **ignore):
file.write('MAVLinkMessage,CodeSource,MAVLinkDialect,MessageType\n')
for data in iterable:
match data:
case str():
current_type = data
case MAVLinkMessage() as message:
print(*message.as_tuple(), current_type, sep=',', file=file)
def export_markdown(self, file, iterable, branch='master', header=None,
use_intro=True, **extra_kwargs):
if header == 'ArduSub':
import time
now = time.strftime('%Y-%m-%dT%H:%M:%S%z')
date = f'{now[:-2]}:{now[-2:]}' # add colon to the timezone
header = '\n'.join((
'+++',
'title = "MAVLink Support"',
'description = "MAVLink message support details."',
f'{date = }', 'template = "docs/page.html"',
'sort_by = "weight"', 'weight = 20', 'draft = false',
'[extra]', 'toc = true', 'top = false',
'+++'
))
if header:
print(header, file=file)
if use_intro:
commands = stream_groups = unsupported = ''
if extra_kwargs['include_commands']:
commands = ' (and commands)'
if extra_kwargs['include_unsupported']:
unsupported = (
'\n\nKnown [unsupported messages](#missing-messages)'
f'{commands} are shown at the end.'
)
if extra_kwargs['include_stream_groups']:
stream_groups = (
'\n\nThe autopilot includes a set of [stream groups]'
'(#stream-groups) for convenience, which allow'
' configuring the stream rates of groups of'
' requestable messages by setting parameter values. '
'It is also possible to manually request messages,'
' and request individual messages be streamed at a'
' specified rate.'
)
vehicle = self.vehicle.replace('ALL', 'ArduPilot')
print(self.MARKDOWN_INTRO.format(
vehicle=vehicle, commands=commands,
stream_groups=stream_groups, unsupported=unsupported
), file=file)
for data in iterable:
match data:
case str() as type_:
heading = type_.title().replace('_', ' ')
source_header = (
'Code Source' if type_ != 'stream_groups' else
'Stream Group Parameter'
)
print(f'## {heading}',
self.get_description(type_),
f'\nMAVLink Message | {source_header} | MAVLink Dialect',
'--- | --- | ---', sep='\n', file=file)
case MAVLinkMessage() as message:
name, source, dialect = message.as_tuple()
if dialect != MAVLinkDialect.UNKNOWN:
msg_url = self.MAVLINK_URL.format(dialect=dialect,
message_name=name.split(':')[0])
name = f'[{name}]({msg_url})'
if source != 'UNSUPPORTED' and not source.startswith('SRn'):
folder = source.split('/')[0]
base = 'libraries/' if folder not in self.VEHICLES else ''
code_url = self.ARDUPILOT_URL.format(branch=branch,
source=base+source)
source = f'[{source}]({code_url})'
print(name, source, dialect, sep=' | ', file=file)
if __name__ == '__main__':
from inspect import signature
from argparse import ArgumentParser, ArgumentDefaultsHelpFormatter
detector_init_params = signature(MAVLinkDetector.__init__).parameters
default_vehicle = detector_init_params['vehicle'].default
vehicle_options = [default_vehicle, *MAVLinkDetector.VEHICLES]
default_exclusions = detector_init_params['exclude_libraries'].default
parser = ArgumentParser(formatter_class=ArgumentDefaultsHelpFormatter)
parse_opts = parser.add_argument_group('parsing options')
parse_opts.add_argument('-v', '--vehicle', default=default_vehicle,
choices=vehicle_options, help='Vehicle folder, or ALL.')
parse_opts.add_argument('-e', '--exclude-library', action='append',
default=default_exclusions,
help='Libraries to exclude from the search.')
parse_opts.add_argument('-c', '--include-commands', action='store_true',
help='Include MAVLink commands as well as messages.')
parse_opts.add_argument('-g', '--include-stream-groups', action='store_true',
help='Include stream group message sets in the output.')
parse_opts.add_argument('-u', '--include-unsupported', action='store_true',
help='Include unsupported messages in the output.')
export_opts = parser.add_argument_group('export options')
export_opts.add_argument('-q', '--quiet', action='store_true',
help='Disable printout, only export a file.')
export_opts.add_argument('-f', '--format', default='markdown',
choices=['csv', 'markdown', 'none'],
help='Desired format for the exported file.')
export_opts.add_argument('-b', '--branch',
help=('The branch to link to in markdown mode.'
' Defaults to the branch in the working directory.'))
export_opts.add_argument('--filename', help='Override default filename.')
export_opts.add_argument('--header', help='Header for the markdown file.')
export_opts.add_argument('--no-intro', action='store_true',
help="Flag to not use the automatic markdown intro.")
args = parser.parse_args()
assert (args.vehicle in MAVLinkDetector.VEHICLES
or not args.include_stream_groups), \
'Determining stream groups requires a single vehicle to be specified.'
common_files = (MAVLinkDetector.BASE_DIR / 'libraries').glob('**/*.cpp')
messages = MAVLinkDetector(common_files, args.vehicle, args.exclude_library)
include_options = dict(
include_commands = args.include_commands,
include_stream_groups = args.include_stream_groups,
include_unsupported = args.include_unsupported,
)
if not args.quiet:
messages.printout(**include_options)
if args.format != 'none':
ext = messages.EXPORT_FILETYPES[args.format]
branch = args.branch
if not branch:
import subprocess
pattern = re.compile(r'On branch ([\S]*)')
result = subprocess.run(['git', 'status'], capture_output=True).stdout
try:
branch, = pattern.search(result.decode()).groups()
except AttributeError as e:
raise Exception(
'No --branch specified, and "git status" failed to find one.'
'Please manually specify an ardupilot firmware branch for '
'code source hyperlinks (e.g. Sub-4.1) or ensure this '
'repository copy is managed by git.'
)
filename = (
args.filename or
f'{args.vehicle}_{branch}_MAVLink_Messages.{ext}'
)
messages.export(filename, type=args.format, branch=branch, header=args.header,
use_intro=not args.no_intro, **include_options)
Reference in New Issue View Git Blame Copy Permalink