diff --git a/.cspell/custom-dictionary-workspace.txt b/.cspell/custom-dictionary-workspace.txt index 1d7df52c1..eb06f3abc 100644 --- a/.cspell/custom-dictionary-workspace.txt +++ b/.cspell/custom-dictionary-workspace.txt @@ -41,6 +41,9 @@ darkred darray datap datapoints +dayname +daynumber +daysymbol dend devcontainer devcontainers @@ -62,6 +65,7 @@ feedin fentry Fingerbot firstparty +fline FLOMASTA foxess futurerate diff --git a/apps/predbat/config/apps.yaml b/apps/predbat/config/apps.yaml index 483c71a2a..9e2af65ed 100644 --- a/apps/predbat/config/apps.yaml +++ b/apps/predbat/config/apps.yaml @@ -321,6 +321,9 @@ pred_bat: # Note: You must enable this event sensor in the Octopus Integration in Home Assistant for it to work octopus_free_session: 're:(event.octopus_energy_([0-9a-z_]+|)_octoplus_free_electricity_session_events)' + # Alternative scraper from Octopus web site if the above is not working + # octopus_free_url: 'http://octopus.energy/free-electricity' + # Energy rates # Please set one of these three, if multiple are set then Octopus is used first, second rates_import/rates_export and latest basic metric diff --git a/apps/predbat/predbat.py b/apps/predbat/predbat.py index f6d3df50d..201b221f6 100644 --- a/apps/predbat/predbat.py +++ b/apps/predbat/predbat.py @@ -484,6 +484,113 @@ def download_predbat_releases(self): return self.releases + def octopus_free_line(self, res, free_sessions): + """ + Parse a line from the octopus free data + """ + if res: + dayname = res.group(1) + daynumber = res.group(2) + daysymbol = res.group(3) + month = res.group(4) + time_from = res.group(5) + time_to = res.group(6) + if "pm" in time_to: + is_pm = True + else: + is_pm = False + if "pm" in time_from: + is_fpm = True + elif "am" in time_from: + is_fpm = False + else: + is_fpm = is_pm + time_from = time_from.replace("am", "") + time_from = time_from.replace("pm", "") + time_to = time_to.replace("am", "") + time_to = time_to.replace("pm", "") + try: + time_from = int(time_from) + time_to = int(time_to) + except (ValueError, TypeError): + return + if is_fpm: + time_from += 12 + if is_pm: + time_to += 12 + # Convert into timestamp object + now = datetime.now() + year = now.year + time_from = str(time_from) + time_to = str(time_to) + daynumber = str(daynumber) + if len(time_from) == 1: + time_from = "0" + time_from + if len(time_to) == 1: + time_to = "0" + time_to + if len(daynumber) == 1: + daynumber = "0" + daynumber + + try: + timestamp_start = datetime.strptime("{} {} {} {} {} Z".format(year, month, daynumber, str(time_from), "00"), "%Y %B %d %H %M %z") + timestamp_end = datetime.strptime("{} {} {} {} {} Z".format(year, month, daynumber, str(time_to), "00"), "%Y %B %d %H %M %z") + # Change to local timezone, but these times were in local zone so push the hour back to the correct one + timestamp_start = timestamp_start.astimezone(self.local_tz) + timestamp_end = timestamp_end.astimezone(self.local_tz) + timestamp_start = timestamp_start.replace(hour=int(time_from)) + timestamp_end = timestamp_end.replace(hour=int(time_to)) + free_sessions.append({"start": timestamp_start.strftime(TIME_FORMAT), "end": timestamp_end.strftime(TIME_FORMAT), "rate": 0.0}) + except (ValueError, TypeError) as e: + pass + + def download_octopus_free_func(self, url): + """ + Download octopus free session data directly from a URL + """ + # Check the cache first + now = datetime.now() + if url in self.octopus_url_cache: + stamp = self.octopus_url_cache[url]["stamp"] + pdata = self.octopus_url_cache[url]["data"] + age = now - stamp + if age.seconds < (30 * 60): + self.log("Return cached octopus data for {} age {} minutes".format(url, self.dp1(age.seconds / 60))) + return pdata + + r = requests.get(url) + if r.status_code not in [200, 201]: + self.log("Warn: Error downloading Octopus data from URL {}, code {}".format(url, r.status_code)) + self.record_status("Warn: Error downloading Octopus free session data", debug=url, had_errors=True) + return None + + # Return new data + self.octopus_url_cache[url] = {} + self.octopus_url_cache[url]["stamp"] = now + self.octopus_url_cache[url]["data"] = r.text + return r.text + + def download_octopus_free(self, url): + """ + Download octopus free session data directly from a URL and process the data + """ + + free_sessions = [] + pdata = self.download_octopus_free_func(url) + if not pdata: + return free_sessions + + for line in pdata.split("\n"): + if "Past sessions" in line: + future_line = line.split("

