From 9d50d38cabe8c697d9df8a289e9615a2ef214a0b Mon Sep 17 00:00:00 2001 From: flopp999 <21694965+flopp999@users.noreply.github.com> Date: Thu, 4 Jan 2024 16:05:34 +0100 Subject: [PATCH] fix for yearly revenue --- pycheckwatt/__init__.py | 371 ++++++++++++---------------------------- 1 file changed, 109 insertions(+), 262 deletions(-) diff --git a/pycheckwatt/__init__.py b/pycheckwatt/__init__.py index a2742ce..cd823c2 100644 --- a/pycheckwatt/__init__.py +++ b/pycheckwatt/__init__.py @@ -1,23 +1,21 @@ -"""CheckWatt module.""" +"""Checkwatt module.""" from __future__ import annotations import base64 +from datetime import datetime, timedelta import json import logging import re -from datetime import datetime, timedelta - -from aiohttp import ClientError, ClientResponseError, ClientSession from dateutil.relativedelta import relativedelta +from aiohttp import ClientError, ClientResponseError, ClientSession _LOGGER = logging.getLogger(__name__) - class CheckwattManager: - """CheckWatt manager.""" + """Checkwatt manager.""" def __init__(self, username, password) -> None: - """Initialize the CheckWatt manager.""" + """Initialize the checkwatt manager.""" if username is None or password is None: raise ValueError("Username and password must be provided.") self.session = None @@ -47,7 +45,7 @@ def __init__(self, username, password) -> None: self.power_data = None self.price_zone = None self.spot_prices = None - self.energy_data = None + self.firstcolor = None async def __aenter__(self): """Asynchronous enter.""" @@ -76,12 +74,10 @@ def _get_headers(self): } def _extract_content_and_logbook(self, input_string): - """Pull the registered information from the logbook.""" + """Pull the registred information from the logbook.""" # Define the pattern to match the content between the tags - pattern = re.compile( - r"#BEGIN_BATTERY_REGISTRATION(.*?)#END_BATTERY_REGISTRATION", re.DOTALL - ) + pattern = re.compile(r"#BEGIN_BATTERY_REGISTRATION(.*?)#END_BATTERY_REGISTRATION", re.DOTALL) # Find all matches in the input string matches = re.findall(pattern, input_string) @@ -99,31 +95,20 @@ def _extract_content_and_logbook(self, input_string): logbook_entries = [ entry.strip() for entry in logbook_entries - if not ( - "#BEGIN_BATTERY_REGISTRATION" in entry - or "#END_BATTERY_REGISTRATION" in entry - ) + if not ("#BEGIN_BATTERY_REGISTRATION" in entry or "#END_BATTERY_REGISTRATION" in entry) ] return battery_registration, logbook_entries def _extract_fcr_d_state(self): - pattern = re.compile( - r"\[ FCR-D (ACTIVATED|DEACTIVATE) \].*?(\d+,\d+/\d+,\d+/\d+,\d+ %).*?(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})" - ) + pattern = re.compile(r"\[ FCR-D (ACTIVATED|DEACTIVATE) \].*?(\d+,\d+/\d+,\d+/\d+,\d+ %).*?(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})") for entry in self.logbook_entries: match = pattern.search(entry) if match: - self.fcrd_state = match.group( - 1 - ) # FCR-D state: ACTIVATED or DEACTIVATED - self.fcrd_percentage = match.group( - 2 - ) # Percentage, e.g., "99,0/2,9/97,7 %" - self.fcrd_timestamp = ( - match.group(3) if match else None - ) # Timestamp, e.g., "2023-12-20 00:11:45" - break # stop so we get the first row in logbook + self.fcrd_state = match.group(1) # FCR-D state: ACTIVATED or DEACTIVATED + self.fcrd_percentage = match.group(2) # Percentage, e.g., "99,0/2,9/97,7 %" + self.fcrd_timestamp = match.group(3) if match else None # Timestamp, e.g., "2023-12-20 00:11:45" + break # stop so we get the first row in logbook, which is the latest information async def handle_client_error(self, endpoint, headers, error): """Handle ClientError and log relevant information.""" @@ -136,12 +121,10 @@ async def handle_client_error(self, endpoint, headers, error): return False async def login(self): - """Login to CheckWatt.""" + """Login to Checkwatt.""" try: credentials = f"{self.username}:{self.password}" - encoded_credentials = base64.b64encode(credentials.encode("utf-8")).decode( - "utf-8" - ) + encoded_credentials = base64.b64encode(credentials.encode("utf-8")).decode("utf-8") endpoint = "/user/LoginEiB?audience=eib" # Define headers with the encoded credentials @@ -160,9 +143,7 @@ async def login(self): return True if response.status == 401: - _LOGGER.error( - "Unauthorized: Check your checkwatt authentication credentials" - ) + _LOGGER.error("Unauthorized: Check your checkwatt authentication credentials") return False _LOGGER.error("Unexpected HTTP status code: %s", response.status) @@ -172,7 +153,7 @@ async def login(self): return await self.handle_client_error(endpoint, headers, error) async def get_customer_details(self): - """Fetch customer details from CheckWatt.""" + """Fetch customer details from Checkwatt.""" try: endpoint = "/controlpanel/CustomerDetail" @@ -191,31 +172,9 @@ async def get_customer_details(self): meters = self.customer_details.get("Meter", []) if meters: - soc_meter = next( - ( - meter - for meter in meters - if meter.get("InstallationType") == "SoC" - ), - None, - ) - charging_meter = next( - ( - meter - for meter in meters - if meter.get("InstallationType") == "Charging" - ), - None, - ) - discharging_meter = next( - ( - meter - for meter in meters - if meter.get("InstallationType") == "Discharging" - ), - None, - ) - + soc_meter = next((meter for meter in meters if meter.get("InstallationType") == "SoC"), None,) + charging_meter = next((meter for meter in meters if meter.get("InstallationType") == "Charging"), None,) + discharging_meter = next((meter for meter in meters if meter.get("InstallationType") == "Discharging"), None,) if not soc_meter: _LOGGER.error("No SoC meter found") return False @@ -223,10 +182,7 @@ async def get_customer_details(self): battery_charge_peak = charging_meter.get("PeakAcKw") battery_discharge_peak = discharging_meter.get("PeakAcKw") if logbook: - ( - self.battery_registration, - self.logbook_entries, - ) = self._extract_content_and_logbook(logbook) + (self.battery_registration, self.logbook_entries) = self._extract_content_and_logbook(logbook) self.battery_charge_peak = battery_charge_peak self.battery_discharge_peak = battery_discharge_peak self._extract_fcr_d_state() @@ -244,23 +200,21 @@ async def get_customer_details(self): return await self.handle_client_error(endpoint, headers, error) async def get_fcrd_revenuemonth(self): - """Fetch FCR-D revenues from checkwatt.""" misseddays = 0 + """Fetch FCR-D revenues from checkwatt.""" try: from_date = datetime.now().strftime("%Y-%m-01") to_date = datetime.now() + timedelta(days=2) to_date = to_date.strftime("%Y-%m-%d") lastday_date = datetime.now() + relativedelta(months=1) - lastday_date = datetime( - year=lastday_date.year, month=lastday_date.month, day=1 - ) + lastday_date = datetime(year=lastday_date.year,month=lastday_date.month,day=1) lastday_date = lastday_date - timedelta(days=1) lastday = lastday_date.strftime("%d") dayssofar = datetime.now() + timedelta(days=1) dayssofar = dayssofar.strftime("%d") - daysleft = int(lastday) - int(dayssofar) + daysleft = int(lastday)-int(dayssofar) endpoint = f"/ems/fcrd/revenue?fromDate={from_date}&toDate={to_date}" @@ -280,12 +234,10 @@ async def get_fcrd_revenuemonth(self): self.revenuemonth += each["Revenue"] if each["Revenue"] == 0: misseddays += 1 - dayswithmoney = int(dayssofar) - misseddays + dayswithmoney = int(dayssofar)-misseddays if response.status == 200: # Then fetch the service fees - endpoint = ( - f"/ems/service/fees?fromDate={from_date}&toDate={to_date}" - ) + endpoint = (f"/ems/service/fees?fromDate={from_date}&toDate={to_date}") async with self.session.get( self.base_url + endpoint, headers=headers ) as response: @@ -293,12 +245,8 @@ async def get_fcrd_revenuemonth(self): fees = await response.json() for each in fees["FCRD"]: self.feesmonth += each["Revenue"] - self.dailyaverage = ( - int(self.revenuemonth) - int(self.feesmonth) - ) / int(dayswithmoney) - self.monthestimate = (self.dailyaverage * daysleft) + ( - int(self.revenuemonth) - int(self.feesmonth) - ) + self.dailyaverage = ( int(self.revenuemonth) - int(self.feesmonth) ) / int(dayswithmoney) + self.monthestimate = ( self.dailyaverage * daysleft ) + ( int(self.revenuemonth) - int(self.feesmonth) ) if response.status == 200: return True @@ -312,6 +260,7 @@ async def get_fcrd_revenuemonth(self): except (ClientResponseError, ClientError) as error: return await self.handle_client_error(endpoint, headers, error) + async def get_fcrd_revenue(self): """Fetch FCR-D revenues from checkwatt.""" try: @@ -335,9 +284,7 @@ async def get_fcrd_revenue(self): self.revenue = await response.json() if response.status == 200: # Then fetch the service fees - endpoint = ( - f"/ems/service/fees?fromDate={from_date}&toDate={to_date}" - ) + endpoint = (f"/ems/service/fees?fromDate={from_date}&toDate={to_date}") async with self.session.get( self.base_url + endpoint, headers=headers ) as response: @@ -356,75 +303,96 @@ async def get_fcrd_revenue(self): except (ClientResponseError, ClientError) as error: return await self.handle_client_error(endpoint, headers, error) + + + async def get_fcrd_revenueyear(self): - """Fetch FCR-D revenues from CheckWatt.""" yesterday_date = datetime.now() yesterday_date = yesterday_date.strftime("-%m-%d") - months = ["-01-01", "-06-30", "-07-01", yesterday_date] + months = ["-01-01","-06-30","-07-01",yesterday_date] loop = 0 - retval = False - try: - while loop < 3: + """Fetch FCR-D revenues from CheckWatt.""" + if (yesterday_date <= "07-01"): + try: year_date = datetime.now().strftime("%Y") - to_date = year_date + months[loop + 1] - from_date = year_date + months[loop] - + to_date = year_date + yesterday_date + from_date = year_date + "-01-01" endpoint = f"/ems/fcrd/revenue?fromDate={from_date}&toDate={to_date}" - # Define headers with the JwtToken - headers = { - **self._get_headers(), - "authorization": f"Bearer {self.jwt_token}", - } - + headers = {**self._get_headers(),"authorization": f"Bearer {self.jwt_token}"} # First fetch the revenue - async with self.session.get( - self.base_url + endpoint, headers=headers - ) as responseyear: + async with self.session.get(self.base_url + endpoint, headers=headers) as responseyear: responseyear.raise_for_status() self.revenueyear = await responseyear.json() for each in self.revenueyear: self.revenueyeartotal += each["Revenue"] if responseyear.status == 200: - # Then fetch the service fees - endpoint = ( - f"/ems/service/fees?fromDate={from_date}&toDate={to_date}" - ) - async with self.session.get( - self.base_url + endpoint, headers=headers - ) as responseyear: + # Then fetch the service fees + endpoint = (f"/ems/service/fees?fromDate={from_date}&toDate={to_date}") + async with self.session.get(self.base_url + endpoint, headers=headers) as responseyear: responseyear.raise_for_status() self.feesyear = await responseyear.json() for each in self.feesyear["FCRD"]: self.feesyeartotal += each["Revenue"] if responseyear.status == 200: loop += 2 - retval = True else: - _LOGGER.error( - "Obtaining data from URL %s failed with status code %d", - self.base_url + endpoint, - responseyear.status, - ) - break + return False else: - _LOGGER.error( - "Obtaining data from URL %s failed with status code %d", - self.base_url + endpoint, - responseyear.status, - ) - break - return retval + False + return True - except (ClientResponseError, ClientError) as error: - return await self.handle_client_error(endpoint, headers, error) + _LOGGER.error("Obtaining data from URL %s failed with status code %d",self.base_url + endpoint,responseyear.status) + return False + + except (ClientResponseError, ClientError) as error: + return await self.handle_client_error(endpoint, headers, error) + + else: + try: + while loop < 2: + year_date = datetime.now().strftime("%Y") + to_date = year_date + months[loop+1] + from_date = year_date + months[loop] + print(from_date,to_date) + + endpoint = f"/ems/fcrd/revenue?fromDate={from_date}&toDate={to_date}" + # Define headers with the JwtToken + headers = {**self._get_headers(),"authorization": f"Bearer {self.jwt_token}"} + # First fetch the revenue + async with self.session.get(self.base_url + endpoint, headers=headers) as responseyear: + responseyear.raise_for_status() + self.revenueyear = await responseyear.json() + for each in self.revenueyear: + self.revenueyeartotal += each["Revenue"] + print(self.revenueyeartotal) + if responseyear.status == 200: + # Then fetch the service fees + endpoint = (f"/ems/service/fees?fromDate={from_date}&toDate={to_date}") + async with self.session.get(self.base_url + endpoint, headers=headers) as responseyear: + responseyear.raise_for_status() + self.feesyear = await responseyear.json() + for each in self.feesyear["FCRD"]: + self.feesyeartotal += each["Revenue"] + print(self.feesyeartotal) + if responseyear.status == 200: + loop += 2 + else: + break + else: + break + return True + + _LOGGER.error("Obtaining data from URL %s failed with status code %d",self.base_url + endpoint,responseyear.status) + return False + + except (ClientResponseError, ClientError) as error: + return await self.handle_client_error(endpoint, headers, error) def _build_series_endpoint(self, grouping): end_date = datetime.now() + timedelta(days=2) to_date = end_date.strftime("%Y") - endpoint = ( - f"/datagrouping/series?grouping={grouping}&fromdate=1923&todate={to_date}" - ) + endpoint = (f"/datagrouping/series?grouping={grouping}&fromdate=1923&todate={to_date}") meters = self.customer_details.get("Meter", []) if meters: @@ -439,9 +407,7 @@ async def get_power_data(self): """Fetch Power Data from checkwatt.""" try: - endpoint = self._build_series_endpoint( - 3 - ) # 0: Hourly, 1: Daily, 2: Monthly, 3: Yearly + endpoint = self._build_series_endpoint(3) # 0: Hourly, 1: Daily, 2: Monthly, 3: Yearly # Define headers with the JwtToken headers = { @@ -468,38 +434,6 @@ async def get_power_data(self): except (ClientResponseError, ClientError) as error: return await self.handle_client_error(endpoint, headers, error) - async def get_energy_flow(self): - """Fetch Power Data from CheckWatt.""" - - try: - endpoint = "/ems/energyflow" - - # Define headers with the JwtToken - headers = { - **self._get_headers(), - "authorization": f"Bearer {self.jwt_token}", - } - - # Fetch Energy Flows - async with self.session.get( - self.base_url + endpoint, headers=headers - ) as response: - response.raise_for_status() - if response.status == 200: - self.energy_data = await response.json() - return True - - _LOGGER.error( - "Obtaining data from URL %s failed with status code %d", - self.base_url + endpoint, - response.status, - ) - return False - - except (ClientResponseError, ClientError) as error: - return await self.handle_client_error(endpoint, headers, error) - - async def get_price_zone(self): """Fetch Price Zone from checkwatt.""" @@ -565,49 +499,10 @@ async def get_spot_price(self): except (ClientResponseError, ClientError) as error: return await self.handle_client_error(endpoint, headers, error) - - async def get_energy_trading_company(self, input_id): - """Translate Energy Company Id to Energy Company Name.""" - try: - endpoint = "/controlpanel/elhandelsbolag" - - # Define headers with the JwtToken - headers = { - **self._get_headers(), - } - - async with self.session.get( - self.base_url + endpoint, headers=headers - ) as response: - response.raise_for_status() - if response.status == 200: - energy_trading_companies = await response.json() - for energy_trading_company in energy_trading_companies: - if energy_trading_company['Id'] == input_id: - return energy_trading_company['DisplayName'] - - - return None - - _LOGGER.error( - "Obtaining data from URL %s failed with status code %d", - self.base_url + endpoint, - response.status, - ) - return None - - except (ClientResponseError, ClientError) as error: - return await self.handle_client_error(endpoint, headers, error) - - - @property def inverter_make_and_model(self): """Property for inverter make and model. Not used by HA integration..""" - if ( - "Inverter" in self.battery_registration - and "InverterModel" in self.battery_registration - ): + if ("Inverter" in self.battery_registration and "InverterModel" in self.battery_registration): resp = f"{self.battery_registration['Inverter']}" resp += f" {self.battery_registration['InverterModel']}" return resp @@ -615,33 +510,27 @@ def inverter_make_and_model(self): @property def battery_make_and_model(self): """Property for battery make and model. Not used by HA integration.""" - if ( - "BatteryModel" in self.battery_registration - and "BatterySystem" in self.battery_registration - ): + if ("BatteryModel" in self.battery_registration and "BatterySystem" in self.battery_registration ): resp = f"{self.battery_registration['BatterySystem']}" resp += f" {self.battery_registration['BatteryModel']}" resp += f" ({self.battery_registration['BatteryPowerKW']}kW, {self.battery_registration['BatteryCapacityKWh']}kWh)" return resp else: - return "Could not get any information about your battery" + return("Could not get any information about your battery") @property def electricity_provider(self): """Property for electricity provides. Not used by HA integration.""" - if ( - "ElectricityCompany" in self.battery_registration - and "Dso" in self.battery_registration - ): + if ("ElectricityCompany" in self.battery_registration and "Dso" in self.battery_registration): resp = f"{self.battery_registration['ElectricityCompany']}" resp += f" via {self.battery_registration['Dso']}" - if "GridAreaId" in self.battery_registration: + if ("GridAreaId" in self.battery_registration): resp += f" ({self.battery_registration['GridAreaId']} {self.battery_registration['Kommun']})" return resp @property - def registered_owner(self): - """Property for registered owner. Not used by HA integration..""" + def registred_owner(self): + """Property for registred owner. Not used by HA integration..""" if "FirstName" in self.customer_details and "LastName" in self.customer_details: resp = f"{self.customer_details['FirstName']}" resp += f" {self.customer_details['LastName']}" @@ -662,7 +551,7 @@ def year_revenue(self): if self.feesyeartotal is not None: feesyear = self.feesyeartotal - return revenueyear, feesyear + return revenueyear,feesyear @property def month_revenue(self): @@ -683,7 +572,8 @@ def month_revenue(self): if self.monthestimate is not None: monthestimate = self.monthestimate - return revenuemonth, feesmonth, dailyaverage, monthestimate + + return revenuemonth,feesmonth, dailyaverage, monthestimate @property def today_revenue(self): @@ -702,7 +592,7 @@ def today_revenue(self): if "Revenue" in self.fees["FCRD"][0]: fees = self.fees["FCRD"][0]["Revenue"] - return revenue, fees + return revenue,fees @property def tomorrow_revenue(self): @@ -721,7 +611,7 @@ def tomorrow_revenue(self): if "Revenue" in self.fees["FCRD"][1]: fees = self.fees["FCRD"][1]["Revenue"] - return revenue, fees + return revenue,fees def _get_meter_total(self, meter_type): """Solar, Charging, Discharging, EDIEL_E17, EDIEL_E18, Soc meter summary.""" @@ -732,7 +622,7 @@ def _get_meter_total(self, meter_type): if meter["InstallationType"] == meter_type: for measurement in meter["Measurements"]: if "Value" in measurement: - meter_total += measurement["Value"] # to get answer to kWh + meter_total += measurement["Value"] # to get answer to kWh return meter_total @property @@ -770,46 +660,3 @@ def get_spot_price_excl_vat(self, now_hour: int): _LOGGER.warning("Unable to retrieve spot price for the current hour") return None - - - @property - def battery_power(self): - """Property for Battery Power.""" - if self.energy_data is not None: - if "BatteryNow" in self.energy_data: - return self.energy_data["BatteryNow"] - - _LOGGER.warning("Unable to retrieve Battery Power") - return None - - @property - def grid_power(self): - """Property for Grid Power.""" - if self.energy_data is not None: - if "GridNow" in self.energy_data: - return self.energy_data["GridNow"] - - _LOGGER.warning("Unable to retrieve Grid Power") - return None - - @property - def solar_power(self): - """Property for Solar Power.""" - if self.energy_data is not None: - if "SolarNow" in self.energy_data: - return self.energy_data["SolarNow"] - - _LOGGER.warning("Unable to retrieve Solar Power") - return None - - @property - def battery_soc(self): - """Property for Battery SoC.""" - if self.energy_data is not None: - if "BatterySoC" in self.energy_data: - return self.energy_data["BatterySoC"] - - _LOGGER.warning("Unable to retrieve Battery SoC") - return None - -