From caf698bfd5cff018c36004c61d96e8bc6a8660d6 Mon Sep 17 00:00:00 2001 From: roberto Date: Wed, 9 Jul 2025 18:43:25 +0200 Subject: [PATCH] cron_include.py --- Python/cron_include.py | 434 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 434 insertions(+) create mode 100644 Python/cron_include.py diff --git a/Python/cron_include.py b/Python/cron_include.py new file mode 100644 index 0000000..b3d7662 --- /dev/null +++ b/Python/cron_include.py @@ -0,0 +1,434 @@ +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.")