diff --git a/dist/pythonlibs/riotctrl_shell/gnrc.py b/dist/pythonlibs/riotctrl_shell/gnrc.py index bb6861900a..d0c97dc63c 100644 --- a/dist/pythonlibs/riotctrl_shell/gnrc.py +++ b/dist/pythonlibs/riotctrl_shell/gnrc.py @@ -18,6 +18,20 @@ from riotctrl.shell import ShellInteraction, ShellInteractionParser # ==== Parsers ==== class GNRCICMPv6EchoParser(ShellInteractionParser): + def __init__(self): + self.c_reply = re.compile(r"\d+ bytes from " + r"(?P[0-9a-f:]+(%\S+)?): " + r"icmp_seq=(?P\d+) ttl=(?P\d+)" + r"( rssi=(?P-?\d+) dBm)?" + r"( time=(?P\d+.\d+) ms)?" + r"(?P \(DUP\))?") + self.c_stats = re.compile(r"(?P\d+) packets transmitted, " + r"(?P\d+) packets received, " + r"((?P\d+) duplicates, )?" + r"(?P\d+)% packet loss") + self.c_rtts = re.compile(r"round-trip min/avg/max = (?P\d+.\d+)/" + r"(?P\d+.\d+)/(?P\d+.\d+) ms") + @staticmethod def _add_reply(res, reply): reply["seq"] = int(reply["seq"]) @@ -106,29 +120,18 @@ class GNRCICMPv6EchoParser(ShellInteractionParser): 8.839 """ res = {} - c_reply = re.compile(r"\d+ bytes from (?P[0-9a-f:]+(%\S+)?): " - r"icmp_seq=(?P\d+) ttl=(?P\d+)" - r"( rssi=(?P-?\d+) dBm)?" - r"( time=(?P\d+.\d+) ms)?" - r"(?P \(DUP\))?") - c_stats = re.compile(r"(?P\d+) packets transmitted, " - r"(?P\d+) packets received, " - r"((?P\d+) duplicates, )?" - r"(?P\d+)% packet loss") - c_rtts = re.compile(r"round-trip min/avg/max = (?P\d+.\d+)/" - r"(?P\d+.\d+)/(?P\d+.\d+) ms") for line in cmd_output.splitlines(): if "stats" not in res: # If final stats were not found yet - m = c_reply.match(line) + m = self.c_reply.match(line) if m is not None: self._add_reply(res, m.groupdict()) continue - m = c_stats.match(line) + m = self.c_stats.match(line) if m is not None: self._set_stats(res, m.groupdict()) continue else: - m = c_rtts.match(line) + m = self.c_rtts.match(line) if m is not None: self._set_rtts(res, m.groupdict()) return res @@ -164,6 +167,21 @@ class GNRCPktbufStatsResults(dict): class GNRCPktbufStatsParser(ShellInteractionParser): + def __init__(self): + self.c_init1 = re.compile( + r"packet buffer: first byte: 0x(?P[0-9A-Fa-f]+), " + r"last byte: 0x(?P[0-9A-Fa-f]+) " + r"\(size: +(?P\d+)\)" + ) + self.c_init2 = re.compile(r" position of last byte used: (\d+)") + self.c_unused = re.compile( + r"~ unused: 0x(?P[0-9A-Fa-f]+) " + # flake reports r'\(', r'\)' as invalid escape sequence + # false positively + r"\(next: ""(0x(?P[0-9A-Fa-f]+)|\(nil\)), " # noqa W605 + r"size: +(?P\d+)\) ~" + ) + @staticmethod def _init_res(first_byte, last_byte, size): return GNRCPktbufStatsResults(( @@ -220,30 +238,22 @@ class GNRCPktbufStatsParser(ShellInteractionParser): >>> res["first_unused"]["size"] 8192 """ - c_init1 = re.compile(r"packet buffer: " - r"first byte: 0x(?P[0-9A-Fa-f]+), " - r"last byte: 0x(?P[0-9A-Fa-f]+) " - r"\(size: +(?P\d+)\)") - c_init2 = re.compile(r" position of last byte used: (\d+)") - c_unused = re.compile(r"~ unused: 0x(?P[0-9A-Fa-f]+) " - r"\(next: (0x(?P[0-9A-Fa-f]+)|\(nil\)), " - r"size: +(?P\d+)\) ~") res = None for line in cmd_output.splitlines(): if res is None: - m = c_init1.match(line) + m = self.c_init1.match(line) if m is not None: res = self._init_res(**m.groupdict()) # no sense in further parsing if we did not find the first line # yet. If we found it just continue parsing with next line continue elif "last_byte_used" not in res: - m = c_init2.match(line) + m = self.c_init2.match(line) if m is not None: self._set_last_byte_used(res, m.group(1)) continue elif "first_unused" not in res: - m = c_unused.match(line) + m = self.c_unused.match(line) if m is not None: self._set_first_unused(res, m.groupdict()) if res is not None and "last_byte_used" not in res: