diff --git a/robotics_application_manager/comms/__init__.py b/robotics_application_manager/comms/__init__.py index 939cc1a..efa404f 100644 --- a/robotics_application_manager/comms/__init__.py +++ b/robotics_application_manager/comms/__init__.py @@ -5,7 +5,6 @@ between the manager backend and client applications. """ -from .new_consumer import ManagerConsumer +from .consumer import ManagerConsumer from .consumer_message import ManagerConsumerMessageException, ManagerConsumerMessage -from .thread import ThreadWithLoggedException, WebsocketServerThread -from .websocket_server import WebsocketServer +from .thread import ThreadWithLoggedException diff --git a/robotics_application_manager/comms/consumer.py b/robotics_application_manager/comms/consumer.py new file mode 100644 index 0000000..0e150f0 --- /dev/null +++ b/robotics_application_manager/comms/consumer.py @@ -0,0 +1,126 @@ +""" +WebSocket consumer module for the Robotics Application Manager (RAM). + +Handles client connections, message processing, and communication with manager queue. +""" + +import json +import logging +from queue import Queue +from uuid import uuid4 +from datetime import datetime + +from .consumer_message import ( + ManagerConsumerMessageException, + ManagerConsumerMessage, +) +from robotics_application_manager import LogManager +from websockets.asyncio.server import serve +from websockets.exceptions import ConnectionClosedOK +import asyncio + +class ManagerConsumer: + """ + Websocket server consumer for new Robotics Application Manager aka: RAM. + + Supports single client connection to RAM + TODO: Better handling of single client connections, closing and redirecting + """ + def __init__(self, host, port, process_callback): + + """ + Initialize the ManagerConsumer with host, port, and manager_queue. + + Args: + host (str): The host address for the WebSocket server. + port (int): The port number for the WebSocket server. + manager_queue (Queue): The queue for communication with the manager. + """ + self.host = host + self.port = port + self.client = None + self.process_callback = process_callback + self.server_task = None + + # Configurar el logger de websocket_server para salida a consola + self.ws_logger = logging.getLogger("websocket_server.websocket_server") + self.ws_logger.propagate = False + self.ws_logger.setLevel(logging.INFO) + self.ws_logger.handlers.clear() + ws_formatter = logging.Formatter( + "%(asctime)s [%(threadName)-12.12s] [%(levelname)-5.5s] " + "(%(name)s) %(message)s", + "%H:%M:%S", + ) + ws_console_handler = logging.StreamHandler() + ws_console_handler.setFormatter(ws_formatter) + self.ws_logger.addHandler(ws_console_handler) + + # Use the __await__ method to make the class awaitable + def __await__(self): + # Call ls the constructor and returns the instance + return self.create().__await__() + + # A method that creates an instance of the class asynchronously + async def create(self): + self.server = await serve(self.manage_conection, self.host, self.port, start_serving=False, max_size=None, logger=self.ws_logger) + return self + + async def manage_conection(self, websocket): + if self.client is not None: + await websocket.close() + return + + try: + self.client = websocket # Register client + LogManager.logger.info(f"client connected: {self.client}") + + await self.process_msg(websocket) + except ConnectionClosedOK as e: + pass + finally: + now = datetime.now() + time_str = now.strftime("%H:%M:%S") + LogManager.logger.info(f"Client disconnected {time_str}: {self.client}") + self.client = None + + async def process_msg(self, websocket): + async for raw_msg in websocket: + try: + json_msg = json.loads(raw_msg) + await self.process_callback(ManagerConsumerMessage(**json_msg)) + except Exception as e: + ex = ManagerConsumerMessageException(id=str(uuid4()), message=str(e)) + await self.send_message(ex) + LogManager.logger.error(e, exc_info=True) + + async def send_message(self, message_data, command=None): + """ + Send a message to the connected client. + + Args: + message_data: The message data to send, can be a ManagerConsumerMessage, + ManagerConsumerMessageException, or other data. + command (str, optional): The command associated with the message, + used if message_data is not a ManagerConsumerMessage. + """ + if self.client is not None and self.server is not None: + if isinstance(message_data, ManagerConsumerMessage): + message = message_data + elif isinstance(message_data, ManagerConsumerMessageException): + message = message_data.consumer_message() + else: + message = ManagerConsumerMessage( + id=str(uuid4()), command=command, data=message_data + ) + + await self.client.send(str(message)) + + async def start(self): + """Start the WebSocket server in a separate thread.""" + await self.server.start_serving() + + async def stop(self): + """Stop the WebSocket server gracefully.""" + self.server.close() + await self.server.wait_closed() diff --git a/robotics_application_manager/comms/new_consumer.py b/robotics_application_manager/comms/new_consumer.py deleted file mode 100644 index 0f2568c..0000000 --- a/robotics_application_manager/comms/new_consumer.py +++ /dev/null @@ -1,151 +0,0 @@ -""" -WebSocket consumer module for the Robotics Application Manager (RAM). - -Handles client connections, message processing, and communication with manager queue. -""" - -import json -import logging -from queue import Queue -from uuid import uuid4 -from datetime import datetime - -from .consumer_message import ( - ManagerConsumerMessageException, - ManagerConsumerMessage, -) -from .websocket_server import WebsocketServer -from robotics_application_manager import LogManager - - -class ManagerConsumer: - """ - Websocket server consumer for new Robotics Application Manager aka: RAM. - - Supports single client connection to RAM - TODO: Better handling of single client connections, closing and redirecting - """ - - def __init__(self, host, port, manager_queue: Queue): - """ - Initialize the ManagerConsumer with host, port, and manager_queue. - - Args: - host (str): The host address for the WebSocket server. - port (int): The port number for the WebSocket server. - manager_queue (Queue): The queue for communication with the manager. - """ - self.host = host - self.port = port - self.server = WebsocketServer(host=host, port=port, loglevel=logging.INFO) - - # Configurar el logger de websocket_server para salida a consola - ws_logger = logging.getLogger("websocket_server.websocket_server") - ws_logger.propagate = False - ws_logger.setLevel(logging.INFO) - ws_logger.handlers.clear() - ws_formatter = logging.Formatter( - "%(asctime)s [%(threadName)-12.12s] [%(levelname)-5.5s] " - "(%(name)s) %(message)s", - "%H:%M:%S", - ) - ws_console_handler = logging.StreamHandler() - ws_console_handler.setFormatter(ws_formatter) - ws_logger.addHandler(ws_console_handler) - - self.server.set_fn_new_client(self.handle_client_new) - self.server.set_fn_client_left(self.handle_client_disconnect) - self.server.set_fn_message_received(self.handle_message_received) - self.client = None - self.manager_queue = manager_queue - - def handle_client_new(self, client, server): - """ - Handle a new client connection event. - - Args: - client: The client object representing the connected client. - server: The WebSocket server instance. - """ - LogManager.logger.info(f"client connected: {client}") - self.client = client - self.server.deny_new_connections() - - def handle_client_disconnect(self, client, server): - """ - Handle a client disconnection event. - - Args: - client: The client object representing the disconnected client. - server: The WebSocket server instance. - """ - if client is None: - return - LogManager.logger.info(f"client disconnected: {client}") - now = datetime.now() - time_string = now.strftime("%H:%M:%S") - print(time_string) - message = ManagerConsumerMessage( - **{"id": str(uuid4()), "command": "disconnect"} - ) - self.manager_queue.put(message) - self.client = None - self.server.allow_new_connections() - - def handle_message_received(self, client, server, websocket_message): - """ - Handle a message received from a client. - - Args: - client: The client object that sent the message. - server: The WebSocket server instance. - websocket_message (str): The message received from the client. - """ - LogManager.logger.info( - f"message received length: {len(websocket_message)} from client {client}" - ) - LogManager.logger.info( - f"message received: {websocket_message} from client {client}" - ) - message = None - try: - s = json.loads(websocket_message) - message = ManagerConsumerMessage(**s) - self.manager_queue.put(message) - except Exception as e: - if message is not None: - ex = ManagerConsumerMessageException(id=message.id, message=str(e)) - else: - ex = ManagerConsumerMessageException(id=str(uuid4()), message=str(e)) - self.server.send_message(client, str(ex)) - raise e - - def send_message(self, message_data, command=None): - """ - Send a message to the connected client. - - Args: - message_data: The message data to send, can be a ManagerConsumerMessage, - ManagerConsumerMessageException, or other data. - command (str, optional): The command associated with the message, - used if message_data is not a ManagerConsumerMessage. - """ - if self.client is not None and self.server is not None: - if isinstance(message_data, ManagerConsumerMessage): - message = message_data - elif isinstance(message_data, ManagerConsumerMessageException): - message = message_data.consumer_message() - else: - message = ManagerConsumerMessage( - id=str(uuid4()), command=command, data=message_data - ) - - self.server.send_message(self.client, str(message)) - - def start(self): - """Start the WebSocket server in a separate thread.""" - self.server.run_forever(threaded=True) - - def stop(self): - """Stop the WebSocket server gracefully.""" - self.server.shutdown_gracefully() diff --git a/robotics_application_manager/comms/thread.py b/robotics_application_manager/comms/thread.py index 4477895..33a2d54 100644 --- a/robotics_application_manager/comms/thread.py +++ b/robotics_application_manager/comms/thread.py @@ -37,8 +37,3 @@ def run(self): finally: del self._target, self._args, self._kwargs - -class WebsocketServerThread(ThreadWithLoggedException): - """Dummy wrapper to make debug messages a bit more readable""" - - pass diff --git a/robotics_application_manager/comms/websocket_server.py b/robotics_application_manager/comms/websocket_server.py deleted file mode 100644 index 0a67118..0000000 --- a/robotics_application_manager/comms/websocket_server.py +++ /dev/null @@ -1,537 +0,0 @@ -# Author: Johan Hanssen Seferidis -# License: MIT -# From https://github.com/Pithikos/python-websocket-server -# To fix OPCODE_CONTINUATION we have to apply this pacth https://github.com/Pithikos/python-websocket-server/pull/70/files. That is why we use this instead of the package - -import sys -import struct -import ssl -from base64 import b64encode -from hashlib import sha1 -import logging -from socket import error as SocketError -import errno -import threading -from socketserver import ThreadingMixIn, TCPServer, StreamRequestHandler - -from .thread import WebsocketServerThread - -logger = logging.getLogger(__name__) -logging.basicConfig() - -""" -+-+-+-+-+-------+-+-------------+-------------------------------+ - 0 1 2 3 - 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 -+-+-+-+-+-------+-+-------------+-------------------------------+ -|F|R|R|R| opcode|M| Payload len | Extended payload length | -|I|S|S|S| (4) |A| (7) | (16/64) | -|N|V|V|V| |S| | (if payload len==126/127) | -| |1|2|3| |K| | | -+-+-+-+-+-------+-+-------------+ - - - - - - - - - - - - - - - + -| Extended payload length continued, if payload len == 127 | -+ - - - - - - - - - - - - - - - +-------------------------------+ -| Payload Data continued ... | -+---------------------------------------------------------------+ -""" - -FIN = 0x80 -OPCODE = 0x0F -MASKED = 0x80 -PAYLOAD_LEN = 0x7F -PAYLOAD_LEN_EXT16 = 0x7E -PAYLOAD_LEN_EXT64 = 0x7F - -OPCODE_CONTINUATION = 0x0 -OPCODE_TEXT = 0x1 -OPCODE_BINARY = 0x2 -OPCODE_CLOSE_CONN = 0x8 -OPCODE_PING = 0x9 -OPCODE_PONG = 0xA - -CLOSE_STATUS_NORMAL = 1000 -DEFAULT_CLOSE_REASON = bytes("", encoding="utf-8") - - -class API: - def run_forever(self, threaded=False): - return self._run_forever(threaded) - - def new_client(self, client, server): - pass - - def client_left(self, client, server): - pass - - def message_received(self, client, server, message): - pass - - def set_fn_new_client(self, fn): - self.new_client = fn - - def set_fn_client_left(self, fn): - self.client_left = fn - - def set_fn_message_received(self, fn): - self.message_received = fn - - def send_message(self, client, msg): - self._unicast(client, msg) - - def send_message_to_all(self, msg): - self._multicast(msg) - - def deny_new_connections( - self, status=CLOSE_STATUS_NORMAL, reason=DEFAULT_CLOSE_REASON - ): - self._deny_new_connections(status, reason) - - def allow_new_connections(self): - self._allow_new_connections() - - def shutdown_gracefully( - self, status=CLOSE_STATUS_NORMAL, reason=DEFAULT_CLOSE_REASON - ): - self._shutdown_gracefully(status, reason) - - def shutdown_abruptly(self): - self._shutdown_abruptly() - - def disconnect_clients_gracefully( - self, status=CLOSE_STATUS_NORMAL, reason=DEFAULT_CLOSE_REASON - ): - self._disconnect_clients_gracefully(status, reason) - - def disconnect_clients_abruptly(self): - self._disconnect_clients_abruptly() - - -class WebsocketServer(ThreadingMixIn, TCPServer, API): - """ - A websocket server waiting for clients to connect. - - Args: - port(int): Port to bind to - host(str): Hostname or IP to listen for connections. By default 127.0.0.1 - is being used. To accept connections from any client, you should use - 0.0.0.0. - loglevel: Logging level from logging module to use for logging. By default - warnings and errors are being logged. - - Properties: - clients(list): A list of connected clients. A client is a dictionary - like below. - { - 'id' : id, - 'handler' : handler, - 'address' : (addr, port) - } - """ - - allow_reuse_address = True - daemon_threads = True # comment to keep threads alive until finished - - def __init__( - self, host="127.0.0.1", port=0, loglevel=logging.WARNING, key=None, cert=None - ): - logger.setLevel(loglevel) - TCPServer.__init__(self, (host, port), WebSocketHandler) - self.host = host - self.port = self.socket.getsockname()[1] - - self.key = key - self.cert = cert - - self.clients = [] - self.id_counter = 0 - self.thread = None - - self._deny_clients = False - - def _run_forever(self, threaded): - cls_name = self.__class__.__name__ - try: - logger.info("Listening on port %d for clients.." % self.port) - if threaded: - self.daemon = True - self.thread = WebsocketServerThread( - target=super().serve_forever, daemon=True, logger=logger - ) - logger.info(f"Starting {cls_name} on thread {self.thread.getName()}.") - self.thread.start() - else: - self.thread = threading.current_thread() - logger.info(f"Starting {cls_name} on main thread.") - super().serve_forever() - except KeyboardInterrupt: - self.server_close() - logger.info("Server terminated.") - except Exception as e: - logger.error(str(e), exc_info=True) - sys.exit(1) - - def _message_received_(self, handler, msg): - self.message_received(self.handler_to_client(handler), self, msg) - - def _ping_received_(self, handler, msg): - handler.send_pong(msg) - - def _pong_received_(self, handler, msg): - pass - - def _new_client_(self, handler): - if self._deny_clients: - status = self._deny_clients["status"] - reason = self._deny_clients["reason"] - handler.send_close(status, reason) - self._terminate_client_handler(handler) - return - - self.id_counter += 1 - client = { - "id": self.id_counter, - "handler": handler, - "address": handler.client_address, - } - self.clients.append(client) - self.new_client(client, self) - - def _client_left_(self, handler): - client = self.handler_to_client(handler) - self.client_left(client, self) - if client in self.clients: - self.clients.remove(client) - - def _unicast(self, receiver_client, msg): - receiver_client["handler"].send_message(msg) - - def _multicast(self, msg): - for client in self.clients: - self._unicast(client, msg) - - def handler_to_client(self, handler): - for client in self.clients: - if client["handler"] == handler: - return client - - def _terminate_client_handler(self, handler): - handler.keep_alive = False - handler.finish() - handler.connection.close() - - def _terminate_client_handlers(self): - """ - Ensures request handler for each client is terminated correctly - """ - for client in self.clients: - self._terminate_client_handler(client["handler"]) - - def _shutdown_gracefully( - self, status=CLOSE_STATUS_NORMAL, reason=DEFAULT_CLOSE_REASON - ): - """ - Send a CLOSE handshake to all connected clients before terminating server - """ - self.keep_alive = False - self._disconnect_clients_gracefully(status, reason) - self.server_close() - self.shutdown() - - def _shutdown_abruptly(self): - """ - Terminate server without sending a CLOSE handshake - """ - self.keep_alive = False - self._disconnect_clients_abruptly() - self.server_close() - self.shutdown() - - def _disconnect_clients_gracefully( - self, status=CLOSE_STATUS_NORMAL, reason=DEFAULT_CLOSE_REASON - ): - """ - Terminate clients gracefully without shutting down the server - """ - for client in self.clients: - client["handler"].send_close(status, reason) - self._terminate_client_handlers() - - def _disconnect_clients_abruptly(self): - """ - Terminate clients abruptly (no CLOSE handshake) without shutting down the server - """ - self._terminate_client_handlers() - - def _deny_new_connections(self, status, reason): - self._deny_clients = { - "status": status, - "reason": reason, - } - - def _allow_new_connections(self): - self._deny_clients = False - - -class WebSocketHandler(StreamRequestHandler): - def __init__(self, socket, addr, server): - self.server = server - assert not hasattr(self, "_send_lock"), "_send_lock already exists" - self._send_lock = threading.Lock() - if server.key and server.cert: - try: - socket = ssl.wrap_socket( - socket, server_side=True, certfile=server.cert, keyfile=server.key - ) - except: # Not sure which exception it throws if the key/cert isn't found - logger.warning( - "SSL not available (are the paths {} and {} correct for the key and cert?)".format( - server.key, server.cert - ) - ) - StreamRequestHandler.__init__(self, socket, addr, server) - - def setup(self): - StreamRequestHandler.setup(self) - self.keep_alive = True - self.handshake_done = False - self.valid_client = False - self.fragment_opcode = 0 - self.fragment_payload_buf = bytearray() - - def handle(self): - while self.keep_alive: - if not self.handshake_done: - self.handshake() - elif self.valid_client: - self.read_next_message() - - def read_bytes(self, num): - return self.rfile.read(num) - - def read_next_message(self): - try: - b1, b2 = self.read_bytes(2) - except SocketError as e: # to be replaced with ConnectionResetError for py3 - if e.errno == errno.ECONNRESET: - logger.info("Client closed connection.") - self.keep_alive = 0 - return - b1, b2 = 0, 0 - except ValueError as e: - b1, b2 = 0, 0 - - fin = b1 & FIN - opcode = b1 & OPCODE - masked = b2 & MASKED - payload_length = b2 & PAYLOAD_LEN - - if opcode == OPCODE_CLOSE_CONN: - logger.info("Client asked to close connection.") - self.keep_alive = 0 - return - if not masked: - logger.warning("Client must always be masked.") - self.keep_alive = 0 - return - if opcode == OPCODE_BINARY: - logger.warning("Binary frames are not supported.") - return - # else: - # logger.warning("Unknown opcode %#x." % opcode) - # self.keep_alive = 0 - # return - - if payload_length == 126: - payload_length = struct.unpack(">H", self.rfile.read(2))[0] - elif payload_length == 127: - payload_length = struct.unpack(">Q", self.rfile.read(8))[0] - - masks = self.read_bytes(4) - payload = bytearray() - - for message_byte in self.read_bytes(payload_length): - message_byte ^= masks[len(payload) % 4] - payload.append(message_byte) - - if fin and opcode != OPCODE_CONTINUATION: # simple msg - if opcode == OPCODE_PING: - self.server._ping_received_(self, payload.decode("utf8")) - elif opcode == OPCODE_PONG: - self.server._pong_received_(self, payload.decode("utf8")) - elif opcode == OPCODE_TEXT: - self.server._message_received_(self, payload.decode("utf8")) - return - - if not fin and opcode: # fragment msg start - self.fragment_opcode = opcode - self.fragment_payload_buf = payload - return - - # "not opcode" is the same as "opcode == OPCODE_CONTINUATION" - if not fin and not opcode: # fragment msg ing - self.fragment_payload_buf.extend(payload) - return - - if fin and opcode == OPCODE_CONTINUATION: # fragment msg end - if self.fragment_opcode == OPCODE_TEXT: - self.server._message_received_( - self, (self.fragment_payload_buf + payload).decode("utf8") - ) - elif self.fragment_opcode == OPCODE_BINARY: - pass - return - - def send_message(self, message): - self.send_text(message) - - def send_pong(self, message): - self.send_text(message, OPCODE_PONG) - - def send_close(self, status=CLOSE_STATUS_NORMAL, reason=DEFAULT_CLOSE_REASON): - """ - Send CLOSE to client - - Args: - status: Status as defined in https://datatracker.ietf.org/doc/html/rfc6455#section-7.4.1 - reason: Text with reason of closing the connection - """ - if status < CLOSE_STATUS_NORMAL or status > 1015: - raise Exception(f"CLOSE status must be between 1000 and 1015, got {status}") - - header = bytearray() - payload = struct.pack("!H", status) + reason - payload_length = len(payload) - assert ( - payload_length <= 125 - ), "We only support short closing reasons at the moment" - - # Send CLOSE with status & reason - header.append(FIN | OPCODE_CLOSE_CONN) - header.append(payload_length) - with self._send_lock: - self.request.send(header + payload) - - def send_text(self, message, opcode=OPCODE_TEXT): - """ - Important: Fragmented(=continuation) messages are not supported since - their usage cases are limited - when we don't know the payload length. - """ - - # Validate message - if isinstance(message, bytes): - message = try_decode_UTF8( - message - ) # this is slower but ensures we have UTF-8 - if not message: - logger.warning("Can't send message, message is not valid UTF-8") - return False - elif not isinstance(message, str): - logger.warning( - "Can't send message, message has to be a string or bytes. Got %s" - % type(message) - ) - return False - - header = bytearray() - payload = encode_to_UTF8(message) - payload_length = len(payload) - - # Normal payload - if payload_length <= 125: - header.append(FIN | opcode) - header.append(payload_length) - - # Extended payload - elif payload_length >= 126 and payload_length <= 65535: - header.append(FIN | opcode) - header.append(PAYLOAD_LEN_EXT16) - header.extend(struct.pack(">H", payload_length)) - - # Huge extended payload - elif payload_length < 18446744073709551616: - header.append(FIN | opcode) - header.append(PAYLOAD_LEN_EXT64) - header.extend(struct.pack(">Q", payload_length)) - - else: - raise Exception("Message is too big. Consider breaking it into chunks.") - return - - with self._send_lock: - self.request.send(header + payload) - - def read_http_headers(self): - headers = {} - # first line should be HTTP GET - http_get = self.rfile.readline().decode().strip() - assert http_get.upper().startswith("GET") - # remaining should be headers - while True: - header = self.rfile.readline().decode().strip() - if not header: - break - head, value = header.split(":", 1) - headers[head.lower().strip()] = value.strip() - return headers - - def handshake(self): - headers = self.read_http_headers() - - try: - assert headers["upgrade"].lower() == "websocket" - except AssertionError: - self.keep_alive = False - return - - try: - key = headers["sec-websocket-key"] - except KeyError: - logger.warning("Client tried to connect but was missing a key") - self.keep_alive = False - return - - response = self.make_handshake_response(key) - with self._send_lock: - self.handshake_done = self.request.send(response.encode()) - self.valid_client = True - self.server._new_client_(self) - - @classmethod - def make_handshake_response(cls, key): - return ( - "HTTP/1.1 101 Switching Protocols\r\n" - "Upgrade: websocket\r\n" - "Connection: Upgrade\r\n" - "Sec-WebSocket-Accept: %s\r\n" - "\r\n" % cls.calculate_response_key(key) - ) - - @classmethod - def calculate_response_key(cls, key): - GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11" - hash = sha1(key.encode() + GUID.encode()) - response_key = b64encode(hash.digest()).strip() - return response_key.decode("ASCII") - - def finish(self): - self.server._client_left_(self) - - -def encode_to_UTF8(data): - try: - return data.encode("UTF-8") - except UnicodeEncodeError as e: - logger.error("Could not encode data to UTF-8 -- %s" % e) - return False - except Exception as e: - raise (e) - return False - - -def try_decode_UTF8(data): - try: - return data.decode("utf-8") - except UnicodeDecodeError: - return False - except Exception as e: - raise (e) diff --git a/robotics_application_manager/libs/process_utils.py b/robotics_application_manager/libs/process_utils.py index 56126d9..49da0a3 100644 --- a/robotics_application_manager/libs/process_utils.py +++ b/robotics_application_manager/libs/process_utils.py @@ -8,7 +8,7 @@ import subprocess import zipfile import base64 - +import importlib import psutil from robotics_application_manager import LogManager @@ -17,10 +17,9 @@ def get_class(kls): parts = kls.split(".") module = ".".join(parts[:-1]) - m = __import__(module) - for comp in parts[1:]: - m = getattr(m, comp) - return m + m = importlib.import_module(module) + imp_class = getattr(m,parts[-1]) + return imp_class def get_class_from_file(file_path: str, class_name: str): diff --git a/robotics_application_manager/libs/server.py b/robotics_application_manager/libs/server.py index 7bb488c..eb9c1c7 100644 --- a/robotics_application_manager/libs/server.py +++ b/robotics_application_manager/libs/server.py @@ -1,56 +1,75 @@ import threading import json -from websocket_server import WebsocketServer +from websockets.asyncio.server import serve +import asyncio from robotics_application_manager import LogManager -class Server(threading.Thread): +class Server: def __init__( self, port, callback, ): - super().__init__() + self.host = "127.0.0.1" + self.port = port self.update_callback = callback - self.server = WebsocketServer(port=port, host="127.0.0.1") - self.server.set_fn_new_client(self.on_open) - self.server.set_fn_client_left(self.on_close) - self.server.set_fn_message_received(self.on_message) - self.current_client = None + self.client = None self.client_lock = threading.Lock() # Used to avoid concurrency problems self._stop = threading.Event() LogManager.logger.info("Server Launched") - def run(self) -> None: - try: - while not self._stop.is_set(): - self.server.run_forever() - except Exception as ex: - LogManager.logger.exception(ex) + # Use the __await__ method to make the class awaitable + def __await__(self): + # Call ls the constructor and returns the instance + return self.create().__await__() - def stop(self) -> None: - self._stop.set() - self.server.shutdown_gracefully() + # A method that creates an instance of the class asynchronously + async def create(self): + self.server = await serve(self.manage_conection, self.host, self.port, start_serving=False) + return self - def send(self, data): + async def manage_conection(self, websocket): with self.client_lock: - if self.current_client is not None: - self.server.send_message(self.current_client, data) + if self.client is not None: + await websocket.close() + return - def on_message(self, client, server, message): - payload = json.loads(message) - self.update_callback(payload) - LogManager.logger.debug(f"Message received from template: {message[:30]}") + try: + with self.client_lock: + self.client = websocket # Register client + LogManager.logger.info(f"Client connected: {self.client}") - def on_close(self, client, server): - LogManager.logger.info("Connection with client closed") - with self.client_lock: - if client == self.current_client: - self.current_client = None + await self.process_msg(websocket) + except: + pass + finally: + LogManager.logger.info("Connection with client closed") + with self.client_lock: + self.client = None - def on_open(self, client, server): - LogManager.logger.info(f"New client connected {client}") + async def process_msg(self, websocket): + async for raw_msg in websocket: + try: + json_msg = json.loads(raw_msg) + await self.update_callback(json_msg) + except Exception as e: + ex = ManagerConsumerMessageException(id=str(uuid4()), message=str(e)) + await self.send_message(ex) + LogManager.logger.error(e, exc_info=True) + + async def send(self, data): with self.client_lock: - self.current_client = client + if self.client is not None: + await self.client.send(data) + + async def start(self): + """Start the WebSocket server in a separate thread.""" + await self.server.start_serving() + + async def stop(self): + """Stop the WebSocket server gracefully.""" + self.server.close() + await self.server.wait_closed() diff --git a/robotics_application_manager/manager/docker_thread/docker_thread.py b/robotics_application_manager/manager/docker_thread/docker_thread.py index 27ba677..9b97099 100644 --- a/robotics_application_manager/manager/docker_thread/docker_thread.py +++ b/robotics_application_manager/manager/docker_thread/docker_thread.py @@ -10,23 +10,36 @@ class DockerThread(threading.Thread): """Threaded Docker Thread Class""" - def __init__(self, cmd, shell=True): + def __init__(self, cmd, shell=True, debug=False): threading.Thread.__init__(self) self.cmd = cmd self.process = None self.shell = shell + self.debug = debug def run(self): - self.process = subprocess.Popen( - self.cmd, - shell=self.shell, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - start_new_session=True, - bufsize=1024, - universal_newlines=True, - executable="/bin/bash", - ) + if self.debug: + self.process = subprocess.Popen( + self.cmd, + shell=self.shell, + stdout=sys.stdout, + stderr=sys.stdout, + start_new_session=True, + bufsize=1024, + universal_newlines=True, + executable="/bin/bash", + ) + else: + self.process = subprocess.Popen( + self.cmd, + shell=self.shell, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + start_new_session=True, + bufsize=1024, + universal_newlines=True, + executable="/bin/bash", + ) self.process.communicate() def terminate(self): diff --git a/robotics_application_manager/manager/launcher/launcher_console.py b/robotics_application_manager/manager/launcher/launcher_console.py index 6848118..f72823c 100644 --- a/robotics_application_manager/manager/launcher/launcher_console.py +++ b/robotics_application_manager/manager/launcher/launcher_console.py @@ -17,7 +17,7 @@ class LauncherConsole(ILauncher): threads: List[Any] = [] console_vnc: Any = Vnc_server() - def run(self, config_file, callback): + async def run(self, config_file, callback): DRI_PATH = self.get_dri_path() ACCELERATION_ENABLED = False @@ -52,7 +52,7 @@ def reset(self, robot_entities=[]): def is_running(self): return self.running - def terminate(self): + async def terminate(self): LogManager.logger.info(f"Terminating console tool") self.console_vnc.terminate() for thread in self.threads[:]: diff --git a/robotics_application_manager/manager/launcher/launcher_gzsim.py b/robotics_application_manager/manager/launcher/launcher_gzsim.py index e375100..67029c1 100644 --- a/robotics_application_manager/manager/launcher/launcher_gzsim.py +++ b/robotics_application_manager/manager/launcher/launcher_gzsim.py @@ -13,11 +13,11 @@ from typing import List, Any from robotics_application_manager import LogManager -from gz.msgs10.world_control_pb2 import WorldControl -from gz.msgs10.world_reset_pb2 import WorldReset -from gz.msgs10.entity_pb2 import Entity -from gz.msgs10.boolean_pb2 import Boolean -from gz.transport13 import Node +from gz.msgs.world_control_pb2 import WorldControl +from gz.msgs.world_reset_pb2 import WorldReset +from gz.msgs.entity_pb2 import Entity +from gz.msgs.boolean_pb2 import Boolean +from gz.transport import Node class LauncherGzsim(ILauncher): @@ -31,7 +31,7 @@ class LauncherGzsim(ILauncher): acceptsMsgs: bool = False gz_vnc: Any = Vnc_server() - def run(self, config_file, callback): + async def run(self, config_file, callback): DRI_PATH = self.get_dri_path() ACCELERATION_ENABLED = self.check_device(DRI_PATH) @@ -72,7 +72,7 @@ def check_device(self, device_path): def is_running(self): return self.running - def terminate(self): + async def terminate(self): LogManager.logger.info(f"Terminating gz tool") self.gz_vnc.terminate() for thread in self.threads[:]: diff --git a/robotics_application_manager/manager/launcher/launcher_interface.py b/robotics_application_manager/manager/launcher/launcher_interface.py index 7654eee..cf284e4 100644 --- a/robotics_application_manager/manager/launcher/launcher_interface.py +++ b/robotics_application_manager/manager/launcher/launcher_interface.py @@ -5,20 +5,26 @@ class ILauncher(BaseModel): - def run(self, callback: callable): + def __await__(self): + return self.create().__await__() + + async def create(self): + return self + + async def run(self, callback: callable): raise NotImplemented("Launcher must implement run method") def is_running(self): raise NotImplemented("Launcher must implement run method") - def terminate(self): + async def terminate(self): raise NotImplemented("Launcher must implement run method") def died(self, callback): raise NotImplemented("Launcher must implement run method") - def from_config(cls, config): - obj = cls(**config) + async def from_config(cls, config): + obj = await cls(**config) return obj @staticmethod diff --git a/robotics_application_manager/manager/launcher/launcher_robot.py b/robotics_application_manager/manager/launcher/launcher_robot.py index f5cb89d..cd0b93d 100644 --- a/robotics_application_manager/manager/launcher/launcher_robot.py +++ b/robotics_application_manager/manager/launcher/launcher_robot.py @@ -46,7 +46,7 @@ class LauncherRobot(BaseModel): entity: str = "" start_pose: Optional[list] = [] - def run(self, entity="", start_pose=None, extra_config=None): + async def run(self, entity="", start_pose=None, extra_config=None): """Run the robot launcher with an optional start pose.""" self.entity = entity @@ -58,19 +58,19 @@ def run(self, entity="", start_pose=None, extra_config=None): for module in worlds[self.type][str(self.ros_version)]: module["launch_file"] = self.launch_file_path - launcher = self.launch_module(module, extra_config) + launcher = await self.launch_module(module, extra_config) self.launchers.append(launcher) LogManager.logger.info(self.launchers) - def terminate(self): + async def terminate(self): """Terminate all robot launchers and clear the launchers list.""" LogManager.logger.info("Terminating robots launchers") if self.launchers: for launcher in self.launchers: - launcher.terminate() + await launcher.terminate() self.launchers = [] - def launch_module(self, configuration, extra_config=None): + async def launch_module(self, configuration, extra_config=None): """Launch a robot module based on the provided configuration.""" def process_terminated(name, exit_code): @@ -86,9 +86,9 @@ def process_terminated(name, exit_code): f"Launcher{class_from_module(launcher_module_name)}" ) launcher_class = get_class(launcher_module) - launcher = launcher_class.from_config(launcher_class, configuration) + launcher = await launcher_class.from_config(launcher_class, configuration) - launcher.run(self.entity, self.start_pose, extra_config, process_terminated) + await launcher.run(self.entity, self.start_pose, extra_config, process_terminated) return launcher def launch_command(self, configuration): diff --git a/robotics_application_manager/manager/launcher/launcher_robot_ros2_api.py b/robotics_application_manager/manager/launcher/launcher_robot_ros2_api.py index ed642b9..81787f6 100644 --- a/robotics_application_manager/manager/launcher/launcher_robot_ros2_api.py +++ b/robotics_application_manager/manager/launcher/launcher_robot_ros2_api.py @@ -21,7 +21,7 @@ class LauncherRobotRos2Api(ILauncher): launch_file: str threads: List[Any] = [] - def run(self, entity, robot_pose, extra_config, callback): + async def run(self, entity, robot_pose, extra_config, callback): DRI_PATH = self.get_dri_path() ACCELERATION_ENABLED = self.check_device(DRI_PATH) @@ -41,7 +41,7 @@ def run(self, entity, robot_pose, extra_config, callback): exercise_launch_thread.start() self.threads.append(exercise_launch_thread) - def terminate(self): + async def terminate(self): LogManager.logger.info(f"Terminating robot launcher") for thread in self.threads[:]: if thread.is_alive(): diff --git a/robotics_application_manager/manager/launcher/launcher_ros2_api.py b/robotics_application_manager/manager/launcher/launcher_ros2_api.py index 1688604..5ca5516 100644 --- a/robotics_application_manager/manager/launcher/launcher_ros2_api.py +++ b/robotics_application_manager/manager/launcher/launcher_ros2_api.py @@ -21,7 +21,7 @@ class LauncherRos2Api(ILauncher): launch_file: str threads: List[Any] = [] - def run(self, callback): + async def run(self, callback): DRI_PATH = self.get_dri_path() ACCELERATION_ENABLED = self.check_device(DRI_PATH) @@ -41,7 +41,7 @@ def run(self, callback): exercise_launch_thread.start() self.threads.append(exercise_launch_thread) - def terminate(self): + async def terminate(self): LogManager.logger.info(f"Terminating world launcher") for thread in self.threads[:]: if thread.is_alive(): diff --git a/robotics_application_manager/manager/launcher/launcher_ros2_gz_api.py b/robotics_application_manager/manager/launcher/launcher_ros2_gz_api.py index e941443..eeadadd 100644 --- a/robotics_application_manager/manager/launcher/launcher_ros2_gz_api.py +++ b/robotics_application_manager/manager/launcher/launcher_ros2_gz_api.py @@ -11,9 +11,9 @@ from robotics_application_manager.manager.docker_thread import DockerThread import subprocess from robotics_application_manager import LogManager -from gz.transport13 import Node -from gz.msgs10.empty_pb2 import Empty -from gz.msgs10.scene_pb2 import Scene +from gz.transport import Node +from gz.msgs.empty_pb2 import Empty +from gz.msgs.scene_pb2 import Scene import logging @@ -23,7 +23,7 @@ class LauncherRos2GzApi(ILauncher): launch_file: str threads: List[Any] = [] - def run(self, callback): + async def run(self, callback): DRI_PATH = self.get_dri_path() ACCELERATION_ENABLED = self.check_device(DRI_PATH) @@ -43,7 +43,7 @@ def run(self, callback): exercise_launch_thread.start() self.threads.append(exercise_launch_thread) - def terminate(self): + async def terminate(self): LogManager.logger.info(f"Terminating world launcher") for thread in self.threads[:]: if thread.is_alive(): diff --git a/robotics_application_manager/manager/launcher/launcher_rviz.py b/robotics_application_manager/manager/launcher/launcher_rviz.py index d08c05e..36fdd33 100644 --- a/robotics_application_manager/manager/launcher/launcher_rviz.py +++ b/robotics_application_manager/manager/launcher/launcher_rviz.py @@ -26,7 +26,7 @@ class LauncherRviz(ILauncher): threads: List[Any] = [] vnc: Any = Vnc_server() - def run(self, config_file, callback): + async def run(self, config_file, callback): """ Launches an RViz instance with a specific display configuration. @@ -72,7 +72,7 @@ def reset(self, robot_entities=[]): def is_running(self): return self.running - def terminate(self): + async def terminate(self): LogManager.logger.info(f"Terminating rviz tool") self.vnc.terminate() for thread in self.threads[:]: diff --git a/robotics_application_manager/manager/launcher/launcher_scene.py b/robotics_application_manager/manager/launcher/launcher_scene.py index 61cd650..00daefb 100644 --- a/robotics_application_manager/manager/launcher/launcher_scene.py +++ b/robotics_application_manager/manager/launcher/launcher_scene.py @@ -55,20 +55,20 @@ class LauncherScene(BaseModel): ros_version: int = get_ros_version() launchers: Optional[ILauncher] = [] - def run(self): + async def run(self): for module in worlds[self.type][str(self.ros_version)]: module["launch_file"] = self.launch_file_path - launcher = self.launch_module(module) + launcher = await self.launch_module(module) self.launchers.append(launcher) - def terminate(self): + async def terminate(self): LogManager.logger.info("Terminating scenes launchers") if self.launchers: for launcher in self.launchers: - launcher.terminate() + await launcher.terminate() self.launchers = [] - def launch_module(self, configuration): + async def launch_module(self, configuration): def process_terminated(name, exit_code): LogManager.logger.info( f"LauncherEngine: {name} exited with code {exit_code}" @@ -79,8 +79,8 @@ def process_terminated(name, exit_code): launcher_module_name = configuration["module"] launcher_module = f"{self.module}.launcher_{launcher_module_name}.Launcher{class_from_module(launcher_module_name)}" launcher_class = get_class(launcher_module) - launcher = launcher_class.from_config(launcher_class, configuration) - launcher.run(process_terminated) + launcher = await launcher_class.from_config(launcher_class, configuration) + await launcher.run(process_terminated) return launcher def launch_command(self, configuration): diff --git a/robotics_application_manager/manager/launcher/launcher_state_monitor.py b/robotics_application_manager/manager/launcher/launcher_state_monitor.py index 94208d6..406dd30 100644 --- a/robotics_application_manager/manager/launcher/launcher_state_monitor.py +++ b/robotics_application_manager/manager/launcher/launcher_state_monitor.py @@ -15,10 +15,10 @@ def __init__(self, type, module, file, consumer): self.consumer = consumer self.server = FileWatchdog("/tmp/tree_state", self.update) - def update(self, data): + async def update(self, data): LogManager.logger.debug(f"Sending update to client") if self.consumer is not None: - self.consumer.send_message({"update": data}, command="update") + await self.consumer.send_message({"update": data}, command="update") def run(self, config_file, callback): self.server.start() diff --git a/robotics_application_manager/manager/launcher/launcher_tools.py b/robotics_application_manager/manager/launcher/launcher_tools.py index 3000042..bede222 100644 --- a/robotics_application_manager/manager/launcher/launcher_tools.py +++ b/robotics_application_manager/manager/launcher/launcher_tools.py @@ -71,7 +71,7 @@ class LauncherTools(BaseModel): tools_config: Optional[dict] = None launchers: Optional[ILauncher] = [] - def run(self, consumer): + async def run(self, consumer): for tool in self.tools: if tool == "simulator": if self.world_type is None or self.world_type == "physical": @@ -80,17 +80,17 @@ def run(self, consumer): module = tools[tool] if module["module"] is None: continue - launcher = self.launch_module(tool, module, consumer) + launcher = await self.launch_module(tool, module, consumer) self.launchers.append(launcher) - def terminate(self): + async def terminate(self): LogManager.logger.info("Terminating tools launchers") for launcher in self.launchers: if launcher.is_running(): - launcher.terminate() + await launcher.terminate() self.launchers = [] - def launch_module(self, name, configuration, consumer): + async def launch_module(self, name, configuration, consumer): def process_terminated(name, exit_code): LogManager.logger.info( f"LauncherEngine: {name} exited with code {exit_code}" @@ -109,8 +109,8 @@ def process_terminated(name, exit_code): if self.tools_config is not None and name in self.tools_config: config = self.tools_config[name] - launcher = launcher_class.from_config(launcher_class, configuration) - launcher.run(config, process_terminated) + launcher = await launcher_class.from_config(launcher_class, configuration) + await launcher.run(config, process_terminated) return launcher def pause(self): @@ -125,10 +125,10 @@ def reset(self, robot_entities=[]): for launcher in self.launchers: launcher.reset(robot_entities) - def pass_msg(self, data): + async def pass_msg(self, data): for launcher in self.launchers: if launcher.acceptsMsgs: - launcher.get_msg(data) + await launcher.get_msg(data) def launch_command(self, configuration): pass diff --git a/robotics_application_manager/manager/launcher/launcher_web_gui.py b/robotics_application_manager/manager/launcher/launcher_web_gui.py index 00f9306..3a3719a 100644 --- a/robotics_application_manager/manager/launcher/launcher_web_gui.py +++ b/robotics_application_manager/manager/launcher/launcher_web_gui.py @@ -13,25 +13,31 @@ class LauncherWebGui: def __init__(self, type, module, internal_port, consumer): self.internal_port = internal_port self.consumer = consumer - self.server = Server(self.internal_port, self.update) - def update(self, data): + def __await__(self): + return self.create().__await__() + + async def create(self): + self.server = await Server(self.internal_port, self.update) + return self + + async def update(self, data): LogManager.logger.debug(f"Sending update to client") if self.consumer is not None: - self.consumer.send_message({"update": data}, command="update") + await self.consumer.send_message({"update": data}, command="update") - def run(self, config_file, callback): - self.server.start() + async def run(self, config_file, callback): + await self.server.start() self.running = True - def get_msg(self, data): - self.server.send(data) + async def get_msg(self, data): + await self.server.send(data) def is_running(self): return self.running - def terminate(self): - self.server.stop() + async def terminate(self): + await self.server.stop() self.running = False def pause(self): @@ -46,6 +52,6 @@ def reset(self, robot_entities=[]): def died(self): pass - def from_config(cls, config): - obj = cls(**config) + async def from_config(cls, config): + obj = await cls(**config) return obj diff --git a/robotics_application_manager/manager/manager.py b/robotics_application_manager/manager/manager.py index 9b5a44c..e6751f1 100644 --- a/robotics_application_manager/manager/manager.py +++ b/robotics_application_manager/manager/manager.py @@ -23,12 +23,14 @@ import zipfile import jedi import traceback +import asyncio from queue import Queue from uuid import uuid4 -from transitions import Machine +from transitions.extensions import AsyncMachine from robotics_application_manager.comms import ( ManagerConsumerMessageException, + ManagerConsumerMessage, ManagerConsumer, ) from robotics_application_manager.libs import ( @@ -211,7 +213,7 @@ def __init__(self, host: str, port: int): host (str): The host address to listen to. port (int): The port number to listen to. """ - self.machine = Machine( + self.machine = AsyncMachine( model=self, states=Manager.states, transitions=Manager.transitions, @@ -220,8 +222,7 @@ def __init__(self, host: str, port: int): after_state_change=self.state_change, ) self.ros_version = subprocess.check_output(["bash", "-c", "echo $ROS_DISTRO"]) - self.queue = Queue() - self.consumer = ManagerConsumer(host, port, self.queue) + self.consumer = None self.scene_launcher = None self.world_type = None self.robot_launchers = [] @@ -230,6 +231,8 @@ def __init__(self, host: str, port: int): self.application_processes = [] self.running = True self.linter = Lint() + self.host = host + self.port = port # Creates workspace directories worlds_dir = "/workspace/worlds" @@ -242,7 +245,7 @@ def __init__(self, host: str, port: int): if not os.path.isdir(binaries_dir): os.makedirs(binaries_dir) - def state_change(self, event): + async def state_change(self, event): """ Handle actions to be performed after a state change in the state machine. @@ -251,9 +254,9 @@ def state_change(self, event): """ LogManager.logger.info(f"State changed to {self.state}") if self.consumer is not None: - self.consumer.send_message({"state": self.state}, command="state-changed") + await self.consumer.send_message({"state": self.state}, command="state-changed") - def update(self, data): + async def update(self, data): """ Send an update message to the client with the provided data. @@ -262,9 +265,9 @@ def update(self, data): """ LogManager.logger.debug("Sending update to client") if self.consumer is not None: - self.consumer.send_message({"update": data}, command="update") + await self.consumer.send_message({"update": data}, command="update") - def update_bt_studio(self, data): + async def update_bt_studio(self, data): """ Send an update message to the client for BT Studio with the provided data. @@ -273,9 +276,9 @@ def update_bt_studio(self, data): """ LogManager.logger.debug("Sending update to client") if self.consumer is not None: - self.consumer.send_message({"update": data}, command="update") + await self.consumer.send_message({"update": data}, command="update") - def on_connect(self, event): + async def on_connect(self, event): """ Triggered when the application transitions to the 'connected' state. @@ -289,7 +292,7 @@ def on_connect(self, event): - `ros_version`: The current ROS (Robot Operating System) distribution version. - `gpu_avaliable`: Boolean indicating whether GPU acceleration is available. """ - self.consumer.send_message( + await self.consumer.send_message( { "robotics_backend_version": subprocess.check_output( ["bash", "-c", "echo $IMAGE_TAG"] @@ -300,7 +303,7 @@ def on_connect(self, event): command="introspection", ) - def on_launch_world(self, event): + async def on_launch_world(self, event): """ Handle the 'launch' event, transitioning the application from 'connected' to 'ready' state. @@ -396,19 +399,19 @@ def make_unique(cfgs, key, param=None): self.robot_configs.append(robot_cfg) LogManager.logger.info(str(robot_launcher)) - self.scene_launcher.run() + await self.scene_launcher.run() robots_data = zip(self.robot_launchers, self.robot_configs) entities = [] for launcher, cfg in robots_data: - launcher.run(cfg["entity"], cfg["start_pose"], cfg["extra_config"]) + await launcher.run(cfg["entity"], cfg["start_pose"], cfg["extra_config"]) entities.append(cfg["entity"]) self.scene_launcher.wait_robots_spawn(entities) LogManager.logger.info("Launch transition finished") - def prepare_custom_world(self, cfg_dict): + async def prepare_custom_world(self, cfg_dict): """ Prepare and extract a custom world from a base64-encoded zip file. @@ -445,10 +448,8 @@ def prepare_custom_world(self, cfg_dict): '/bin/bash -c "cd /workspace/worlds; source /opt/ros/humble/setup.bash; colcon build --symlink-install; source install/setup.bash; cd ../.."' ) - def on_prepare_tools(self, event): - + async def on_prepare_tools(self, event): LogManager.logger.info("Tools transition started") - cfg_dict = event.kwargs.get("data", {}) tools = cfg_dict["tools"] config = cfg_dict["config"] @@ -457,7 +458,7 @@ def on_prepare_tools(self, event): world_type=self.world_type, tools=tools, tools_config=config ) - self.tools_launcher.run(self.consumer) + await self.tools_launcher.run(self.consumer) LogManager.logger.info("Tools transition finished") def write_to_tool_terminal(self, msg): @@ -481,7 +482,7 @@ def write_to_tool_terminal(self, msg): with open(i, "w") as console: console.write(msg) - def on_style_check_application(self, event): + async def on_style_check_application(self, event): """ Handle the 'style_check' event. @@ -527,7 +528,7 @@ def on_style_check_application(self, event): self.write_to_tool_terminal(errors + "\n\n") raise Exception(errors) - def on_code_analysis(self, event): + async def on_code_analysis(self, event): """ Handle the 'code_analysis' event. @@ -592,7 +593,7 @@ def on_code_analysis(self, event): command="code-analysis", ) - def on_code_format(self, event): + async def on_code_format(self, event): """ Handle the 'code_format' event. @@ -764,7 +765,7 @@ def on_change_style(self, event): except Exception as e: LogManager.logger.exception(f"Error refreshing GTK applications: {e}") - def on_run_application(self, event): + async def on_run_application(self, event): """ Handle the 'run_application' event. @@ -829,7 +830,7 @@ def on_run_application(self, event): if needs_compile: compile_process = subprocess.Popen( [ - "cd /workspace/code && source /opt/ros/humble/setup.bash && colcon build && source install/setup.bash && cd ../.." + "cd /workspace/code && source /opt/ros/lyrical/setup.bash && colcon build && source install/setup.bash && cd ../.." ], stdin=open("/dev/pts/" + console_fd, "r"), stdout=open("/dev/pts/" + console_fd, "w"), @@ -841,7 +842,9 @@ def on_run_application(self, event): ) returncode = compile_process.wait() if returncode != 0: - raise Exception("Failed to compile") + # raise Exception("Failed to compile") + LogManager.logger.info("Failed to compile") + return for entrypoint in entrypoints: if not os.path.isfile(entrypoint): @@ -886,7 +889,7 @@ def on_run_application(self, event): LogManager.logger.info("Run application transition finished") - def on_terminate_application(self, event): + async def on_terminate_application(self, event): """ Handle the 'terminate_application' event. @@ -901,19 +904,18 @@ def on_terminate_application(self, event): stop_process_and_children(process) except Exception: LogManager.logger.exception("No application running") - print(traceback.format_exc()) if len(self.application_processes) > 0: self.pause_sim() - self.reset_sim() + await self.reset_sim() self.application_processes = [] - def on_terminate_tools(self, event): + async def on_terminate_tools(self, event): - self.tools_launcher.terminate() + await self.tools_launcher.terminate() self.tools_launcher = None - def on_terminate_world(self, event): + async def on_terminate_world(self, event): """ Handle the 'terminate_world' event. @@ -924,22 +926,22 @@ def on_terminate_world(self, event): event: The event object associated with the termination request. """ if self.scene_launcher is not None: - self.scene_launcher.terminate() + await self.scene_launcher.terminate() self.scene_launcher = None self.world_type = None for launcher in self.robot_launchers: - launcher.terminate() + await launcher.terminate() self.robot_launchers = [] - def on_disconnect(self, event): + async def on_disconnect(self, event): """ Handle the 'disconnect' event. This method stops all running processes, terminates launchers, and restarts the script. """ - + LogManager.logger.info("-------------------Disconnecting--------------------------") try: for process in self.application_processes: stop_process_and_children(process) @@ -949,34 +951,35 @@ def on_disconnect(self, event): if self.tools_launcher: try: - self.tools_launcher.terminate() + await self.tools_launcher.terminate() except Exception as e: LogManager.logger.exception("Exception terminating tools launcher") try: for launcher in self.robot_launchers: - launcher.terminate() + await launcher.terminate() self.robot_launchers = [] except Exception as e: LogManager.logger.exception("Exception terminating robot launcher") if self.scene_launcher: try: - self.scene_launcher.terminate() + await self.scene_launcher.terminate() except Exception as e: LogManager.logger.exception("Exception terminating scene launcher") - def process_message(self, message): + async def process_message(self, message): if message.command == "gui": if self.tools_launcher is not None: - self.tools_launcher.pass_msg(message.data) + await self.tools_launcher.pass_msg(message.data) return - self.trigger(message.command, data=message.data or None) + await self.trigger(message.command, data=message.data or None) + response = {"message": f"Exercise state changed to {self.state}"} - self.consumer.send_message(message.response(response)) + await self.consumer.send_message(message.response(response)) - def on_pause(self, msg): + async def on_pause(self, msg): if len(self.application_processes) > 0: for process in self.application_processes: proc = psutil.Process(process.pid) @@ -993,9 +996,9 @@ def on_pause(self, msg): "Application process was None during pause. Calling termination." ) self.pause_sim() - self.reset_sim() + await self.reset_sim() - def on_resume(self, msg): + async def on_resume(self, msg): """ Resume the application process if it exists, otherwise reset the simulation. @@ -1017,7 +1020,7 @@ def on_resume(self, msg): LogManager.logger.warning( "Application process was None during resume. Calling termination." ) - self.reset_sim() + await self.reset_sim() def pause_sim(self): try: @@ -1033,7 +1036,7 @@ def unpause_sim(self): self.write_to_tool_terminal(f"{e}\n\n") raise Exception("Failed to start simulator") - def reset_sim(self): + async def reset_sim(self): """ Reset the simulation environment and relaunch the robot if applicable. @@ -1043,7 +1046,7 @@ def reset_sim(self): """ for robot_launcher in self.robot_launchers: - robot_launcher.terminate() + await robot_launcher.terminate() try: entities = [] @@ -1056,30 +1059,30 @@ def reset_sim(self): robots_data = zip(self.robot_launchers, self.robot_configs) for launcher, cfg in robots_data: - launcher.run(cfg["entity"], cfg["start_pose"], cfg["extra_config"]) + await launcher.run(cfg["entity"], cfg["start_pose"], cfg["extra_config"]) - def start(self): + async def start(self): """ Start the RAM. RAM must be run in main thread to be able to handle signaling other processes, for instance ROS launcher. """ + self.consumer = await ManagerConsumer(self.host, self.port, self.process_message) LogManager.logger.info( - f"Starting RAM consumer in {self.consumer.server}:{self.consumer.port}" + f"Starting RAM consumer in {self.host}:{self.port}" ) - self.consumer.start() + await self.consumer.start() - def signal_handler(sign, frame): - print("\nprogram exiting gracefully") + async def signal_handler(sign, frame): + LogManager.logger.info("Program exiting gracefully") self.running = False try: - self.consumer.stop() + await self.consumer.stop() except Exception as e: LogManager.logger.exception("Exception stopping consumer") - try: for process in self.application_processes: stop_process_and_children(process) @@ -1089,44 +1092,28 @@ def signal_handler(sign, frame): if self.tools_launcher: try: - self.tools_launcher.terminate() + await self.tools_launcher.terminate() except Exception as e: LogManager.logger.exception("Exception terminating tools launcher") try: for launcher in self.robot_launchers: - launcher.terminate() + await launcher.terminate() self.robot_launchers = [] except Exception as e: LogManager.logger.exception("Exception terminating robot launcher") if self.scene_launcher: try: - self.scene_launcher.terminate() + await self.scene_launcher.terminate() except Exception as e: LogManager.logger.exception("Exception terminating scene launcher") - exit() signal.signal(signal.SIGINT, signal_handler) while self.running: - message = None - try: - if self.queue.empty(): - time.sleep(0.1) - else: - message = self.queue.get() - self.process_message(message) - except Exception as e: - if message is not None: - ex = ManagerConsumerMessageException(id=message.id, message=str(e)) - else: - ex = ManagerConsumerMessageException( - id=str(uuid4()), message=str(e) - ) - self.consumer.send_message(ex) - LogManager.logger.error(e, exc_info=True) + await asyncio.sleep(1) if __name__ == "__main__": @@ -1134,10 +1121,10 @@ def signal_handler(sign, frame): parser = argparse.ArgumentParser() parser.add_argument( - "host", type=str, help="Host to listen to (0.0.0.0 or all hosts)" + "host", type=str, help="Host to listen to (0.0.0.0 or all hosts)" ) parser.add_argument("port", type=int, help="Port to listen to") args = parser.parse_args() RAM = Manager(args.host, args.port) - RAM.start() + asyncio.run(RAM.start())