From cb6220a059caea52af128d9edc60ff67ae9ba6b3 Mon Sep 17 00:00:00 2001 From: roberto Date: Wed, 9 Jul 2025 18:56:43 +0200 Subject: [PATCH] Delete cron_import.py --- cron_import.py | 434 ------------------------------------------------- 1 file changed, 434 deletions(-) delete mode 100644 cron_import.py diff --git a/cron_import.py b/cron_import.py deleted file mode 100644 index b3d7662..0000000 --- a/cron_import.py +++ /dev/null @@ -1,434 +0,0 @@ -import os -import re -import logging -from crontab import CronTab, CronItem # Importa le classi necessarie da python-crontab - -# --- Mock/Placeholder per le dipendenze esterne --- -# In un'applicazione reale, queste funzioni verrebbero fornite dal tuo sistema piGarden principale. - -# Configura un logger di base per le funzioni di log -logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') - -def log_write(log_type, level, message): - """ - Simula la funzione log_write dal tuo script Bash. - In un'applicazione reale, useresti il modulo logging di Python. - """ - if level == "info": - logging.info(f"[{log_type}] {message}") - elif level == "warning": - logging.warning(f"[{log_type}] {message}") - elif level == "error": - logging.error(f"[{log_type}] {message}") - else: - logging.debug(f"[{log_type}] {message}") - -def trigger_event(event_name, *args): - """ - Simula la funzione trigger_event dal tuo script Bash. - """ - log_write("event", "info", f"Triggered event: {event_name} with args: {args}") - # Qui potresti aggiungere la logica per chiamare handler di eventi reali - -def alias_exists(alias_name): - """ - Simula la funzione alias_exists. - Dovrebbe essere integrata con la tua configurazione delle elettrovalvole. - Per questo esempio, restituisce True solo per alias "1" a "6". - """ - try: - num = int(alias_name) - return 1 <= num <= EV_TOTAL # Assumiamo EV_TOTAL sia definito globalmente o passato - except ValueError: - return False - -# EV_TOTAL deve essere definito o passato, simuliamo un valore qui -EV_TOTAL = 6 -# Percorso dello script principale (es. piGarden.py) -# Questo dovrebbe essere il percorso assoluto del tuo script piGarden principale -# In un'applicazione reale, lo passeresti dalla tua classe PiGarden -PI_GARDEN_SCRIPT_PATH = "/home/pi/piGarden/piGarden.py" - -# --- Classe CronManager --- - -class CronManager: - def __init__(self, script_path, ev_total_val, log_writer, event_trigger, alias_checker): - self.script_path = script_path - self.ev_total = ev_total_val - self.log_write = log_writer - self.trigger_event = event_trigger - self.alias_exists = alias_checker - self.cron_user = True # Gestisce il crontab dell'utente corrente - - def _get_crontab(self): - """Ottiene l'oggetto CronTab per l'utente corrente.""" - try: - return CronTab(user=self.cron_user) - except Exception as e: - self.log_write("cron", "error", f"Impossibile accedere al crontab: {e}") - raise - - def cron_del(self, cron_type, cron_arg=""): - """ - Elimina una tipologia di schedulazione dal crontab dell'utente. - :param cron_type: Tipologia del cron (es. "init", "open", "close"). - :param cron_arg: Argomento della tipologia (es. alias dell'elettrovalvola). - """ - if not cron_type: - self.log_write("cron", "error", "Tipo cron vuoto") - print("Tipo cron vuoto", file=os.sys.stderr) - return False - - crontab = self._get_crontab() - jobs_to_remove = [] - - # Il tuo script Bash usa commenti START/END. - # Possiamo cercare lavori che contengono questi commenti. - # Alternativamente, si potrebbe assegnare un commento specifico ad ogni job creato. - start_comment_pattern = re.compile(rf"^# START cron {re.escape(cron_type)} {re.escape(cron_arg)}$") - end_comment_pattern = re.compile(rf"^# END cron {re.escape(cron_type)} {re.escape(cron_arg)}$") - - found_block = False - in_block = False - for job in list(crontab.jobs): # Iterate over a copy because we might modify - if start_comment_pattern.match(job.comment or ""): - in_block = True - found_block = True - jobs_to_remove.append(job) # Include the START comment job itself - elif end_comment_pattern.match(job.comment or ""): - if in_block: - jobs_to_remove.append(job) # Include the END comment job itself - in_block = False - elif in_block: - jobs_to_remove.append(job) - - if not found_block: - print(f"{cron_type} {cron_arg} cron non presente", file=os.sys.stderr) - return True # Considerato un successo se non c'è nulla da eliminare - - self.trigger_event("cron_del_before", cron_type, cron_arg) - - for job in jobs_to_remove: - crontab.remove(job) - - try: - crontab.write() - self.log_write("cron", "info", f"Cron '{cron_type} {cron_arg}' eliminato con successo.") - self.trigger_event("cron_del_after", cron_type, cron_arg) - return True - except Exception as e: - self.log_write("cron", "error", f"Errore durante la scrittura del crontab: {e}") - print(f"Errore durante la scrittura del crontab: {e}", file=os.sys.stderr) - return False - - def _get_cron_command(self, cron_type, cron_arg, cron_arg2): - """Determina il comando Bash da eseguire per il cron job.""" - base_command = f"{self.script_path}" - - if cron_type == "init": - return f"{base_command} init" - elif cron_type == "start_socket_server": - return f"{base_command} start_socket_server force" - elif cron_type == "check_rain_online": - return f"{base_command} check_rain_online 2> /tmp/check_rain_online.err" - elif cron_type == "check_rain_sensor": - return f"{base_command} check_rain_sensor 2> /tmp/check_rain_sensor.err" - elif cron_type == "close_all_for_rain": - return f"{base_command} close_all_for_rain 2> /tmp/close_all_for_rain.err 1> /dev/null" - elif cron_type == "open": - return f"{base_command} open {cron_arg}" - elif cron_type == "open_in": - return f"{base_command} open {cron_arg} {cron_arg2}" - elif cron_type == "open_in_stop": - return f"{base_command} close {cron_arg}" - elif cron_type == "close": - return f"{base_command} close {cron_arg}" - else: - self.log_write("cron", "error", f"Tipo cron errato: {cron_type}") - print(f"Tipo cron errato: {cron_type}", file=os.sys.stderr) - raise ValueError(f"Tipo cron errato: {cron_type}") - - def cron_add(self, cron_type, minute="*", hour="*", dom="*", month="*", dow="*", cron_arg="", cron_arg2=""): - """ - Aggiunge una schedulazione nel crontab dell'utente. - :param cron_type: Tipologia del cron. - :param minute: Minuto (0-59, *, */n, @reboot). - :param hour: Ora (0-23, *, */n). - :param dom: Giorno del mese (1-31, *, */n). - :param month: Mese (1-12, *, */n). - :param dow: Giorno della settimana (0-6, *, */n). - :param cron_arg: Primo argomento specifico della tipologia. - :param cron_arg2: Secondo argomento specifico della tipologia (es. "disabled"). - """ - if not cron_type: - self.log_write("cron", "error", "Tipo cron vuoto") - print("Tipo cron vuoto", file=os.sys.stderr) - return False - - # Elimina prima qualsiasi blocco esistente per garantire l'idempotenza - self.cron_del(cron_type, cron_arg) - - crontab = self._get_crontab() - - # Determina il comando e se deve essere disabilitato - cron_command = self._get_cron_command(cron_type, cron_arg, cron_arg2) - cron_disabled = (cron_arg2 == "disabled") - - # Crea i commenti START e END per il blocco - start_comment = f"# START cron {cron_type} {cron_arg}" - end_comment = f"# END cron {cron_type} {cron_arg}" - - # Aggiungi il commento START - job_start = crontab.new(command=f"echo '{start_comment}'", comment=start_comment) - job_start.minute.every(1) # Un cron job fittizio per il commento START - job_start.enabled = False # Disabilita il job commento - - # Aggiungi il job principale - job = crontab.new(command=cron_command) - if minute == "@reboot": - job.set_every("reboot") - else: - job.minute.on(minute) - job.hour.on(hour) - job.dom.on(dom) - job.month.on(month) - job.dow.on(dow) - job.enabled = not cron_disabled - job.comment = f"piGarden {cron_type} {cron_arg}" # Un commento più descrittivo per il job reale - - # Aggiungi il commento END - job_end = crontab.new(command=f"echo '{end_comment}'", comment=end_comment) - job_end.minute.every(1) # Un cron job fittizio per il commento END - job_end.enabled = False # Disabilita il job commento - - try: - crontab.write() - self.log_write("cron", "info", f"Cron '{cron_type} {cron_arg}' aggiunto con successo: {job.render()}") - self.trigger_event("cron_add_after", cron_type, cron_arg, job.render()) - return True - except Exception as e: - self.log_write("cron", "error", f"Errore durante la scrittura del crontab: {e}") - print(f"Errore durante la scrittura del crontab: {e}", file=os.sys.stderr) - return False - - def cron_get(self, cron_type, cron_arg=""): - """ - Legge una tipologia di schedulazione dal crontab dell'utente. - :param cron_type: Tipologia del cron. - :param cron_arg: Argomento della tipologia. - :return: Stringa contenente le schedulazioni trovate, separate da newline. - """ - if not cron_type: - self.log_write("cron", "error", "Tipo cron vuoto") - print("Tipo cron vuoto", file=os.sys.stderr) - return "" - - crontab = self._get_crontab() - found_jobs = [] - - # Cerca i job principali che corrispondono al tipo e all'argomento - for job in crontab.jobs: - if job.comment and job.comment.startswith(f"piGarden {cron_type} {cron_arg}"): - found_jobs.append(job.render()) - - return "\n".join(found_jobs) - - # --- Funzioni wrapper per tipi di cron specifici --- - - def set_cron_init(self): - self.cron_del("init") # Assicurati che non ci siano duplicati - self.cron_add("init", minute="@reboot") - - def del_cron_init(self): - self.cron_del("init") - - def set_cron_start_socket_server(self): - self.cron_del("start_socket_server") - self.cron_add("start_socket_server", minute="@reboot") - - def del_cron_start_socket_server(self): - self.cron_del("start_socket_server") - - def set_cron_check_rain_sensor(self): - self.cron_del("check_rain_sensor") - self.cron_add("check_rain_sensor", minute="*") # Ogni minuto - - def del_cron_check_rain_sensor(self): - self.cron_del("check_rain_sensor") - - def set_cron_check_rain_online(self): - self.cron_del("check_rain_online") - self.cron_add("check_rain_online", minute="*/3") # Ogni 3 minuti - - def del_cron_check_rain_online(self): - self.cron_del("check_rain_online") - - def set_cron_close_all_for_rain(self): - self.cron_del("close_all_for_rain") - self.cron_add("close_all_for_rain", minute="*/5") # Ogni 5 minuti - - def del_cron_close_all_for_rain(self): - self.cron_del("close_all_for_rain") - - def add_cron_open(self, alias, minute, hour, dom, month, dow, disabled=""): - if not self.alias_exists(alias): - self.log_write("cron", "error", f"Alias '{alias}' non trovato") - print(f"Alias '{alias}' non trovato", file=os.sys.stderr) - return False - self.cron_add("open", minute, hour, dom, month, dow, alias, disabled) - return True - - def del_cron_open(self, alias): - if not self.alias_exists(alias): - self.log_write("cron", "error", f"Alias '{alias}' non trovato") - print(f"Alias '{alias}' non trovato", file=os.sys.stderr) - return False - self.cron_del("open", alias) - return True - - def get_cron_open(self, alias): - if not self.alias_exists(alias): - self.log_write("cron", "error", f"Alias '{alias}' non trovato") - print(f"Alias '{alias}' non trovato", file=os.sys.stderr) - return "" - return self.cron_get("open", alias) - - def del_cron_open_in(self, alias): - if not self.alias_exists(alias): - self.log_write("cron", "error", f"Alias '{alias}' non trovato") - print(f"Alias '{alias}' non trovato", file=os.sys.stderr) - return False - self.cron_del("open_in", alias) - self.cron_del("open_in_stop", alias) - return True - - def get_cron_close(self, alias): - if not self.alias_exists(alias): - self.log_write("cron", "error", f"Alias '{alias}' non trovato") - print(f"Alias '{alias}' non trovato", file=os.sys.stderr) - return "" - return self.cron_get("close", alias) - - def add_cron_close(self, alias, minute, hour, dom, month, dow, disabled=""): - if not self.alias_exists(alias): - self.log_write("cron", "error", f"Alias '{alias}' non trovato") - print(f"Alias '{alias}' non trovato", file=os.sys.stderr) - return False - self.cron_add("close", minute, hour, dom, month, dow, alias, disabled) - return True - - def del_cron_close(self, alias): - if not self.alias_exists(alias): - self.log_write("cron", "error", f"Alias '{alias}' non trovato") - print(f"Alias '{alias}' non trovato", file=os.sys.stderr) - return False - self.cron_del("close", alias) - return True - - def cron_disable_all_open_close(self): - crontab = self._get_crontab() - for i in range(1, self.ev_total + 1): - alias = str(i) # Assumendo che gli alias siano i numeri delle EV - # Disabilita le schedulazioni di apertura - for job in list(crontab.jobs): - if job.comment and job.comment.startswith(f"piGarden open {alias}") and job.enabled: - job.enabled = False - self.log_write("cron", "info", f"Disabilitato cron 'open' per alias {alias}: {job.render()}") - - # Disabilita le schedulazioni di chiusura - for job in list(crontab.jobs): - if job.comment and job.comment.startswith(f"piGarden close {alias}") and job.enabled: - job.enabled = False - self.log_write("cron", "info", f"Disabilitato cron 'close' per alias {alias}: {job.render()}") - try: - crontab.write() - self.log_write("cron", "info", "Tutte le schedulazioni di apertura/chiusura disabilitate.") - return True - except Exception as e: - self.log_write("cron", "error", f"Errore durante la disabilitazione dei cron: {e}") - return False - - def cron_enable_all_open_close(self): - crontab = self._get_crontab() - for i in range(1, self.ev_total + 1): - alias = str(i) # Assumendo che gli alias siano i numeri delle EV - # Abilita le schedulazioni di apertura - for job in list(crontab.jobs): - if job.comment and job.comment.startswith(f"piGarden open {alias}") and not job.enabled: - job.enabled = True - self.log_write("cron", "info", f"Abilitato cron 'open' per alias {alias}: {job.render()}") - - # Abilita le schedulazioni di chiusura - for job in list(crontab.jobs): - if job.comment and job.comment.startswith(f"piGarden close {alias}") and not job.enabled: - job.enabled = True - self.log_write("cron", "info", f"Abilitato cron 'close' per alias {alias}: {job.render()}") - try: - crontab.write() - self.log_write("cron", "info", "Tutte le schedulazioni di apertura/chiusura abilitate.") - return True - except Exception as e: - self.log_write("cron", "error", f"Errore durante l'abilitazione dei cron: {e}") - return False - -# --- Esempio di utilizzo (per testare la classe CronManager) --- -if __name__ == "__main__": - # Inizializza il gestore cron con le dipendenze mock - cron_manager = CronManager( - script_path=PI_GARDEN_SCRIPT_PATH, - ev_total_val=EV_TOTAL, - log_writer=log_write, - event_trigger=trigger_event, - alias_checker=alias_exists - ) - - print("--- Test Cron Manager ---") - - # Esempio: Aggiungi un cron per l'inizializzazione - print("\nAggiungo cron 'init'...") - cron_manager.set_cron_init() - - # Esempio: Aggiungi un cron per aprire l'elettrovalvola "1" ogni giorno alle 7:00 - print("\nAggiungo cron 'open' per EV 1 alle 07:00...") - cron_manager.add_cron_open("1", "0", "7", "*", "*", "*") - - # Esempio: Aggiungi un cron per chiudere l'elettrovalvola "2" ogni 5 minuti (disabilitato) - print("\nAggiungo cron 'close' per EV 2 ogni 5 minuti (disabilitato)...") - cron_manager.add_cron_close("2", "*/5", "*", "*", "*", "*", "disabled") - - # Esempio: Ottieni i cron per l'elettrovalvola "1" - print("\nCron 'open' per EV 1:") - print(cron_manager.get_cron_open("1")) - - # Esempio: Ottieni i cron per l'elettrovalvola "2" - print("\nCron 'close' per EV 2:") - print(cron_manager.get_cron_close("2")) - - # Esempio: Disabilita tutti i cron di apertura/chiusura - print("\nDisabilito tutti i cron di apertura/chiusura...") - cron_manager.cron_disable_all_open_close() - - # Verifica lo stato dopo la disabilitazione - print("\nCron 'open' per EV 1 dopo disabilitazione:") - print(cron_manager.get_cron_open("1")) # Dovrebbe mostrare il job ma disabilitato - - # Esempio: Abilita tutti i cron di apertura/chiusura - print("\nAbilito tutti i cron di apertura/chiusura...") - cron_manager.cron_enable_all_open_close() - - # Verifica lo stato dopo l'abilitazione - print("\nCron 'open' per EV 1 dopo abilitazione:") - print(cron_manager.get_cron_open("1")) # Dovrebbe mostrare il job abilitato - - # Esempio: Elimina un cron specifico - print("\nElimino cron 'init'...") - cron_manager.del_cron_init() - - print("\nElimino cron 'open' per EV 1...") - cron_manager.del_cron_open("1") - - print("\nElimino cron 'close' per EV 2...") - cron_manager.del_cron_close("2") - - print("\n--- Test Completato ---") - print("Controlla il tuo crontab con 'crontab -l' per vedere le modifiche.")