diff --git a/custom_components/pun_sensor/config_flow.py b/custom_components/pun_sensor/config_flow.py index cac837a..0a8fce3 100644 --- a/custom_components/pun_sensor/config_flow.py +++ b/custom_components/pun_sensor/config_flow.py @@ -32,7 +32,7 @@ def __init__(self, entry: config_entries.ConfigEntry) -> None: """Inizializzazione opzioni.""" if AwesomeVersion(HA_VERSION) < AwesomeVersion("2024.12.0b0"): self.config_entry = entry - + async def async_step_init(self, user_input=None) -> FlowResult: """Gestisce le opzioni di configurazione.""" errors: dict[str, str] | None = {} diff --git a/custom_components/pun_sensor/coordinator.py b/custom_components/pun_sensor/coordinator.py index d830fcb..7e13e13 100644 --- a/custom_components/pun_sensor/coordinator.py +++ b/custom_components/pun_sensor/coordinator.py @@ -29,8 +29,8 @@ EVENT_UPDATE_PUN, WEB_RETRIES_MINUTES, ) -from .interfaces import DEFAULT_ZONA, Fascia, PunData, PunValues, Zona -from .utils import extract_xml, get_fascia, get_hour_datetime, get_next_date +from .interfaces import DEFAULT_ZONA, Fascia, PunData, PunValues, PunDataMP, PunValuesMP, Zona +from .utils import extract_xml, extract_xml2, get_fascia, get_hour_datetime, get_next_date # Ottiene il logger _LOGGER = logging.getLogger(__name__) @@ -67,6 +67,8 @@ def __init__(self, hass: HomeAssistant, config: ConfigEntry) -> None: # Inizializza i dati PUN e la zona geografica self.pun_data: PunData = PunData() + # Inizializza i dati PUN e la zona geografica + self.pun_data_mp: PunDataMP = PunDataMP() try: # Estrae il valore dalla configurazione come stringa zona_string = config.options.get( @@ -77,6 +79,15 @@ def __init__(self, hass: HomeAssistant, config: ConfigEntry) -> None: # (per verificare che sia corretta) self.pun_data.zona = Zona[zona_string] + # Estrae il valore dalla configurazione come stringa + zona_string = config.options.get( + CONF_ZONA, config.data.get(CONF_ZONA, DEFAULT_ZONA) + ) + + # Tenta di associare la stringa all'enum + # (per verificare che sia corretta) + self.pun_data_mp.zona = Zona[zona_string] + except KeyError: # La zona non è valida _LOGGER.error( @@ -88,6 +99,7 @@ def __init__(self, hass: HomeAssistant, config: ConfigEntry) -> None: # Aggiorna la configurazione salvata con la zona di default # (per la prossima esecuzione) self.pun_data.zona = DEFAULT_ZONA + self.pun_data_mp.zona = DEFAULT_ZONA @callback async def async_restore_default_zona() -> None: @@ -127,6 +139,7 @@ async def async_restore_default_zona() -> None: self.web_retries = WEB_RETRIES_MINUTES self.schedule_token = None self.pun_values: PunValues = PunValues() + self.pun_values_mp: PunValuesMP = PunValuesMP() self.fascia_corrente: Fascia | None = None self.fascia_successiva: Fascia | None = None self.prossimo_cambio_fascia: datetime | None = None @@ -174,20 +187,26 @@ def async_update_entry() -> None: # Carica i minuti dalla configurazione self.scan_minute = config.data.get(CONF_SCAN_MINUTE, 0) - async def _async_update_data(self): + async def _async_update_data(self, mp): """Aggiornamento dati a intervalli prestabiliti.""" # Calcola l'intervallo di date per il mese corrente - date_end = dt_util.now().date() - date_start = date(date_end.year, date_end.month, 1) + if mp=="N": + # Calcola l'intervallo di date per il mese corrente + date_end = dt_util.now().date() + date_start = date(date_end.year, date_end.month, 1) + else: + date_end = dt_util.now().date().replace(day=1) - timedelta(days=1) + date_start = date(date_end.year, date_end.month, 1) # All'inizio del mese, aggiunge i valori del mese precedente # a meno che CONF_ACTUAL_DATA_ONLY non sia impostato - if (not self.actual_data_only) and (date_end.day < 4): + if (not self.actual_data_only) and (date_end.day < 4) and mp=="N": date_start = date_start - timedelta(days=3) - + # Aggiunge un giorno (domani) per il calcolo del prezzo zonale - date_end += timedelta(days=1) + if mp=="N": + date_end += timedelta(days=1) # Converte le date in stringa da passare all'API Mercato elettrico start_date_param = date_start.strftime("%Y%m%d") @@ -244,55 +263,102 @@ async def _async_update_data(self): ", ".join(str(fn) for fn in archive.namelist()), ) - # Estrae i dati dall'archivio - self.pun_data = extract_xml( - archive, self.pun_data, dt_util.now(time_zone=tz_pun).date() - ) - archive.close() - - # Per ogni fascia, calcola il valore del pun - for fascia, value_list in self.pun_data.pun.items(): - # Se abbiamo valori nella fascia - if len(value_list) > 0: - # Calcola la media dei pun e aggiorna il valore del pun attuale - # per la fascia corrispondente - self.pun_values.value[fascia] = mean(self.pun_data.pun[fascia]) + if mp == "N": + # Estrae i dati dall'archivio + self.pun_data = extract_xml( + archive, self.pun_data, dt_util.now(time_zone=tz_pun).date() + ) + archive.close() + + # Per ogni fascia, calcola il valore del pun + for fascia, value_list in self.pun_data.pun.items(): + # Se abbiamo valori nella fascia + if len(value_list) > 0: + # Calcola la media dei pun e aggiorna il valore del pun attuale + # per la fascia corrispondente + self.pun_values.value[fascia] = mean(self.pun_data.pun[fascia]) + else: + # Skippiamo i dict se vuoti + pass + + # Calcola la fascia F23 (a partire da F2 ed F3) + # NOTA: la motivazione del calcolo è oscura ma sembra corretta; vedere: + # https://github.com/virtualdj/pun_sensor/issues/24#issuecomment-1829846806 + if ( + len(self.pun_data.pun[Fascia.F2]) and len(self.pun_data.pun[Fascia.F3]) + ) > 0: + self.pun_values.value[Fascia.F23] = ( + 0.46 * self.pun_values.value[Fascia.F2] + + 0.54 * self.pun_values.value[Fascia.F3] + ) else: - # Skippiamo i dict se vuoti - pass - - # Calcola la fascia F23 (a partire da F2 ed F3) - # NOTA: la motivazione del calcolo è oscura ma sembra corretta; vedere: - # https://github.com/virtualdj/pun_sensor/issues/24#issuecomment-1829846806 - if ( - len(self.pun_data.pun[Fascia.F2]) and len(self.pun_data.pun[Fascia.F3]) - ) > 0: - self.pun_values.value[Fascia.F23] = ( - 0.46 * self.pun_values.value[Fascia.F2] - + 0.54 * self.pun_values.value[Fascia.F3] + self.pun_values.value[Fascia.F23] = 0 + + # Logga i dati + _LOGGER.debug( + "Numero di dati: %s", + ", ".join( + str(f"{len(dati)} ({fascia.value})") + for fascia, dati in self.pun_data.pun.items() + if fascia != Fascia.F23 + ), + ) + _LOGGER.debug( + "Valori PUN: %s", + ", ".join( + f"{prezzo} ({fascia.value})" + for fascia, prezzo in self.pun_values.value.items() + ), ) - else: - self.pun_values.value[Fascia.F23] = 0 - - # Logga i dati - _LOGGER.debug( - "Numero di dati: %s", - ", ".join( - str(f"{len(dati)} ({fascia.value})") - for fascia, dati in self.pun_data.pun.items() - if fascia != Fascia.F23 - ), - ) - _LOGGER.debug( - "Valori PUN: %s", - ", ".join( - f"{prezzo} ({fascia.value})" - for fascia, prezzo in self.pun_values.value.items() - ), - ) - # Notifica che i dati PUN (prezzi) sono stati aggiornati - self.async_set_updated_data({COORD_EVENT: EVENT_UPDATE_PUN}) + # Notifica che i dati PUN (prezzi) sono stati aggiornati + self.async_set_updated_data({COORD_EVENT: EVENT_UPDATE_PUN}) + else: + # Estrae i dati dall'archivio + self.pun_data_mp = extract_xml2(archive, self.pun_data_mp, dt_util.now(time_zone=tz_pun).date()) + + # Per ogni fascia, calcola il valore del pun + for fascia, value_list in self.pun_data_mp.pun.items(): + # Se abbiamo valori nella fascia + if len(value_list) > 0: + # Calcola la media dei pun e aggiorna il valore del pun attuale + # per la fascia corrispondente + self.pun_values_mp.value[fascia] = mean(self.pun_data_mp.pun[fascia]) + else: + # Skippiamo i dict se vuoti + pass + + # Calcola la fascia F23 (a partire da F2 ed F3) + # NOTA: la motivazione del calcolo è oscura ma sembra corretta; vedere: + # https://github.com/virtualdj/pun_sensor/issues/24#issuecomment-1829846806 + if ( + len(self.pun_data_mp.pun[Fascia.F2_MP]) and len(self.pun_data_mp.pun[Fascia.F3_MP]) + ) > 0: + self.pun_values_mp.value[Fascia.F23_MP] = ( + 0.46 * self.pun_values_mp.value[Fascia.F2_MP] + + 0.54 * self.pun_values_mp.value[Fascia.F3_MP] + ) + else: + self.pun_values_mp.value[Fascia.F23_MP] = 0 + + # Logga i dati + _LOGGER.debug( + "Numero di dati: %s", + ", ".join( + str(f"{len(dati)} ({fascia.value})") + for fascia, dati in self.pun_data_mp.pun.items() + if fascia != Fascia.F23_MP + ), + ) + _LOGGER.debug( + "Valori PUN: %s", + ", ".join( + f"{prezzo} ({fascia.value})" + for fascia, prezzo in self.pun_values_mp.value.items() + ), + ) + # Notifica che i dati PUN (prezzi) sono stati aggiornati + self.async_set_updated_data({COORD_EVENT: EVENT_UPDATE_PUN}) async def update_fascia(self, now=None): """Aggiorna la fascia oraria corrente (al cambio fascia).""" @@ -336,7 +402,8 @@ async def update_pun(self, now=None): # Aggiorna i dati da web try: # Esegue l'aggiornamento - await self._async_update_data() + await self._async_update_data("N") + await self._async_update_data("Y") # Se non ci sono eccezioni, ha avuto successo # Ricarica i tentativi per la prossima esecuzione diff --git a/custom_components/pun_sensor/interfaces.py b/custom_components/pun_sensor/interfaces.py index 330561b..6e4bea1 100644 --- a/custom_components/pun_sensor/interfaces.py +++ b/custom_components/pun_sensor/interfaces.py @@ -21,7 +21,24 @@ def __init__(self) -> None: self.prezzi_zonali: dict[str, float | None] = {} self.pun_orari: dict[str, float | None] = {} +class PunDataMP: + """Classe che contiene i valori del PUN orario per ciascuna fascia.""" + + def __init__(self) -> None: + """Inizializza le liste di ciascuna fascia.""" + self.pun: dict[Fascia, list[float]] = { + Fascia.MONO_MP: [], + Fascia.F1_MP: [], + Fascia.F2_MP: [], + Fascia.F3_MP: [], + Fascia.F23_MP: [], + } + + self.zona: Zona | None = None + self.prezzi_zonali: dict[str, float | None] = {} + self.pun_orari: dict[str, float | None] = {} + class Fascia(Enum): """Enumerazione con i tipi di fascia oraria.""" @@ -30,6 +47,11 @@ class Fascia(Enum): F2 = "F2" F3 = "F3" F23 = "F23" + MONO_MP = "MONO_MP" + F1_MP = "F1_MP" + F2_MP = "F2_MP" + F3_MP = "F3_MP" + F23_MP = "F23_MP" class PunValues: @@ -44,6 +66,17 @@ class PunValues: Fascia.F23: 0.0, } +class PunValuesMP: + """Classe che contiene il PUN del mese precedente di ciascuna fascia.""" + + value: dict[Fascia, float] + value = { + Fascia.MONO_MP: 0.0, + Fascia.F1_MP: 0.0, + Fascia.F2_MP: 0.0, + Fascia.F3_MP: 0.0, + Fascia.F23_MP: 0.0, + } class Zona(Enum): """Enumerazione con i nomi delle zone per i prezzi zonali.""" diff --git a/custom_components/pun_sensor/sensor.py b/custom_components/pun_sensor/sensor.py index f434814..58eea45 100644 --- a/custom_components/pun_sensor/sensor.py +++ b/custom_components/pun_sensor/sensor.py @@ -31,7 +31,7 @@ EVENT_UPDATE_PREZZO_ZONALE, EVENT_UPDATE_PUN, ) -from .interfaces import Fascia, PunValues +from .interfaces import Fascia, PunValues, PunValuesMP from .utils import datetime_to_packed_string, get_next_date ATTR_PREFIX_PREZZO_OGGI = "oggi_h_" @@ -57,6 +57,9 @@ async def async_setup_entry( entities.extend( PUNSensorEntity(coordinator, fascia) for fascia in PunValues().value ) + entities.extend( + PUNSensorEntity(coordinator, fascia) for fascia in PunValuesMP().value + ) # Crea sensori aggiuntivi entities.append(FasciaPUNSensorEntity(coordinator)) @@ -92,6 +95,16 @@ def __init__(self, coordinator: PUNDataUpdateCoordinator, fascia: Fascia) -> Non self.entity_id = ENTITY_ID_FORMAT.format("pun_fascia_f3") case Fascia.F23: self.entity_id = ENTITY_ID_FORMAT.format("pun_fascia_f23") + case Fascia.MONO_MP: + self.entity_id = ENTITY_ID_FORMAT.format("pun_mono_orario_mp") + case Fascia.F1_MP: + self.entity_id = ENTITY_ID_FORMAT.format("pun_fascia_f1_mp") + case Fascia.F2_MP: + self.entity_id = ENTITY_ID_FORMAT.format("pun_fascia_f2_mp") + case Fascia.F3_MP: + self.entity_id = ENTITY_ID_FORMAT.format("pun_fascia_f3_mp") + case Fascia.F23_MP: + self.entity_id = ENTITY_ID_FORMAT.format("pun_fascia_f23_mp") case _: self.entity_id = None self._attr_unique_id = self.entity_id @@ -116,25 +129,42 @@ def _handle_coordinator_update(self) -> None: if coordinator_event != EVENT_UPDATE_PUN: return - if self.fascia != Fascia.F23: + if self.fascia != Fascia.F23 and self.fascia != Fascia.F23_MP: # Tutte le fasce tranne F23 - if len(self.coordinator.pun_data.pun[self.fascia]) > 0: - # Ci sono dati, sensore disponibile - self._available = True - self._native_value = self.coordinator.pun_values.value[self.fascia] - else: - # Non ci sono dati, sensore non disponibile - self._available = False - + if "_MP" not in self.fascia.value: + if len(self.coordinator.pun_data.pun[self.fascia]) > 0: + # Ci sono dati, sensore disponibile + self._available = True + self._native_value = self.coordinator.pun_values.value[self.fascia] + else: + # Non ci sono dati, sensore non disponibile + self._available = False + elif "_MP" in self.fascia.value: + if len(self.coordinator.pun_data_mp.pun[self.fascia]) > 0: + # Ci sono dati, sensore disponibile + self._available = True + self._native_value = self.coordinator.pun_values_mp.value[self.fascia] + else: + # Non ci sono dati, sensore non disponibile + self._available = False elif ( len(self.coordinator.pun_data.pun[Fascia.F2]) and len(self.coordinator.pun_data.pun[Fascia.F3]) - ) > 0: + ) > 0 and self.fascia == Fascia.F23: # Caso speciale per fascia F23: affinché sia disponibile devono # esserci dati sia sulla fascia F2 che sulla F3, # visto che è calcolata a partire da questi self._available = True self._native_value = self.coordinator.pun_values.value[self.fascia] + elif ( + len(self.coordinator.pun_data_mp.pun[Fascia.F2_MP]) + and len(self.coordinator.pun_data_mp.pun[Fascia.F3_MP]) + ) > 0 and self.fascia == Fascia.F23_MP: + # Caso speciale per fascia F23: affinché sia disponibile devono + # esserci dati sia sulla fascia F2 che sulla F3, + # visto che è calcolata a partire da questi + self._available = True + self._native_value = self.coordinator.pun_values_mp.value[self.fascia] else: # Non ci sono dati, sensore non disponibile self._available = False @@ -189,8 +219,12 @@ def name(self) -> str | None: """Restituisce il nome del sensore.""" if self.fascia == Fascia.MONO: return "PUN mono-orario" - if self.fascia: + if self.fascia == Fascia.MONO_MP: + return "PUN mono-orario mese precedente" + if self.fascia and "_MP" not in str(self.fascia.value): return f"PUN fascia {self.fascia.value}" + if self.fascia and "_MP" in str(self.fascia.value): + return f"PUN fascia {self.fascia.value.replace("_MP","")} mese precedente" return None diff --git a/custom_components/pun_sensor/utils.py b/custom_components/pun_sensor/utils.py index 6c9794a..edf8480 100644 --- a/custom_components/pun_sensor/utils.py +++ b/custom_components/pun_sensor/utils.py @@ -7,7 +7,7 @@ import defusedxml.ElementTree as et # type: ignore[import-untyped] import holidays -from .interfaces import Fascia, PunData +from .interfaces import Fascia, PunData, PunDataMP # Ottiene il logger _LOGGER = logging.getLogger(__name__) @@ -36,6 +36,28 @@ def get_fascia_for_xml(data: date, festivo: bool, ora: int) -> Fascia: return Fascia.F1 return Fascia.F3 +def get_fascia_for_xml2(data: date, festivo: bool, ora: int) -> Fascia: + """Restituisce la fascia oraria di un determinato giorno/ora.""" + # F1 = lu-ve 8-19 + # F2 = lu-ve 7-8, lu-ve 19-23, sa 7-23 + # F3 = lu-sa 0-7, lu-sa 23-24, do, festivi + + # Festivi e domeniche + if festivo or (data.weekday() == 6): + return Fascia.F3_MP + + # Sabato + if data.weekday() == 5: + if 7 <= ora < 23: + return Fascia.F2_MP + return Fascia.F3_MP + + # Altri giorni della settimana + if ora == 7 or 19 <= ora < 23: + return Fascia.F2_MP + if 8 <= ora < 19: + return Fascia.F1_MP + return Fascia.F3_MP def get_fascia(dataora: datetime) -> tuple[Fascia, datetime]: """Restituisce la fascia della data/ora indicata e la data del prossimo cambiamento.""" @@ -289,3 +311,117 @@ def extract_xml(archive: ZipFile, pun_data: PunData, today: date) -> PunData: pun_data.prezzi_zonali[orario_prezzo] = None return pun_data + +def extract_xml2(archive: ZipFile, pun_data: PunDataMP, today: date) -> PunDataMP: + """Estrae i valori del pun per ogni fascia da un archivio zip contenente un XML. + + Args: + archive (ZipFile): archivio ZIP con i file XML all'interno. + pun_data (PunData): riferimento alla struttura che verrà modificata con i dati da XML. + today (date): data di oggi, utilizzata per memorizzare il prezzo zonale. + + Returns: + List[ list[MONO: float], list[F1: float], list[F2: float], list[F3: float] ] + + """ + # Carica le festività + it_holidays = holidays.IT() # type: ignore[attr-defined] + + # Azzera i dati precedenti + for fascia_da_svuotare in pun_data.pun.values(): + fascia_da_svuotare.clear() + + # Esamina ogni file XML nello ZIP (ordinandoli prima) + for fn in sorted(archive.namelist()): + # Scompatta il file XML in memoria + xml_tree = et.parse(archive.open(fn)) + + # Parsing dell'XML (1 file = 1 giorno) + xml_root = xml_tree.getroot() + + # Estrae la data dal primo elemento (sarà identica per gli altri) + dat_string = xml_root.find("Prezzi").find("Data").text # YYYYMMDD + + # Converte la stringa giorno in data + dat_date = date( + int(dat_string[0:4]), + int(dat_string[4:6]), + int(dat_string[6:8]), + ) + + # Verifica la festività + festivo = dat_date in it_holidays + + # Estrae le rimanenti informazioni + for prezzi in xml_root.iter("Prezzi"): + # Estrae l'ora dall'XML + ora = int(prezzi.find("Ora").text) - 1 # 1..24 + + # Estrae il prezzo PUN dall'XML in un float + if (prezzo_xml := prezzi.find("PUN")) is not None: + prezzo_string = prezzo_xml.text.replace(".", "").replace(",", ".") + prezzo = float(prezzo_string) / 1000 + + # Per le medie mensili, considera solo i dati fino ad oggi + if dat_date <= today: + # Estrae la fascia oraria + fascia = get_fascia_for_xml2(dat_date, festivo, ora) + + # Calcola le statistiche + pun_data.pun[Fascia.MONO_MP].append(prezzo) + pun_data.pun[fascia].append(prezzo) + + # Per il PUN orario, considera solo oggi e domani + if dat_date >= today: + # Compone l'orario + orario_prezzo = datetime_to_packed_string( + datetime( + year=dat_date.year, + month=dat_date.month, + day=dat_date.day, + hour=ora, + minute=0, + second=0, + microsecond=0, + ) + ) + # E salva il prezzo per quell'orario + pun_data.pun_orari[orario_prezzo] = prezzo + else: + # PUN non valido + _LOGGER.warning( + "PUN non specificato per %s ad orario: %s.", dat_string, ora + ) + + # Per i prezzi zonali, considera solo oggi e domani + if dat_date >= today: + # Compone l'orario + orario_prezzo = datetime_to_packed_string( + datetime( + year=dat_date.year, + month=dat_date.month, + day=dat_date.day, + hour=ora, + minute=0, + second=0, + microsecond=0, + ) + ) + + # Controlla che la zona del prezzo zonale sia impostata + if pun_data.zona is not None: + # Estrae il prezzo zonale dall'XML in un float + # basandosi sul nome dell'enum + if ( + prezzo_zonale_xml := prezzi.find(pun_data.zona.name) + ) is not None: + prezzo_zonale_string = prezzo_zonale_xml.text.replace( + ".", "" + ).replace(",", ".") + pun_data.prezzi_zonali[orario_prezzo] = ( + float(prezzo_zonale_string) / 1000 + ) + else: + pun_data.prezzi_zonali[orario_prezzo] = None + + return pun_data