|
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