Octopus free scraping from web site (#1651)

* Octopus free scraping from web site

* Add files via upload

* [pre-commit.ci lite] apply automatic fixes

* Update apps.yaml

* Update energy-rates.md

* Update custom-dictionary-workspace.txt

* [pre-commit.ci lite] apply automatic fixes

---------

Co-authored-by: pre-commit-ci-lite[bot] <117423508+pre-commit-ci-lite[bot]@users.noreply.github.com>

.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

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
 

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("<p data-block-key")
+                for fline in future_line:
+                    res = re.search(r"<i>\s*(\S+)\s+(\d+)(\S+)\s+(\S+)\s+(\S+)-(\S+)\s*</i>", 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)))

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:

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