Refactor the Python code for readability

benchmarks/__main__.py

@@ -96,11 +96,14 @@ def _command(parser: argparse.ArgumentParser) -> Command:
     namespace = parser.parse_args()
     quick: bool = namespace.quick
     repeat: int | None = namespace.repeat
+    if repeat is None:
+        repeat = 1 if quick else 3
+
     command = Command(
         suites=namespace.suite,
         libraries=namespace.libs,
         quick=quick,
-        repeat=repeat if repeat is not None else (1 if quick else 3),
+        repeat=repeat,
         cases=namespace.case,
         out=namespace.out,
     )
@@ -115,19 +118,26 @@ def _select(parser: argparse.ArgumentParser, command: Command) -> Command:
     # nothing runs yet, so nothing logs
     options = command.options(io.BytesIO())
     names = {suite: SUITES[suite].cases(options) for suite in command.suites}
+    every_name = [name for cases in names.values() for name in cases]
     unmatched = [
-        pattern
-        for pattern in command.cases
-        if not any(fnmatch.fnmatchcase(name, pattern) for suite in names.values() for name in suite)
+        pattern for pattern in command.cases if not any(fnmatch.fnmatchcase(name, pattern) for name in every_name)
     ]
     if len(unmatched) != 0:
-        listing = ''.join(f'\n{suite}:' + ''.join(f'\n  {name}' for name in cases) for suite, cases in names.items())
-        parser.error(f'--case {", ".join(unmatched)} matches no case; the cases are:{listing}')
+        parser.error(f'--case {", ".join(unmatched)} matches no case; the cases are:{_listing(names)}')
 
     suites = [suite for suite, cases in names.items() if any(options.selects(name) for name in cases)]
     return dataclasses.replace(command, suites=suites)
 
 
+def _listing(names: dict[str, list[str]]) -> str:
+    """The cases of each suite, a line each, under the suite's name."""
+    lines: list[str] = []
+    for suite, cases in names.items():
+        lines.append(f'{suite}:')
+        lines += [f'  {name}' for name in cases]
+    return ''.join(f'\n{line}' for line in lines)
+
+
 async def _versions(libraries: Sequence[str], log: IO[bytes]) -> dict[str, Hello]:
     versions: dict[str, Hello] = {}
     for library in libraries:
@@ -171,6 +181,7 @@ def main() -> None:
     command = _command(_parser())
     directory = command.out if command.out is not None else RESULTS
     directory.mkdir(parents=True, exist_ok=True)
+
     stamp = dt.datetime.now().astimezone().strftime('%Y-%m-%dT%H%M%S')
     name = f'{stamp}-{platform.system().lower()}-{platform.machine().lower()}'
     with (directory / f'{name}.log').open('ab') as log:

benchmarks/charts.py

@@ -64,9 +64,10 @@ def _label(library: str, versions: dict[str, Hello]) -> str:
 
 def _footer(found: Newest) -> str:
     """The machine and the conditions of the runs, for the foot of their charts."""
-    manifest = found.sources[-1][0]
+    manifest, _ = found.sources[-1]
     go = f', {manifest["go"]}' if manifest['go'] is not None else ''
     dates = sorted({source['date'][:10] for source, _ in found.sources})
+
     if any(source['quick'] for source, _ in found.sources):
         runs = 'a quick run, not results'
     elif len(found.sources) > 1:
@@ -82,16 +83,20 @@ def _times(value: float, _: int | None) -> str:
 
 def _detail(ax: Axes, detail: Detail, *, categories: list[str], legend: dict[str, str]) -> None:
     """Draws a panel of absolute values under a chart, its bars in the chart's categories and colors."""