\s*(\S+)\s+(\d+)(\S+)\s+(\S+)\s+(\S+)-(\S+)\s*", fline) + self.octopus_free_line(res, free_sessions) + if "Free Electricity:" in line: + # Free Electricity: Sunday 24th November 7-9am + res = re.search(r"Free Electricity:\s+(\S+)\s+(\d+)(\S+)\s+(\S+)\s+(\S+)-(\S+)", line) + self.octopus_free_line(res, free_sessions) + return free_sessions + def download_octopus_rates(self, url): """ Download octopus rates directly from a URL or return from cache if recent @@ -9230,6 +9337,9 @@ def fetch_sensor_data(self): octopus_free_slot["end"] = end octopus_free_slot["rate"] = 0 octopus_free_slots.append(octopus_free_slot) + # Direct Octopus URL + if "octopus_free_url" in self.args: + octopus_free_slots.extend(self.download_octopus_free(self.get_arg("octopus_free_url", indirect=False))) # Octopus saving session octopus_saving_slots = [] @@ -10005,11 +10115,11 @@ def update_time(self, print=True): """ Update the current time/date """ - local_tz = pytz.timezone(self.args.get("timezone", "Europe/London")) + self.local_tz = pytz.timezone(self.args.get("timezone", "Europe/London")) skew = self.args.get("clock_skew", 0) if skew: self.log("Warn: Clock skew is set to {} minutes".format(skew)) - self.now_utc_real = datetime.now(local_tz) + self.now_utc_real = datetime.now(self.local_tz) now_utc = self.now_utc_real + timedelta(minutes=skew) now = datetime.now() + timedelta(minutes=skew) now = now.replace(second=0, microsecond=0, minute=(now.minute - (now.minute % PREDICT_STEP))) diff --git a/apps/predbat/unit_test.py b/apps/predbat/unit_test.py index c133877a7..bbc359bb4 100644 --- a/apps/predbat/unit_test.py +++ b/apps/predbat/unit_test.py @@ -3370,6 +3370,13 @@ def main(): print("**** Testing Predbat ****") failed = False + + free_sessions = my_predbat.download_octopus_free("http://octopus.energy/free-electricity") + free_sessions = my_predbat.download_octopus_free("http://octopus.energy/free-electricity") + if not free_sessions: + print("**** ERROR: No free sessions found ****") + failed = 1 + if not failed: failed |= run_intersect_window_tests(my_predbat) if not failed: diff --git a/docs/energy-rates.md b/docs/energy-rates.md index 5a632f97b..a307ab793 100644 --- a/docs/energy-rates.md +++ b/docs/energy-rates.md @@ -143,6 +143,13 @@ increase in this period. E.g. setting to a value of 1.2 would indicate you will If you do not want Predbat to see these sessions then comment out the **octopus_free_session** setting. +Note: If the above is not working due to lack of data (via a 3rd party service) Predbat can scrape directly from the Octopus Web Site, this may +have its own issues due to change of format. If you enable this then sessions will be considered even if you forget to sign-up so be careful! + +```yaml +octopus_free_url: 'http://octopus.energy/free-electricity' +``` + ## Octopus Rates URL If you do not wish to use the Octopus Energy integration and are an Octopus Energy customer then you can configure Predbat to get the electricity rates