tiphys.ipc.client
IPC Client implementation.
Client for communicating with the host via IPC.
1""" 2IPC Client implementation. 3 4Client for communicating with the host via IPC. 5""" 6 7import asyncio 8import json 9from typing import Any 10 11from tiphys.ipc.protocol import IPCRequest, IPCResponse 12 13 14class IPCClient: 15 """ 16 Client for connecting to the local IPC server. 17 """ 18 19 def __init__(self, socket_path: str = "/tmp/tiphys.sock"): 20 self.socket_path = socket_path 21 self._reader: asyncio.StreamReader | None = None 22 self._writer: asyncio.StreamWriter | None = None 23 24 async def connect(self) -> None: 25 """Connect to the IPC server.""" 26 self._reader, self._writer = await asyncio.open_unix_connection(self.socket_path) 27 28 async def close(self) -> None: 29 """Close the connection.""" 30 if self._writer: 31 self._writer.close() 32 await self._writer.wait_closed() 33 self._writer = None 34 self._reader = None 35 36 async def call(self, method: str, **params: Any) -> Any: 37 """ 38 Call a method on the host. 39 40 Args: 41 method: The method name. 42 **params: Arguments for the method. 43 44 Returns: 45 The result of the call. 46 47 Raises: 48 RuntimeError: If the call fails or returns an error. 49 """ 50 if not self._writer or not self._reader: 51 await self.connect() 52 53 request = IPCRequest(method=method, params=params) 54 55 assert self._writer is not None 56 assert self._reader is not None 57 58 # Send request 59 data = request.model_dump_json() + "\n" 60 self._writer.write(data.encode()) 61 await self._writer.drain() 62 63 # Wait for response 64 response_line = await self._reader.readline() 65 if not response_line: 66 raise RuntimeError("Connection closed by server") 67 68 try: 69 response_dict = json.loads(response_line.decode().strip()) 70 response = IPCResponse(**response_dict) 71 72 if response.error: 73 raise RuntimeError(f"IPC Error: {response.error}") 74 75 return response.result 76 77 except json.JSONDecodeError as exc: 78 raise RuntimeError("Invalid response from server") from exc
class
IPCClient:
15class IPCClient: 16 """ 17 Client for connecting to the local IPC server. 18 """ 19 20 def __init__(self, socket_path: str = "/tmp/tiphys.sock"): 21 self.socket_path = socket_path 22 self._reader: asyncio.StreamReader | None = None 23 self._writer: asyncio.StreamWriter | None = None 24 25 async def connect(self) -> None: 26 """Connect to the IPC server.""" 27 self._reader, self._writer = await asyncio.open_unix_connection(self.socket_path) 28 29 async def close(self) -> None: 30 """Close the connection.""" 31 if self._writer: 32 self._writer.close() 33 await self._writer.wait_closed() 34 self._writer = None 35 self._reader = None 36 37 async def call(self, method: str, **params: Any) -> Any: 38 """ 39 Call a method on the host. 40 41 Args: 42 method: The method name. 43 **params: Arguments for the method. 44 45 Returns: 46 The result of the call. 47 48 Raises: 49 RuntimeError: If the call fails or returns an error. 50 """ 51 if not self._writer or not self._reader: 52 await self.connect() 53 54 request = IPCRequest(method=method, params=params) 55 56 assert self._writer is not None 57 assert self._reader is not None 58 59 # Send request 60 data = request.model_dump_json() + "\n" 61 self._writer.write(data.encode()) 62 await self._writer.drain() 63 64 # Wait for response 65 response_line = await self._reader.readline() 66 if not response_line: 67 raise RuntimeError("Connection closed by server") 68 69 try: 70 response_dict = json.loads(response_line.decode().strip()) 71 response = IPCResponse(**response_dict) 72 73 if response.error: 74 raise RuntimeError(f"IPC Error: {response.error}") 75 76 return response.result 77 78 except json.JSONDecodeError as exc: 79 raise RuntimeError("Invalid response from server") from exc
Client for connecting to the local IPC server.
async def
connect(self) -> None:
25 async def connect(self) -> None: 26 """Connect to the IPC server.""" 27 self._reader, self._writer = await asyncio.open_unix_connection(self.socket_path)
Connect to the IPC server.
async def
close(self) -> None:
29 async def close(self) -> None: 30 """Close the connection.""" 31 if self._writer: 32 self._writer.close() 33 await self._writer.wait_closed() 34 self._writer = None 35 self._reader = None
Close the connection.
async def
call(self, method: str, **params: Any) -> Any:
37 async def call(self, method: str, **params: Any) -> Any: 38 """ 39 Call a method on the host. 40 41 Args: 42 method: The method name. 43 **params: Arguments for the method. 44 45 Returns: 46 The result of the call. 47 48 Raises: 49 RuntimeError: If the call fails or returns an error. 50 """ 51 if not self._writer or not self._reader: 52 await self.connect() 53 54 request = IPCRequest(method=method, params=params) 55 56 assert self._writer is not None 57 assert self._reader is not None 58 59 # Send request 60 data = request.model_dump_json() + "\n" 61 self._writer.write(data.encode()) 62 await self._writer.drain() 63 64 # Wait for response 65 response_line = await self._reader.readline() 66 if not response_line: 67 raise RuntimeError("Connection closed by server") 68 69 try: 70 response_dict = json.loads(response_line.decode().strip()) 71 response = IPCResponse(**response_dict) 72 73 if response.error: 74 raise RuntimeError(f"IPC Error: {response.error}") 75 76 return response.result 77 78 except json.JSONDecodeError as exc: 79 raise RuntimeError("Invalid response from server") from exc
Call a method on the host.
Args: method: The method name. **params: Arguments for the method.
Returns: The result of the call.
Raises: RuntimeError: If the call fails or returns an error.