-    rows = [
-        (category, legend[library], value)
-        for category in categories
-        for library, value in detail.values.get(category, {}).items()
-        if library in legend and math.isfinite(value)
-    ]
+    bar_categories: list[str] = []
+    bar_labels: list[str] = []
+    bar_values: list[float] = []
+    for category in categories:
+        for library, value in detail.values.get(category, {}).items():
+            if library in legend and math.isfinite(value):
+                bar_categories.append(category)
+                bar_labels.append(legend[library])
+                bar_values.append(value)
+
     sns.barplot(
-        x=[row[0] for row in rows],
-        y=[row[2] for row in rows],
-        hue=[row[1] for row in rows],
+        x=bar_categories,
+        y=bar_values,
+        hue=bar_labels,
         hue_order=list(legend.values()),
         order=categories,
         ax=ax,

benchmarks/compare.py

@@ -54,18 +54,29 @@ def _number(value: float | None) -> str:
     """At least three significant digits, without an exponent."""
     if value is None or value == 0:
         return markdown.number(value, 0)
-    return markdown.number(value, max(0, min(3, 2 - math.floor(math.log10(abs(value))))))
+    magnitude = math.floor(math.log10(abs(value)))
+    digits = max(0, min(3, 2 - magnitude))
+    return markdown.number(value, digits)
 
 
 def _change(metric: Metric, before: float | None, after: float | None) -> tuple[str, str]:
     """After over before, and whether that's better, worse or the same."""
-    if before is None or after is None:
-        return '-', 'only after' if before is None else 'only before'
+    if before is None:
+        return '-', 'only after'
+    if after is None:
+        return '-', 'only before'
     if before == 0:
         return '-', '-'
+
     change = (after - before) / before
+    return f'{change:+.1%}', _verdict(metric, change)
+
+
+def _verdict(metric: Metric, change: float) -> str:
+    if change == 0:
+        return 'same'
     better = change < 0 if metric.lower_is_better else change > 0
-    return f'{change:+.1%}', 'same' if change == 0 else 'better' if better else 'worse'
+    return 'better' if better else 'worse'
 
 
 def _row(metric: Metric, before: float | None, after: float | None, *, with_library: bool) -> list[str]:
@@ -83,7 +94,8 @@ def _row(metric: Metric, before: float | None, after: float | None, *, with_libr
 
 def _suite(suite: str, before: Run, after: Run) -> list[str]:
     """The table of a suite: a row a metric, case and library in either run, grouped by metric."""
-    old, new = _values(suite, before), _values(suite, after)
+    old = _values(suite, before)
+    new = _values(suite, after)
     metrics = list(dict.fromkeys([*old, *new]))
     titles = list(dict.fromkeys(metric.title for metric in metrics))
     metrics.sort(key=lambda metric: titles.index(metric.title))

benchmarks/drivers/aiortc/media.py

@@ -49,7 +49,8 @@ async def recv(self) -> av.AudioFrame:
         """The next frame, once it's due."""
         if self._start is None:
             self._start = time.monotonic()
-        await asyncio.sleep(max(0.0, self._start + self._count * AUDIO_FRAMES / AUDIO_RATE - time.monotonic()))
+        due = self._start + self._count * AUDIO_FRAMES / AUDIO_RATE
+        await asyncio.sleep(max(0.0, due - time.monotonic()))
 
         frame = av.AudioFrame(format='s16', layout='mono', samples=AUDIO_FRAMES)
         frame.planes[0].update(self._chunk)
@@ -69,15 +70,19 @@ class VideoTrack(aiortc.MediaStreamTrack):
     def __init__(self, raw: list[bytes], *, size: tuple[int, int], fps: float, numbered: bool) -> None:
         super().__init__()
         self.tally = frames.Tally()
-        self._raw, self._size, self._fps, self._numbered = raw, size, fps, numbered
+        self._raw = raw
+        self._size = size
+        self._fps = fps
+        self._numbered = numbered
         self._start: float | None = None
 
     async def recv(self) -> av.VideoFrame:
         """The next frame, once it's due."""
         count = self.tally.frames
         if self._start is None:
             self._start = time.monotonic()
-        await asyncio.sleep(max(0.0, self._start + count / self._fps - time.monotonic()))
+        due = self._start + count / self._fps
+        await asyncio.sleep(max(0.0, due - time.monotonic()))
 
         width, height = self._size
         data = self._raw[count % len(self._raw)]
@@ -96,8 +101,11 @@ async def recv(self) -> av.VideoFrame:
 
 def _i420(data: bytes, width: int, height: int) -> av.VideoFrame:
     frame = av.VideoFrame(width, height, 'yuv420p')
-    luma, chroma = width * height, width * height // 4
-    for plane, start, size in zip(frame.planes, (0, luma, luma + chroma), (luma, chroma, chroma)):
+    luma = width * height
+    chroma = luma // 4
+    offsets = (0, luma, luma + chroma)
+    sizes = (luma, chroma, chroma)
+    for plane, start, size in zip(frame.planes, offsets, sizes):
         plane.update(data[start : start + size])
     return frame
 

benchmarks/drivers/base.py

@@ -80,7 +80,8 @@ def _answered(self, number: int) -> None:
     async def flood(self, *, size: int, seconds: float, low: int, high: int) -> None:
         """Sends messages for a while, keeping the buffered amount between the watermarks."""
         self.set_low_threshold(low)
-        payload, end = bytes(size), now() + seconds
+        payload = bytes(size)
+        end = now() + seconds
         slice_end = now() + FLOOD_SLICE
 
         while (remaining := end - now()) > 0:

benchmarks/drivers/frames.py

@@ -17,7 +17,11 @@
 # the frame number, as bits in blocks of luma at the top left of a frame
 BITS = 16
 BLOCK = 16
-LUMA_ONE, LUMA_ZERO, LUMA_THRESHOLD = 235, 16, 125
+#: the rows and the columns of a block that are read
+BLOCK_CENTER = range(5, 11)
+LUMA_ONE = 235
+LUMA_ZERO = 16
+LUMA_THRESHOLD = 125
 TONE = 440.0
 
 
@@ -32,7 +36,8 @@ def numbered(frame: bytes, width: int, number: int) -> bytes:
     """A copy of an I420 frame with the number drawn in its luma."""
     copy = bytearray(frame)
     for bit in range(BITS):
-        block = bytes([LUMA_ONE if number >> bit & 1 != 0 else LUMA_ZERO]) * BLOCK
+        is_one = (number >> bit) & 1 != 0
+        block = bytes([LUMA_ONE if is_one else LUMA_ZERO]) * BLOCK
         for y in range(BLOCK):
             start = y * width + bit * BLOCK
             copy[start : start + BLOCK] = block
@@ -43,7 +48,7 @@ def read_number(luma: bytes | bytearray, stride: int) -> int:
     """The number drawn in a luma plane, read at the centers of the blocks, away from compression blur."""
     number = 0
     for bit in range(BITS):
-        samples = [luma[y * stride + bit * BLOCK + x] for y in range(5, 11) for x in range(5, 11)]
+        samples = [luma[y * stride + bit * BLOCK + x] for y in BLOCK_CENTER for x in BLOCK_CENTER]
         if sum(samples) / len(samples) > LUMA_THRESHOLD:
             number |= 1 << bit
     return number

benchmarks/drivers/wrtc/__main__.py

@@ -103,7 +103,8 @@ async def video_send(
     ) -> Message:
         """Sends the raw frames of a file, encoded in the connection's codec, at most at the bitrate."""
         raw = frames.load_raw(path, width, height)
-        sender = self._senders[pc] = media.VideoSource(raw, size=(width, height), fps=fps, numbered=numbered)
+        sender = media.VideoSource(raw, size=(width, height), fps=fps, numbered=numbered)
+        self._senders[pc] = sender
         self.tallies[pc] = sender.tally
 
         encoding = webrtc.RTCRtpEncodingParameters(max_bitrate=bitrate)

benchmarks/drivers/wrtc/media.py

@@ -18,6 +18,8 @@
 
 AUDIO_RATE = 48_000
 AUDIO_FRAMES = 960
+#: the length of a 20 ms chunk of the tone, in microseconds
+CHUNK_US = 20_000
 #: the MIME type and the parts of the fmtp line of each codec name of the protocol
 CODECS: dict[str, tuple[str, tuple[str, ...]]] = {
     'VP8': ('video/VP8', ()),
@@ -38,22 +40,24 @@ def __init__(self) -> None:
     async def _write(self) -> None:
         writer = self.track.writable.get_writer()
         loop = asyncio.get_running_loop()
-        chunk, start, written = frames.tone(AUDIO_RATE, AUDIO_FRAMES), loop.time(), 0
+        chunk = frames.tone(AUDIO_RATE, AUDIO_FRAMES)
+        start = loop.time()
+        written = 0
 
         while True:
             init = webrtc.AudioDataInit(
                 format='s16',
                 sample_rate=AUDIO_RATE,
                 number_of_frames=AUDIO_FRAMES,
                 number_of_channels=1,
-                timestamp=written * 20_000,
+                timestamp=written * CHUNK_US,
                 data=chunk,
             )
             await writer.write(webrtc.AudioData(init))
             written += 1
 
-            # the next chunk is due 20 ms after the last
-            await asyncio.sleep(max(0.0, start + written / 50 - loop.time()))
+            due = start + written * CHUNK_US / 1_000_000
+            await asyncio.sleep(max(0.0, due - loop.time()))
 
 
 class VideoSource:
@@ -68,7 +72,8 @@ async def _write(self, raw: list[bytes], size: tuple[int, int], fps: float, *, n
         width, height = size
         writer = self.generator.writable.get_writer()
         loop = asyncio.get_running_loop()
-        start, count = loop.time(), 0
+        start = loop.time()
+        count = 0
 
         while True:
             data = raw[count % len(raw)]
@@ -97,11 +102,12 @@ def codecs(codec: str) -> list[webrtc.RTCRtpCodec]:
     mime, parts = CODECS[codec]
     if capabilities is None:
         return []
-    return [c for c in capabilities.codecs if c.mime_type == mime and all(p in _fmtp(c) for p in parts)]
+    return [c for c in capabilities.codecs if c.mime_type == mime and _has_parts(c, parts)]
 
 
-def _fmtp(codec: webrtc.RTCRtpCodec) -> str:
-    return codec.sdp_fmtp_line if codec.sdp_fmtp_line is not None else ''
+def _has_parts(codec: webrtc.RTCRtpCodec, parts: tuple[str, ...]) -> bool:
+    fmtp = codec.sdp_fmtp_line if codec.sdp_fmtp_line is not None else ''
+    return all(part in fmtp for part in parts)
 
 
 async def read_number(frame: webrtc.VideoFrame) -> int:

benchmarks/fixtures.py

@@ -79,7 +79,8 @@ def _generate(path_name: str, arguments: list[str]) -> str:
 def encoded(video: Video, codec: str) -> str:
     """An IVF VP8 or Annex B H.264 file of the video, as Pion sends it unchanged."""
     extension = 'ivf' if codec == 'VP8' else 'h264'
-    return _generate(f'{codec.lower()}-{video.stem}.{extension}', [*_testsrc(video, SECONDS), *_encoder(video, codec)])
+    arguments = [*_testsrc(video, SECONDS), *_encoder(video, codec)]
+    return _generate(f'{codec.lower()}-{video.stem}.{extension}', arguments)
 
 
 def raw(video: Video) -> str:

benchmarks/protocol.py

@@ -190,13 +190,18 @@ def _dispatch(self, driver: object, line: str) -> None:
         task.add_done_callback(self._tasks.discard)
 
     async def _run(self, driver: object, message: Message) -> None:
-        command_id, name = message.pop('id'), str(message.pop('cmd'))
+        command_id = message.pop('id')
+        name = str(message.pop('cmd'))
         if name == 'exit':
             self._send({'id': command_id, 'result': {}})
             self._lines.put_nowait(None)
             return
+
         try:
-            result = await (self._probe() if name == 'probe' else getattr(driver, name)(**message))
+            if name == 'probe':
+                result: object = await self._probe()
+            else:
+                result = await getattr(driver, name)(**message)
         except Exception as error:  # ruff: ignore[blind-except] - the runner reports what failed
             self._send({'id': command_id, 'error': f'{type(error).__name__}: {error}'})
         else:

benchmarks/report.py

@@ -97,25 +97,35 @@ def newest(paths: Iterable[pathlib.Path]) -> dict[str, Newest]:
         with_suite = [run for run in runs if name in run['suites']]
         full = [run for run in with_suite if not run['manifest']['quick']]
         chosen = full if len(full) != 0 else with_suite
-        if len(chosen) == 0:
-            continue
-
-        data: object = None
-        versions: dict[str, Hello] = {}
-        given: dict[str, int] = {}
-        for index, run in enumerate(chosen):
-            part = run['suites'][name]
-            data = part if data is None else suite.merge(data, part)
-            versions.update(run['manifest']['libraries'])
-            given.update(dict.fromkeys(suite.libraries(part), index))
-
-        sources = [
-            (run['manifest'], [lib for lib, at in given.items() if at == index]) for index, run in enumerate(chosen)
-        ]
-        found[name] = Newest(data, versions, [(manifest, libs) for manifest, libs in sources if len(libs) != 0])
+        if len(chosen) != 0:
+            found[name] = _merged(name, suite, chosen)
     return found
 
 
+def _merged(name: str, suite: Suite, runs: list[Run]) -> Newest:
+    """The suite's results of the runs, oldest first, each library's from the newest run that has it."""
+    data: object = None
+    versions: dict[str, Hello] = {}
+    given: dict[str, int] = {}
+    for index, run in enumerate(runs):
+        part = run['suites'][name]
+        data = part if data is None else suite.merge(data, part)
+        versions.update(run['manifest']['libraries'])
+        given.update(dict.fromkeys(suite.libraries(part), index))
+
+    sources: list[tuple[Manifest, list[str]]] = []
+    for index, run in enumerate(runs):
+        libraries = [library for library, at in given.items() if at == index]
+        if len(libraries) != 0:
+            sources.append((run['manifest'], libraries))
+    return Newest(data, versions, sources)
+
+
+def _run_details(row: list[str]) -> list[str]:
+    """The date, the runs, the machine and the commit of a row of the runs table."""
+    return row[2:]
+
+
 def _run_rows(name: str, found: Newest) -> list[list[str]]:
     rows: list[list[str]] = []
     for manifest, libraries in found.sources:
@@ -137,9 +147,9 @@ def render(paths: Iterable[pathlib.Path]) -> str:
         f"Medians; % columns are of {markdown.BASELINE}'s. How: [README](README.md#method).",
         '',
     ]
-    if len({tuple(row[2:]) for row in runs}) == 1:
+    if len({tuple(_run_details(row)) for row in runs}) == 1:
         # one run for every suite and library: a line
-        date, repeats, machine, commit = runs[0][2:]
+        date, repeats, machine, commit = _run_details(runs[0])
         lines += [f'{date}, {repeats} runs a case, {machine}, commit {commit}.', '']
     else:
         lines += markdown.table(['Suite', 'Libraries', 'Date', 'Runs', 'Machine', 'Commit'], runs)