|
6 | 6 | import logging |
7 | 7 | import os |
8 | 8 | import struct |
9 | | -from typing import Awaitable, Optional, TypeVar |
| 9 | +from typing import Awaitable, List, Optional, TypeVar |
10 | 10 |
|
11 | 11 | import reactivex.operators as op |
12 | 12 | import semver |
@@ -93,6 +93,16 @@ def __init__(self): |
93 | 93 | # REVISIT: this can potentially waste a lot of RAM if not drained |
94 | 94 | self._stdout_line_queue = asyncio.Queue() |
95 | 95 |
|
| 96 | + # REVISIT: It would be better to be able to subscribe to output instead |
| 97 | + # of always capturing it even if it is not used. This is currently |
| 98 | + # used in motor test code in pybricks-micropython. |
| 99 | + self.output: List[bytes] = [] |
| 100 | + """ |
| 101 | + Contains lines printed to stdout of the hub as a a list of bytes. |
| 102 | +
|
| 103 | + List is reset each time :meth:`run()` is called. |
| 104 | + """ |
| 105 | + |
96 | 106 | # prior to Pybricks Profile v1.3.0, NUS was used for stdio |
97 | 107 | self._legacy_stdio = False |
98 | 108 |
|
@@ -148,6 +158,8 @@ def _line_handler(self, line: bytes) -> None: |
148 | 158 | print(line_str, file=self.log_file) |
149 | 159 | return |
150 | 160 |
|
| 161 | + self.output.append(line) |
| 162 | + |
151 | 163 | if self.print_output: |
152 | 164 | print(line_str) |
153 | 165 | return |
@@ -425,6 +437,7 @@ async def run( |
425 | 437 |
|
426 | 438 | # Reset output buffer |
427 | 439 | self.log_file = None |
| 440 | + self.output = [] |
428 | 441 | self._stdout_buf.clear() |
429 | 442 | self._stdout_line_queue = asyncio.Queue() |
430 | 443 | self.print_output = print_output |
|
0 commit comments