diff --git a/Dockerfile b/Dockerfile index 89416d6..ce3b2ae 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,6 +1,7 @@ FROM python:3.9-slim-bookworm WORKDIR /app COPY requirements.txt ./ +RUN apt-get update && apt-get install build-essential -y RUN pip install --upgrade pip RUN pip install -r requirements.txt COPY src ./ diff --git a/requirements.txt b/requirements.txt index a454b02..57893fd 100644 --- a/requirements.txt +++ b/requirements.txt @@ -2,6 +2,7 @@ annotated-types==0.6.0 certifi==2024.2.2 charset-normalizer==3.3.2 click==8.1.7 +google-cloud-firestore==2.10.0 idna==3.7 markdown-it-py==3.0.0 mdurl==0.1.2 @@ -18,3 +19,4 @@ typing_extensions==4.11.0 urllib3==2.2.1 watchdog==4.0.0 sdnotify==0.3.2 +pyzmq==25.1.2 \ No newline at end of file diff --git a/src/communication_module.py b/src/communication_module.py new file mode 100644 index 0000000..11ee1b4 --- /dev/null +++ b/src/communication_module.py @@ -0,0 +1,33 @@ +#!/usr/bin/env python3 + +import logging +import zmq + +logger = logging.getLogger(__name__) + + +class CommunicationModule(object): + + def __init__(self, host="127.0.0.1", port=5555): + self.host = host + self.port = port + self.context = zmq.Context() + self.socket = self.context.socket(zmq.REQ) + + def start(self): + logger.debug( + f"Starting communication module ('{self.host}:{self.port}')") + self.socket.connect(f"tcp://{self.host}:{self.port}") + + def stop(self): + logger.debug("Stopping communication module") + self.socket.close() + self.context.term() + + def execute_command(self, command, params=None): + if not command: + logger.error("The command argument is required") + return + self.socket.send_json({"command": command, "params": params}) + response = self.socket.recv_json() + return response diff --git a/src/firestore_client_handler.py b/src/firestore_client_handler.py new file mode 100644 index 0000000..4cc5b18 --- /dev/null +++ b/src/firestore_client_handler.py @@ -0,0 +1,167 @@ +#!/usr/bin/env python3 + +from google.cloud.firestore import Client +from google.cloud.firestore_v1 import watch +from google.oauth2.credentials import Credentials +import google.api_core +import json +import logging +import requests +import threading +import time + +logger = logging.getLogger(__name__) + +watch._should_recover = lambda _: False +watch._should_terminate = lambda _: True + + +class FirestoreClientHandler: + + REFRESH_TOKEN_TIME_MIN = 58 + INITIALIZATION_RETRY_TIME_MIN = 1 + SERVER_RESPONSE_TIMEOUT_S = 60 + + def __init__(self, api_key: str, project_id: str, refresh_token: str): + self.api_key = api_key + self.project_id = project_id + self.refresh_token = refresh_token + self.token_expired_timer = None + self.initialization_retry_timer = None + self.server_responde_message_handler = None + self.user_id = None + self.client = None + + def initialize_client(self, notify=True): + if self.client: + if notify: + threading.Thread( + target=self.on_client_initialized, daemon=True).start() + return + + logger.debug("Initializing Firestore client") + creds = self._get_credentials() + + if not creds: + if self.initialization_retry_timer: + return + self.initialization_retry_timer = threading.Timer( + self.INITIALIZATION_RETRY_TIME_MIN*60.0, self.on_initialization_retry) + self.initialization_retry_timer.start() + return + + self.client = Client(self.project_id, creds) + logger.debug("Firebase client initialized") + if notify: + threading.Thread(target=self.on_client_initialized, + daemon=True).start() + self.server_responde_message_handler = ServerResponseMessageHandler( + timeout_s=self.SERVER_RESPONSE_TIMEOUT_S, on_server_not_responding=self.on_server_not_responding) + google.api_core.bidi._LOGGER = BidiCustomLogger() + google.api_core.bidi._LOGGER.addHandler( + self.server_responde_message_handler) + self.token_expired_timer = threading.Timer( + self.REFRESH_TOKEN_TIME_MIN*60.0, self.on_token_expired) + self.token_expired_timer.start() + + def stop_client(self): + logger.debug("Stopping Firestore client") + self.client = None + if self.server_responde_message_handler: + self.server_responde_message_handler.stop() + self.server_responde_message_handler = None + if self.initialization_retry_timer: + self.initialization_retry_timer.cancel() + self.initialization_retry_timer.join() + self.initialization_retry_timer = None + if self.token_expired_timer: + self.token_expired_timer.cancel() + self.token_expired_timer.join() + self.token_expired_timer = None + + def on_client_initialized(self): + logger.warning( + "This function should be overwritten by the child class") + + def on_server_not_responding(self): + logger.warning( + "This function should be overwritten by the child class") + + def on_token_expired(self): + logger.warning( + "This function should be overwritten by the child class") + + def on_initialization_retry(self): + logger.warning("Retrying firebase client initialization...") + self.initialization_retry_timer = None + threading.Thread(target=self.initialize_client).start() + + def _get_credentials(self): + user_id, token_id = self._get_ids() + if not user_id or not token_id: + logger.debug(f"Invalid user and token ids ({user_id}, {token_id})") + return None + self.user_id = user_id + creds = Credentials(token_id, self.refresh_token) + return creds + + def _get_ids(self): + token_response = self._get_token_response() + user_id = token_response.get("user_id") + token_id = token_response.get("id_token") + return (user_id, token_id) + + def _get_token_response(self): + request_ref = f"https://securetoken.googleapis.com/v1/token?key={self.api_key}" + headers = {"content-type": "application/json; charset=UTF-8"} + data = json.dumps({"grantType": "refresh_token", + "refreshToken": self.refresh_token}) + try: + response_object = requests.post( + request_ref, headers=headers, data=data) + return response_object.json() + except Exception as e: + return {} + + +class BidiCustomLogger(logging.Logger): + + def __init__(self, name="google.api_core.bidi", level=logging.DEBUG): + super().__init__(name, level) + + +class ServerResponseMessageHandler(logging.StreamHandler): + + def __init__(self, timeout_s=60, on_server_not_responding=lambda _: None, server_response_msg="recved response.", watchdog_timeout_msg="watchdog timeout"): + super().__init__() + self.timeout_s = timeout_s + self.on_server_not_responding = on_server_not_responding + self.server_response_msg = server_response_msg + self.server_response_last_time = time.time() + self.watchdog_timeout_msg = watchdog_timeout_msg + self.not_responding_timer = None + self.setFormatter(logging.Formatter( + '%(asctime)s %(levelname)-8s - %(name)-16s - %(message)s', None, '%')) + + def emit(self, record): + msg = self.format(record) + if self.watchdog_timeout_msg in msg: + logger.debug("Server connection timeout detected") + threading.Thread(target=self.on_server_not_responding).start() + if self.server_response_msg in msg: + logger.debug("Server response message caught ({} seconds)".format( + time.time() - self.server_response_last_time)) + self.server_response_last_time = time.time() + if self.not_responding_timer: + self.stop() + self.not_responding_timer = threading.Timer( + self.timeout_s, self.on_server_not_responding) + self.not_responding_timer.start() + if record.levelno >= logger.getEffectiveLevel(): + super().emit(record) + + def stop(self): + if self.not_responding_timer: + self.not_responding_timer.cancel() + self.not_responding_timer.join() + self.not_responding_timer = None diff --git a/src/installed_service_handler.py b/src/installed_service_handler.py index e2b26ea..056bbaf 100644 --- a/src/installed_service_handler.py +++ b/src/installed_service_handler.py @@ -1,7 +1,7 @@ import logging import os import pathlib -from threading import Timer +from threading import Timer, Thread from typing import List, Optional from python_on_whales import DockerClient, docker @@ -36,66 +36,95 @@ class InstalledServiceHandler(FileSystemEventHandler): docker: Optional[DockerClient] """The docker client of the service compose""" - def __init__(self, service_path: str, wait_seconds: float): + def __init__(self, service_path: str, wait_seconds: float, service_state_change_callback: callable=lambda _: None): self.service_path = service_path self.service_name = service_path.split("/")[-1] self.wait_seconds = wait_seconds + self.service_state_change_callback = service_state_change_callback self.timer = None - self.observer = Observer() + self.observer = None self.files = os.listdir(service_path) self.docker = self._get_docker() - def start(self): + def start(self, mode: str="offline"): """Start the handler by starting the observer of the service folder.""" + logger.debug(f"Starting '{self.service_name}' Installed Service Handler.") try: - self.observer.schedule(self, self.service_path, recursive=True) - self.observer.start() - logger.debug(f"{self.service_name} Installed Service Handler started.") + if not self.is_running(): + logger.error(f"Service '{self.service_name}' is not running, starting it up") + self.up_compose_services() + if mode == "offline": + self.observer = Observer() + self.observer.schedule(self, self.service_path, recursive=True) + self.observer.start() except FileNotFoundError: logger.error( f"Couldn't start a watcher for the {self.service_name} service, the folder no longer exists." ) - self.down() + self.down_compose_services() + self.clean_compose_services() def stop(self): """Stop the handler by stopping the observer of the service folder.""" - logger.debug("Installed Service Handler stopped.") - self.observer.stop() - - def up(self): - """Start the compose of the service by doing "docker compose up".""" - if self.docker: - try: - self.docker.compose.up(detach=True) - logger.info(f"{self.service_name} service compose started.") - except: - logger.error( - f"An error occurred, couldn't start service {self.service_name}." - ) - - def down(self): - """Stop the compose of the service. + logger.debug(f"Stopping '{self.service_name}' Installed Service Handler.") + if self.observer: + self.observer.stop() + + def up_compose_services(self): + """Start the services of the compose file by executing "docker compose pull" and "docker compose up".""" + if not self.docker: + return + try: + Thread(target=self.service_state_change_callback, args=[self.service_name, "pulling"]).start() + self.docker.compose.pull() + Thread(target=self.service_state_change_callback, args=[self.service_name, "starting"]).start() + self.docker.compose.up(detach=True, pull="never") + Thread(target=self.service_state_change_callback, args=[self.service_name, "started"]).start() + logger.info(f"'{self.service_name}' compose services started.") + except Exception as e: + logger.error( + f"An error occurred, couldn't start service {self.service_name}: {e}." + ) + Thread(target=self.service_state_change_callback, args=[self.service_name, "unknown"]).start() + + def down_compose_services(self, remove_images: bool = False): + """Shut down the services of the compose file by executing "docker compose down".""" + if not self.docker: + return + self.docker.compose.down(remove_images="all" if remove_images else None) + logger.info(f"'{self.service_name}' compose services downed.") + + def stop_compose_services(self): + """Stop the services of the compose file by executing "docker compose stop".""" + if not self.docker: + return + self.docker.compose.stop() + logger.info(f"'{self.service_name}' compose services stopped.") + + def clean_compose_services(self): + """Clean the services of the compose file. This is done by getting the containers started by the compose file and killing them. The volumes are pruned to remove any unused ones. - This is done like this because, when the service folder is removed the compose needs to stop. - But, in that case, the compose file no longer exists, so "docker compose down" can't be called. + This is done like this because, when the service folder is removed and, as such, the compose + file no longer exists, "docker compose down" can not be directly called. """ - if self.docker: - logger.info(f"{self.service_name} service compose stopped.") - containers = self.docker.ps( - filters={ - "label": f"com.docker.compose.project.config_files={self.service_path}/{self.compose_file}" - } - ) - for container in containers: - docker.container.stop(container) - docker.container.remove(container) - - docker.volume.prune() - - def reload_service_compose(self): + if not self.docker: + return + containers = self.docker.ps( + filters={ + "label": f"com.docker.compose.project.config_files={self.service_path}/{self.compose_file}" + } + ) + for container in containers: + docker.container.stop(container) + docker.container.remove(container) + + docker.volume.prune() + logger.info(f"'{self.service_name}' compose services cleaned.") + + def reload_compose_services(self): """Stop the compose, get the docker client again and start the compose. The docker client is loaded again because, if the compose file is changed, the docker client will not be valid. @@ -105,12 +134,20 @@ def reload_service_compose(self): logger.debug(f"{self.service_name} service modified.") - self.down() + self.down_compose_services() self.docker = self._get_docker() - self.up() + self.up_compose_services() self.timer = None + def is_running(self): + """Check if the service is running""" + running_services = self.docker.compose.ls() + for running_service in running_services: + if running_service.name == self.service_name and running_service.running == 1: + return True + return False + def on_any_event(self, event: FileSystemEvent): """Reload the service when the service changes. @@ -133,7 +170,7 @@ def on_any_event(self, event: FileSystemEvent): self.timer.cancel() self.timer = None - self.timer = Timer(self.wait_seconds, self.reload_service_compose) + self.timer = Timer(self.wait_seconds, self.reload_compose_services) self.timer.start() def _get_compose_file_name(self): diff --git a/src/installed_services_remote_provider.py b/src/installed_services_remote_provider.py new file mode 100644 index 0000000..92c3694 --- /dev/null +++ b/src/installed_services_remote_provider.py @@ -0,0 +1,116 @@ +#!/usr/bin/env python3 + +from firestore_client_handler import FirestoreClientHandler +import logging +import threading + +from google.cloud.firestore_v1 import DocumentSnapshot +from google.cloud.firestore_v1.watch import ChangeType, DocumentChange +from proto.datetime_helpers import DatetimeWithNanoseconds +from typing import List + +logger = logging.getLogger(__name__) + + +class InstalledServicesRemoteProvider(FirestoreClientHandler): + + RESTART_DELAY_TIME_S = 0.5 + + def __init__(self, api_key: str, project_id: str, refresh_token: str, device_id: str, connection_state_change_callback: callable=lambda _: None, service_state_change_callback: callable=lambda _: None): + super().__init__(api_key, project_id, refresh_token) + self.device_id = device_id + self.connection_state_change_callback = connection_state_change_callback + self.service_state_change_callback = service_state_change_callback + self.start_timeout_timer = None + self.device = None + self.watch = None + + def start(self, timeout: int=0): + logger.debug("Starting IoMBian Installed Services Remote Provider") + if timeout > 0: + self.start_timeout_timer = threading.Timer(interval=timeout, function=self.connection_state_change_callback, args=["timeout"]) + self.start_timeout_timer.start() + self.initialize_client() + + def stop(self): + logger.debug("Stopping IoMBian Installed Services Remote Provider") + if self.watch is not None: + self.watch.unsubscribe() + self.device = None + self.stop_client() + + def restart(self): + """Restart the Installed Services Remote Provider by calling `stop()` and `start()`.""" + self.stop() + self.start() + + def on_client_initialized(self): + """Callback function when the client is initialized.""" + logger.debug("Firestore client initialized") + if self.start_timeout_timer: + self.start_timeout_timer.cancel() + self.start_timeout_timer = None + threading.Thread(target=self.connection_state_change_callback, args=["connected"]).start() + self.device = ( + self.client.collection("users") + .document(self.user_id) + .collection("devices") + .document(self.device_id) + ) + self.watch = self.device.collection("installed_services").on_snapshot( + self._on_installed_service_change + ) + + def on_server_not_responding(self): + """Callback function when the server is not responding.""" + logger.error("Firestore server not responding") + threading.Timer(self.RESTART_DELAY_TIME_S, self.restart).start() + threading.Thread(target=self.connection_state_change_callback, args=["disconnected"]).start() + + def on_token_expired(self): + """Callback function when the token is expired.""" + logger.debug("Refreshing Firebase client token id") + threading.Timer(self.RESTART_DELAY_TIME_S, self.restart).start() + + def update_remote_service_status(self, service_name: str, service_status: str): + """Given the service name and a status, update the service status in firebase.""" + logger.debug( + f"Updating '{service_name}' service status to {service_status} in Firebase") + service_reference = self.device.collection("installed_services").document( + service_name + ) + try: + service_reference.update({"status": service_status}) + except: + logger.warning(f"The '{service_name}' service is not installed in Firebase, status cannot be updated") + + def _on_installed_service_change( + self, + snapshots: List[DocumentSnapshot], + changes: List[DocumentChange], + read_time: DatetimeWithNanoseconds, + ): + """For each change in the installed services collection, check the status field and act accordingly. + + Here is the list of the status values that should be handled: + - downloaded: the installation information is downloaded to the device. + - to-be-uninstalled: the user has indicated that the service should be uninstalled. + """ + for change in changes: + service_snapshot = change.document + service_name = service_snapshot.id + + if change.type == ChangeType.REMOVED: + continue + + service_status = service_snapshot.to_dict().get("status") + + logger.debug( + f"Firebase notification received for '{service_name}' service ({service_status})") + + if not service_status: + logger.error(f"'{service_name}' service does not have any status in Firebase") + continue + + if service_status in ["downloaded", "reconfigured", "updated", "to-be-updated", "to-be-uninstalled"]: + threading.Thread(target=self.service_state_change_callback, args=[service_name, service_status]).start() diff --git a/src/iombian_services_handler.py b/src/iombian_services_handler.py index b18db34..569be1f 100644 --- a/src/iombian_services_handler.py +++ b/src/iombian_services_handler.py @@ -1,6 +1,7 @@ import logging import os import pathlib +import threading from typing import List from watchdog.events import ( @@ -31,39 +32,94 @@ class IombianServicesHandler(FileSystemEventHandler): """The `Observer` of the "iombian-services" folder""" services: List[InstalledServiceHandler] """List of the `InstalledServiceHandler` on the "iombian-services" folder""" + local_service_state_change_callback: callable + """Callback function to be called when a service state changes""" + mode: str + """Mode of the installation process: 'online' (based on Firebase) or 'offline' (based on folder events)""" - def __init__(self, base_path: str, wait_seconds: float) -> None: + def __init__(self, base_path: str, wait_seconds: float, local_service_state_change_callback: callable=lambda _: None): self.base_path = base_path self.wait_seconds = wait_seconds - self.observer = Observer() + self.local_service_state_change_callback = local_service_state_change_callback + self.observer = None + self.mode = "offline" self.services = [] + def set_mode(self, mode: str): + """Set the handler mode: 'online' or 'offline'""" + logger.debug(f"Setting mode to '{mode}'") + self.mode = mode + def start(self): """Start the handler by starting the observer of the "iombian-services" folder.""" - logger.info("IoMBian Installed Services Handler started.") - self.observer.schedule(self, self.base_path) - self.observer.start() + logger.debug(f"IoMBian Services Handler started.") + if self.mode == "offline": + self.observer = Observer() + self.observer.schedule(self, self.base_path) + self.observer.start() def stop(self): """Stop the handler by stopping the observer of the "iombian-services" folder. Also stop all of the services in the "iombian-services" folder, but not the composes. """ - logger.info("IoMBian Installed Services Handler stopped.") - self.observer.stop() + logger.debug("IoMBian Services Handler stopped.") + if self.observer: + self.observer.stop() for service in self.services: service.stop() def read_local_services(self): """Read the services in "iombian-services", start them and add them to the `services` list.""" + logger.debug("Reading local services") self.services = [] service_names = os.listdir(self.base_path) + logger.debug(f"{len(service_names)} local services detected") for service_name in service_names: service_path = f"{self.base_path}/{service_name}" - service = InstalledServiceHandler(service_path, self.wait_seconds) - service.start() + service = InstalledServiceHandler( + service_path, self.wait_seconds, self.on_local_service_state_changed) + service.start(self.mode) self.services.append(service) + def start_service(self, service_name: str): + """Start a service by it's name: up + + Method exposed so that external programs can interact with the services. + """ + service = self._get_service_by_name(service_name) + if service: + service.reload_compose_services() + return + service_path = "/".join([self.base_path, service_name]) + service = InstalledServiceHandler( + service_path, self.wait_seconds, self.on_local_service_state_changed) + service.up_compose_services() + service.start(self.mode) + self.services.append(service) + + def reconfigure_service(self, service_name: str): + """Reconfigure a service by it's name: stop and up + + Method exposed so that external programs can interact with the services. + """ + service = self._get_service_by_name(service_name) + if service: + service.stop_compose_services() + service.up_compose_services() + + def delete_service(self, service_name: str): + """delete a service by it's name: stop, down and clean + + Method exposed so that external programs can interact with the services. + """ + service = self._get_service_by_name(service_name) + if service: + service.stop() + service.down_compose_services(remove_images=True) + service.clean_compose_services() + self.services.remove(service) + def on_created(self, event: FileSystemEvent): """When a new service is added to "iombian-services", start the service and the compose of the services. @@ -75,9 +131,10 @@ def on_created(self, event: FileSystemEvent): service_path = event.src_path service_name = pathlib.Path(service_path).stem logger.debug(f"{service_name} service folder was created.") - service = InstalledServiceHandler(service_path, self.wait_seconds) - service.up() - service.start() + service = InstalledServiceHandler( + service_path, self.wait_seconds, self.on_local_service_state_changed) + service.up_compose_services() + service.start(self.mode) self.services.append(service) logger.debug(f"{service_path} service added.") @@ -94,10 +151,25 @@ def on_deleted(self, event: FileSystemEvent): service = self._get_service_by_name(service_name) if service: service.stop() - service.down() + service.clean_compose_services() self.services.remove(service) logger.debug(f"{service_name} service removed.") + def on_local_service_state_changed(self, service_name: str, new_state: str): + """Called when a local service changes it's state. + + :param service_name: + Name of the service that changed it's state. + :type event: + :class:`str` + :param new_state: + New state of the service. + :type event: + :class:`str` + """ + if self.local_service_state_change_callback: + threading.Thread(target=self.local_service_state_change_callback, args=[service_name, new_state]).start() + def _get_service_by_name(self, service_name: str): """Given the service name, return the service in "iombian-services".""" for service in self.services: diff --git a/src/main.py b/src/main.py index f43e4ca..f8963b2 100644 --- a/src/main.py +++ b/src/main.py @@ -2,10 +2,12 @@ import os import signal -import sdnotify - +from communication_module import CommunicationModule +from installed_services_remote_provider import InstalledServicesRemoteProvider from iombian_services_handler import IombianServicesHandler +CONFIG_HOST = os.environ.get("CONFIG_HOST", "127.0.0.1") +CONFIG_PORT = int(os.environ.get("CONFIG_PORT", 5555)) BASE_PATH = os.environ.get("BASE_PATH", "/opt/iombian-services") WAIT_SECONDS = int(os.environ.get("WAIT_SECONDS", 1)) LOG_LEVEL = os.environ.get("LOG_LEVEL", "INFO") @@ -16,16 +18,76 @@ logger = logging.getLogger(__name__) +def on_remote_connection_state_changed(state: str): + logger.debug(f"New remote connection state: {state}") + if state == "connected": + iombian_services_handler.set_mode("online") + iombian_services_handler.read_local_services() + iombian_services_handler.start() + elif state == "timeout": + logger.warning("The remote connection timeout, start the services handler in 'offline' mode") + iombian_services_handler.read_local_services() + iombian_services_handler.start() + + +def on_local_service_state_changed(service_name: str, new_state: str): + logger.debug(f"New state for local service: {service_name}: {new_state}") + if iombian_services_remote_provider: + iombian_services_remote_provider.update_remote_service_status(service_name, new_state) + + +def on_remote_service_state_changed(service_name: str, new_state: str): + logger.debug(f"New state for remote service: {service_name}: {new_state}") + if new_state == "downloaded": + logger.info(f"'{service_name}' service downloaded, starting it...") + iombian_services_handler.start_service(service_name) + elif new_state == "reconfigured": + logger.info(f"'{service_name}' service reconfigured, starting it...") + iombian_services_handler.reconfigure_service(service_name) + elif new_state == "updated": + logger.info(f"'{service_name}' service updated, starting it...") + iombian_services_handler.start_service(service_name) + elif new_state == "to-be-updated": + # Stopping the service and removing the images is needed before the old files are removed + logger.info(f"'{service_name}' service to be updated, starting it...") + iombian_services_handler.delete_service(service_name) + elif new_state == "to-be-uninstalled": + logger.info(f"'{service_name}' service to be uninstalled, stopping it...") + iombian_services_handler.delete_service(service_name) + if iombian_services_remote_provider: + iombian_services_remote_provider.update_remote_service_status(service_name, "to-be-removed") + + def signal_handler(sig, frame): + logger.info("Stopping IoMBian Installed Services Handler Service") iombian_services_handler.stop() + if iombian_services_remote_provider: + iombian_services_remote_provider.stop() if __name__ == "__main__": - iombian_services_handler = IombianServicesHandler(BASE_PATH, WAIT_SECONDS) - iombian_services_handler.read_local_services() - iombian_services_handler.start() - notifier = sdnotify.SystemdNotifier() - notifier.notify("READY=1") + logger.info("Starting IoMBian Installed Services Handler Service") + + comm_module = CommunicationModule(host=CONFIG_HOST, port=CONFIG_PORT) + comm_module.start() + + api_key = str(comm_module.execute_command("get_api_key")) + project_id = str(comm_module.execute_command("get_project_id")) + refresh_token = str(comm_module.execute_command("get_refresh_token")) + device_id = str(comm_module.execute_command("get_device_id")) + + iombian_services_handler = IombianServicesHandler( + BASE_PATH, WAIT_SECONDS, on_local_service_state_changed) + + if not (api_key and project_id and refresh_token and device_id): + logger.warning( + "Wasn't able to get the necessary information from the config file handler" + ) + iombian_services_handler.read_local_services() + iombian_services_handler.start() + else: + iombian_services_remote_provider = InstalledServicesRemoteProvider(api_key, project_id, refresh_token, device_id, on_remote_connection_state_changed, on_remote_service_state_changed) + iombian_services_remote_provider.start(timeout=30) signal.signal(signal.SIGINT, signal_handler) signal.signal(signal.SIGTERM, signal_handler)