import asyncio import json from defence360agent.contracts.config import ( GENERIC_SENSOR_SOCKET_PATH, ) async def send_to_socket(msg, timeout=15, wait_for_response=True): result = b"" try: reader, writer = await asyncio.open_unix_connection( path=GENERIC_SENSOR_SOCKET_PATH ) except ConnectionRefusedError: return "Failed to send to socket, check your socket active" except FileNotFoundError: return "Failed to send to socket, socket file not found" except OSError as e: return f"OSError when connectiong to {GENERIC_SENSOR_SOCKET_PATH}: {e}" try: writer.write(json.dumps(msg).encode() + b"\n") await writer.drain() if not wait_for_response: return {} while True: data = await asyncio.wait_for(reader.read(8192), timeout=timeout) if not data: break result += data if b"\n" in data: break return json.loads(result.decode()) except json.JSONDecodeError: return "Failed to decode json answer from a plugin" except ( asyncio.TimeoutError, asyncio.IncompleteReadError, asyncio.CancelledError, BrokenPipeError, ConnectionResetError, OSError, ) as e: return ( result.decode() if result else ( "Failed to process operation with socket. Reason:" f" {type(e).__name__}" ) ) finally: writer.close() await writer.wait_closed()