Client for communicating with Serial Studio API Server.
| 242 | |
| 243 | |
| 244 | class SerialStudioAPI: |
| 245 | """Client for communicating with Serial Studio API Server.""" |
| 246 | |
| 247 | def __init__( |
| 248 | self, host: str = DEFAULT_HOST, port: int = DEFAULT_PORT, verbose: bool = False |
| 249 | ): |
| 250 | self.host = host |
| 251 | self.port = port |
| 252 | self.verbose = verbose |
| 253 | self.socket: Optional[socket.socket] = None |
| 254 | self.receive_buffer = b"" |
| 255 | |
| 256 | def connect(self) -> bool: |
| 257 | try: |
| 258 | self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) |
| 259 | self.socket.settimeout(SOCKET_TIMEOUT) |
| 260 | self.socket.connect((self.host, self.port)) |
| 261 | if self.verbose: |
| 262 | print(f"[INFO] Connected to {self.host}:{self.port}") |
| 263 | return True |
| 264 | except Exception as e: |
| 265 | print(f"[ERROR] Failed to connect: {e}") |
| 266 | return False |
| 267 | |
| 268 | def disconnect(self): |
| 269 | if self.socket: |
| 270 | try: |
| 271 | self.socket.close() |
| 272 | except Exception: |
| 273 | pass |
| 274 | self.socket = None |
| 275 | self.receive_buffer = b"" |
| 276 | |
| 277 | def recv_message(self, timeout: float = SOCKET_TIMEOUT) -> Optional[dict]: |
| 278 | """Receive a newline-delimited JSON message. Returns None on timeout/error.""" |
| 279 | if not self.socket: |
| 280 | return None |
| 281 | |
| 282 | try: |
| 283 | end_time = time.time() + timeout |
| 284 | while True: |
| 285 | newline_pos = self.receive_buffer.find(b"\n") |
| 286 | if newline_pos != -1: |
| 287 | line = self.receive_buffer[:newline_pos] |
| 288 | self.receive_buffer = self.receive_buffer[newline_pos + 1 :] |
| 289 | |
| 290 | if line.strip(): |
| 291 | if self.verbose: |
| 292 | print(f"[RECV] {line.decode('utf-8', errors='replace')}") |
| 293 | return json.loads(line.decode("utf-8")) |
| 294 | continue |
| 295 | |
| 296 | remaining_time = end_time - time.time() |
| 297 | if remaining_time <= 0: |
| 298 | return None |
| 299 | |
| 300 | ready = select.select([self.socket], [], [], min(remaining_time, 0.1)) |
| 301 | if ready[0]: |