#!/usr/bin/env python3 import sys import json import logging import time from elmo.api.client import ElmoClient from elmo.systems import ELMO_E_CONNECT, IESS_METRONET import elmo.query as q # Importa le eccezioni specifiche di Elmo try: from elmo.api.exceptions import ( LockNotAcquiredException, InvalidCredentialsException, PermissionDeniedException, CommandException, ParseException ) ELMO_EXCEPTIONS_AVAILABLE = True except ImportError: # Fallback per versioni più vecchie ELMO_EXCEPTIONS_AVAILABLE = False logging.warning("Modulo elmo.api.exceptions non disponibile, usando gestione errori generica") # Configurazione del logging logging.basicConfig( level=logging.DEBUG if len(sys.argv) > 1 and sys.argv[1] == "--debug" else logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[logging.StreamHandler()] ) logger = logging.getLogger("ElmoClient") def get_base_url(system): if system == "e-connect": return ELMO_E_CONNECT elif system == "metronet": return IESS_METRONET else: raise ValueError(f"Sistema non supportato: {system}") def handle_elmo_exception(e, operation="operazione"): """ Gestisce le eccezioni specifiche di Elmo e restituisce messaggi di errore appropriati """ if not ELMO_EXCEPTIONS_AVAILABLE: return f"Errore durante {operation}: {str(e)}" if isinstance(e, LockNotAcquiredException): return f"Sistema occupato da un'altra sessione (app mobile, web, ecc.). Riprova tra qualche minuto." elif isinstance(e, InvalidCredentialsException): return f"Credenziali non valide. Verifica username e password." elif isinstance(e, PermissionDeniedException): return f"Permessi insufficienti per eseguire {operation}." elif isinstance(e, CommandException): return f"Comando non riuscito: {str(e)}" elif isinstance(e, ParseException): return f"Errore nella risposta del server durante {operation}." else: return f"Errore durante {operation}: {str(e)}" def authenticate_with_retry(username, password, system, domain, max_retries=3): """ Autentica con retry logic per gestire conflitti di lock """ for attempt in range(max_retries): try: logger.debug(f"Tentativo di autenticazione {attempt + 1}/{max_retries}") base_url = get_base_url(system) client = ElmoClient(base_url=base_url, domain=domain) client.auth(username, password) return client, True, "Autenticazione riuscita" except Exception as e: error_message = handle_elmo_exception(e, "autenticazione") logger.warning(f"Tentativo {attempt + 1} fallito: {error_message}") # Se è un errore di lock, aspetta prima di riprovare if ELMO_EXCEPTIONS_AVAILABLE and isinstance(e, LockNotAcquiredException): if attempt < max_retries - 1: wait_time = (attempt + 1) * 5 # Aspetta 5, 10, 15 secondi logger.debug(f"Lock non acquisito, attendo {wait_time} secondi") time.sleep(wait_time) continue # Per altri errori o ultimo tentativo, restituisci l'errore if attempt == max_retries - 1: logger.error(f"Tutti i tentativi di autenticazione falliti: {error_message}") return None, False, error_message return None, False, "Errore sconosciuto durante l'autenticazione" def authenticate(username, password, system, domain): """Wrapper per mantenere compatibilità""" return authenticate_with_retry(username, password, system, domain) def get_status_with_retry(client, max_retries=3): """ Ottieni lo stato con retry logic per gestire conflitti temporanei """ for attempt in range(max_retries): try: logger.debug(f"Tentativo di lettura stato {attempt + 1}/{max_retries}") # Ottieni lo stato dei settori sectors = client.query(q.SECTORS) # Ottieni lo stato degli ingressi inputs = client.query(q.INPUTS) # Ottieni lo stato degli avvisi alerts = client.query(q.ALERTS) return { "sectors": sectors, "inputs": inputs, "alerts": alerts }, True, "Stato ottenuto con successo" except Exception as e: error_message = handle_elmo_exception(e, "lettura stato") logger.warning(f"Tentativo {attempt + 1} fallito: {error_message}") if attempt < max_retries - 1: wait_time = (attempt + 1) * 2 logger.debug(f"Attendo {wait_time} secondi prima del prossimo tentativo") time.sleep(wait_time) else: logger.error(f"Tutti i tentativi di lettura stato falliti: {error_message}") return None, False, error_message return None, False, "Errore sconosciuto durante la lettura dello stato" def get_status(client): """Wrapper per mantenere compatibilità""" return get_status_with_retry(client) def get_devices_with_retry(client, max_retries=3): """ Ottieni i dispositivi con retry logic """ for attempt in range(max_retries): try: logger.debug(f"Tentativo di lettura dispositivi {attempt + 1}/{max_retries}") # Ottieni lo stato degli ingressi (sensori) inputs = client.query(q.INPUTS) logger.debug(f"Risposta query INPUTS: {inputs}") logger.debug(f"Tipo di inputs: {type(inputs)}") devices = [] # Gestisci il formato corretto: dizionario con chiave 'inputs' if inputs and isinstance(inputs, dict) and 'inputs' in inputs: inputs_dict = inputs['inputs'] logger.debug(f"Trovato dizionario inputs con {len(inputs_dict)} elementi") for key, input_device in inputs_dict.items(): logger.debug(f"Dispositivo {key}: {input_device}") device_info = { "id": input_device.get('element', input_device.get('index', key)), "name": input_device.get('name', f"Dispositivo {input_device.get('element', key)}"), "type": "unknown", # Impostiamo sempre unknown per permettere configurazione manuale "status": input_device.get('status', False), "enabled": not input_device.get('excluded', False), "zone": input_device.get('area', None), "device_type": "sensor", "element": input_device.get('element'), "index": input_device.get('index'), "excluded": input_device.get('excluded', False) } devices.append(device_info) logger.debug(f"Dispositivo processato: {device_info}") else: logger.warning("Formato inputs non riconosciuto o inputs vuoto") logger.debug(f"Inputs ricevuto: {inputs}") logger.info(f"Trovati {len(devices)} dispositivi") if len(devices) == 0: logger.warning("ATTENZIONE: Nessun dispositivo trovato nel sistema Elmo") logger.warning("Verifica che il tuo sistema Elmo abbia dispositivi configurati") logger.warning("e che l'utente abbia i permessi per accedervi") return devices, True, "Dispositivi ottenuti con successo" except Exception as e: error_message = handle_elmo_exception(e, "lettura dispositivi") logger.warning(f"Tentativo {attempt + 1} fallito: {error_message}") logger.exception("Dettagli dell'errore:") if attempt < max_retries - 1: wait_time = (attempt + 1) * 2 logger.debug(f"Attendo {wait_time} secondi prima del prossimo tentativo") time.sleep(wait_time) else: logger.error(f"Tutti i tentativi di lettura dispositivi falliti: {error_message}") return [], False, error_message return [], False, "Errore sconosciuto durante la lettura dei dispositivi" def get_devices(client): """Wrapper per mantenere compatibilità""" return get_devices_with_retry(client) def get_status_with_devices(client): """ Ottieni lo stato completo del sistema inclusi i dispositivi """ try: # Ottieni lo stato dei settori sectors = client.query(q.SECTORS) # Ottieni lo stato degli ingressi inputs = client.query(q.INPUTS) # Ottieni lo stato degli avvisi alerts = client.query(q.ALERTS) # Ottieni i dispositivi devices, devices_success, devices_message = get_devices(client) return { "sectors": sectors, "inputs": inputs, "alerts": alerts, "devices": devices if devices_success else [] }, True, "Stato completo ottenuto con successo" except Exception as e: logger.error(f"Errore nell'ottenere lo stato completo: {e}") return None, False, str(e) def check_system_ready(client): """ Verifica se il sistema è pronto per ricevere comandi """ try: logger.debug("Verifica dello stato del sistema") status, success, message = get_status(client) if not success: return False, f"Impossibile verificare lo stato: {message}" # Verifica se ci sono allarmi attivi che potrebbero impedire l'operazione if status.get("alerts") and len(status["alerts"]) > 0: active_alerts = [alert for alert in status["alerts"] if alert.get("active", False)] if active_alerts: logger.warning(f"Trovati {len(active_alerts)} allarmi attivi") return False, f"Sistema non pronto: {len(active_alerts)} allarmi attivi" return True, "Sistema pronto" except Exception as e: logger.error(f"Errore nella verifica dello stato del sistema: {e}") return False, str(e) def arm_system(client, code, sectors=None, max_retries=5): """ Arma il sistema con retry logic migliorato per gestire conflitti di lock """ for attempt in range(max_retries): try: logger.debug(f"Tentativo di armamento {attempt + 1}/{max_retries}") # Verifica lo stato attuale prima di tentare l'armamento status, success, message = get_status_with_retry(client, 2) if not success: logger.warning(f"Impossibile verificare lo stato del sistema: {message}") # Attendi un momento prima di acquisire il lock wait_before_lock = 2 + (attempt * 2) # Aumenta il tempo di attesa ad ogni tentativo logger.debug(f"Attendo {wait_before_lock} secondi prima di acquisire il lock") time.sleep(wait_before_lock) with client.lock(code) as c: logger.debug("Lock acquisito con successo, procedendo con l'armamento") if sectors: logger.debug(f"Armamento settori specifici: {sectors}") c.arm(sectors=sectors) else: logger.debug("Armamento di tutti i settori") c.arm() logger.debug("Comando di armamento inviato") # Attendi un momento per permettere al sistema di processare il comando time.sleep(3) return True, "Sistema armato con successo" except Exception as e: error_message = handle_elmo_exception(e, "armamento") logger.warning(f"Tentativo {attempt + 1} fallito: {error_message}") # Gestione specifica per errori di lock if ELMO_EXCEPTIONS_AVAILABLE and isinstance(e, LockNotAcquiredException): if attempt < max_retries - 1: # Tempo di attesa progressivo per conflitti di lock wait_time = (attempt + 1) * 10 # 10, 20, 30, 40 secondi logger.info(f"Sistema occupato, attendo {wait_time} secondi prima del prossimo tentativo") time.sleep(wait_time) continue else: # Per altri errori, attesa più breve if attempt < max_retries - 1: wait_time = (attempt + 1) * 3 logger.debug(f"Attendo {wait_time} secondi prima del prossimo tentativo") time.sleep(wait_time) # Ultimo tentativo fallito if attempt == max_retries - 1: logger.error(f"Tutti i tentativi di armamento falliti: {error_message}") return False, f"Errore nell'armare il sistema dopo {max_retries} tentativi: {error_message}" def disarm_system(client, code, sectors=None, max_retries=5): """ Disarma il sistema con retry logic migliorato per gestire conflitti di lock """ for attempt in range(max_retries): try: logger.debug(f"Tentativo di disarmamento {attempt + 1}/{max_retries}") # Verifica lo stato attuale prima di tentare il disarmamento status, success, message = get_status_with_retry(client, 2) if not success: logger.warning(f"Impossibile verificare lo stato del sistema: {message}") # Attendi un momento prima di acquisire il lock wait_before_lock = 2 + (attempt * 2) logger.debug(f"Attendo {wait_before_lock} secondi prima di acquisire il lock") time.sleep(wait_before_lock) with client.lock(code) as c: logger.debug("Lock acquisito con successo, procedendo con il disarmamento") if sectors: logger.debug(f"Disarmamento settori specifici: {sectors}") c.disarm(sectors=sectors) else: logger.debug("Disarmamento di tutti i settori") c.disarm() logger.debug("Comando di disarmamento inviato") # Attendi un momento per permettere al sistema di processare il comando time.sleep(3) return True, "Sistema disarmato con successo" except Exception as e: error_message = handle_elmo_exception(e, "disarmamento") logger.warning(f"Tentativo {attempt + 1} fallito: {error_message}") # Gestione specifica per errori di lock if ELMO_EXCEPTIONS_AVAILABLE and isinstance(e, LockNotAcquiredException): if attempt < max_retries - 1: # Tempo di attesa progressivo per conflitti di lock wait_time = (attempt + 1) * 10 # 10, 20, 30, 40 secondi logger.info(f"Sistema occupato, attendo {wait_time} secondi prima del prossimo tentativo") time.sleep(wait_time) continue else: # Per altri errori, attesa più breve if attempt < max_retries - 1: wait_time = (attempt + 1) * 3 logger.debug(f"Attendo {wait_time} secondi prima del prossimo tentativo") time.sleep(wait_time) # Ultimo tentativo fallito if attempt == max_retries - 1: logger.error(f"Tutti i tentativi di disarmamento falliti: {error_message}") return False, f"Errore nel disarmare il sistema dopo {max_retries} tentativi: {error_message}" def main(): if len(sys.argv) < 2: print("Uso: python elmo_client.py [argomenti...]") sys.exit(1) command = sys.argv[1] if command == "authenticate": if len(sys.argv) < 6: print("Uso: python elmo_client.py authenticate ") sys.exit(1) username = sys.argv[2] password = sys.argv[3] system = sys.argv[4] domain = sys.argv[5] _, success, message = authenticate(username, password, system, domain) result = {"success": success, "message": message} print(json.dumps(result)) elif command == "status": if len(sys.argv) < 6: print("Uso: python elmo_client.py status ") sys.exit(1) username = sys.argv[2] password = sys.argv[3] system = sys.argv[4] domain = sys.argv[5] client, success, message = authenticate(username, password, system, domain) if not success: result = {"success": False, "message": message} print(json.dumps(result)) sys.exit(1) status, success, message = get_status(client) result = {"success": success, "message": message, "status": status} print(json.dumps(result)) elif command == "arm": if len(sys.argv) < 7: print("Uso: python elmo_client.py arm [sectors]") sys.exit(1) username = sys.argv[2] password = sys.argv[3] system = sys.argv[4] domain = sys.argv[5] code = sys.argv[6] sectors = None if len(sys.argv) > 7: sectors = [int(s) for s in sys.argv[7].split(",")] client, success, message = authenticate(username, password, system, domain) if not success: result = {"success": False, "message": message} print(json.dumps(result)) sys.exit(1) success, message = arm_system(client, code, sectors) result = {"success": success, "message": message} print(json.dumps(result)) elif command == "disarm": if len(sys.argv) < 7: print("Uso: python elmo_client.py disarm [sectors]") sys.exit(1) username = sys.argv[2] password = sys.argv[3] system = sys.argv[4] domain = sys.argv[5] code = sys.argv[6] sectors = None if len(sys.argv) > 7: sectors = [int(s) for s in sys.argv[7].split(",")] client, success, message = authenticate(username, password, system, domain) if not success: result = {"success": False, "message": message} print(json.dumps(result)) sys.exit(1) success, message = disarm_system(client, code, sectors) result = {"success": success, "message": message} print(json.dumps(result)) elif command == "devices": if len(sys.argv) < 6: print("Uso: python elmo_client.py devices ") sys.exit(1) username = sys.argv[2] password = sys.argv[3] system = sys.argv[4] domain = sys.argv[5] client, success, message = authenticate(username, password, system, domain) if not success: result = {"success": False, "message": message} print(json.dumps(result)) sys.exit(1) devices, success, message = get_devices(client) result = {"success": success, "message": message, "devices": devices} print(json.dumps(result)) elif command == "status_with_devices": if len(sys.argv) < 6: print("Uso: python elmo_client.py status_with_devices ") sys.exit(1) username = sys.argv[2] password = sys.argv[3] system = sys.argv[4] domain = sys.argv[5] client, success, message = authenticate(username, password, system, domain) if not success: result = {"success": False, "message": message} print(json.dumps(result)) sys.exit(1) status, success, message = get_status_with_devices(client) result = {"success": success, "message": message, "status": status} print(json.dumps(result)) else: print(f"Comando non riconosciuto: {command}") sys.exit(1) if __name__ == "__main__": main()