Delete cron_import.py
This commit is contained in:
434
cron_import.py
434
cron_import.py
@@ -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.")
|
||||
Reference in New Issue
Block a user