Creates a PTY pair: test writes to master_fd, Serial Studio opens slave path.
| 26 | |
| 27 | |
| 28 | class VirtualSerialPort: |
| 29 | """ |
| 30 | Creates a PTY pair: test writes to master_fd, Serial Studio opens slave path. |
| 31 | """ |
| 32 | |
| 33 | def __init__(self, name: str = "VirtualPort"): |
| 34 | self.name = name |
| 35 | self._master_fd: Optional[int] = None |
| 36 | self._slave_fd: Optional[int] = None |
| 37 | self._device_path: Optional[str] = None |
| 38 | self._closed = True |
| 39 | |
| 40 | def open(self) -> None: |
| 41 | """Create the PTY pair and configure raw mode.""" |
| 42 | self._master_fd, self._slave_fd = pty.openpty() |
| 43 | tty.setraw(self._master_fd) |
| 44 | tty.setraw(self._slave_fd) |
| 45 | self._device_path = os.ttyname(self._slave_fd) |
| 46 | self._closed = False |
| 47 | |
| 48 | def close(self) -> None: |
| 49 | """Close both file descriptors safely.""" |
| 50 | if self._closed: |
| 51 | return |
| 52 | |
| 53 | self._closed = True |
| 54 | for fd in (self._master_fd, self._slave_fd): |
| 55 | if fd is not None: |
| 56 | try: |
| 57 | os.close(fd) |
| 58 | except OSError: |
| 59 | pass |
| 60 | |
| 61 | self._master_fd = None |
| 62 | self._slave_fd = None |
| 63 | |
| 64 | @property |
| 65 | def device_path(self) -> str: |
| 66 | """Device path that Serial Studio should connect to.""" |
| 67 | if self._device_path is None: |
| 68 | raise RuntimeError(f"{self.name}: PTY not open, call open() first") |
| 69 | return self._device_path |
| 70 | |
| 71 | @property |
| 72 | def is_open(self) -> bool: |
| 73 | return not self._closed |
| 74 | |
| 75 | def write(self, data: bytes) -> int: |
| 76 | """Write raw bytes to the master side of the PTY.""" |
| 77 | if self._master_fd is None or self._closed: |
| 78 | raise RuntimeError(f"{self.name}: PTY not open") |
| 79 | return os.write(self._master_fd, data) |
| 80 | |
| 81 | def write_csv_frame( |
| 82 | self, |
| 83 | values: list, |
| 84 | start_delimiter: str = "/*", |
| 85 | end_delimiter: str = "*/", |