Skip to content

Decouple octopus direct from Predbat using database #2825

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
springfall2008 merged 7 commits into from
Oct 25, 2025
Merged

Decouple octopus direct from Predbat using database #2825

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
b9c3388
Re-work octopus direct to use HA interface
springfall2008 Oct 25, 2025
aa7ffb4
Typo
springfall2008 Oct 25, 2025
cc354af
Join events via selector
springfall2008 Oct 25, 2025
706cdaa
[pre-commit.ci lite] apply automatic fixes
pre-commit-ci-lite[bot] Oct 25, 2025
03c4f50
Typos
springfall2008 Oct 25, 2025
66cf697
Merge branch 'fixes8' of github.com:springfall2008/batpred into fixes8
springfall2008 Oct 25, 2025
c23db3d
Fix unit test
springfall2008 Oct 25, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions apps/predbat/components.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -107,6 +107,11 @@
"required": True,
"config": "octopus_api_account",
},
"automatic": {
"required": False,
"default": True,
"config": "octopus_automatic",
},
},
},
"ohme": {
Expand Down
104 changes: 28 additions & 76 deletions apps/predbat/fetch.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1061,18 +1061,10 @@ def fetch_sensor_data(self):
# Log current values
self.log("Current data so far today: load {} kWh import {} kWh export {} kWh pv {} kWh".format(dp2(self.load_minutes_now), dp2(self.import_today_now), dp2(self.export_today_now), dp2(self.pv_today_now)))

octopus_api_direct = self.components.get_component("octopus")
if "rates_import_octopus_url" in self.args:
# Fixed URL for rate import
self.log("Downloading import rates directly from URL {}".format(self.get_arg("rates_import_octopus_url", indirect=False)))
self.rate_import = self.download_octopus_rates(self.get_arg("rates_import_octopus_url", indirect=False))
elif octopus_api_direct:
self.log("Downloading import rates directly from Octopus API")
self.rate_import = self.get_octopus_direct("import")
if not self.rate_import:
self.log("Error: Unable to download rates from Octopus API")
self.record_status(message="Error: Unable to download rates from Octopus API", had_errors=True)
raise ValueError
elif "metric_octopus_import" in self.args:
# Octopus import rates
entity_id = self.get_arg("metric_octopus_import", None, indirect=False)
Expand All @@ -1094,12 +1086,7 @@ def fetch_sensor_data(self):
self.rate_import = self.basic_rates(self.get_arg("rates_import", [], indirect=False), "rates_import")

# Gas rates if set
if octopus_api_direct:
self.log("Downloading gas rates directly from Octopus API")
self.rate_gas = self.get_octopus_direct("gas")
self.rate_gas, self.rate_gas_replicated = self.rate_replicate(self.rate_gas, is_import=False, is_gas=False)
self.rate_scan_gas(self.rate_gas, print=True)
elif "metric_octopus_gas" in self.args:
if "metric_octopus_gas" in self.args:
entity_id = self.get_arg("metric_octopus_gas", None, indirect=False)
self.rate_gas = self.fetch_octopus_rates(entity_id)
if not self.rate_gas:
Expand Down Expand Up @@ -1138,40 +1125,28 @@ def fetch_sensor_data(self):
# Work out current car SoC and limit
self.car_charging_loss = 1 - float(self.get_arg("car_charging_loss"))

slot_is_configured = "octopus_intelligent_slot" in self.args
entity_id = self.get_arg("octopus_intelligent_slot", indirect=False)
ohme_automatic = self.get_arg("ohme_automatic", False)

if (octopus_api_direct and octopus_api_direct.get_intelligent_device()) or (not octopus_api_direct and slot_is_configured):
if entity_id:
completed = []
planned = []
vehicle = {}
vehicle_pref = {}

if octopus_api_direct and not ohme_automatic:
completed = octopus_api_direct.get_intelligent_completed_dispatches()
planned = octopus_api_direct.get_intelligent_planned_dispatches()
vehicle = octopus_api_direct.get_intelligent_vehicle()
vehicle_pref = vehicle
self.log("Octopus API planned and completed slots: {} - {}".format(completed, planned))
else:
entity_id = self.get_arg("octopus_intelligent_slot", indirect=False)
if entity_id and "octopus_intelligent_slot_action_config" in self.args:
config_entry = self.get_arg("octopus_intelligent_slot_action_config", None, indirect=False)
service_name = entity_id.replace(".", "/")
result = self.call_service_wrapper(service_name, config_entry=config_entry, return_response=True)
if result and ("slots" in result):
planned = result["slots"]
else:
self.log("Warn: Unable to get data from {} - octopus_intelligent_slot using action config {} result was {}".format(entity_id, config_entry, result))

if entity_id and "octopus_intelligent_slot_action_config" in self.args:
config_entry = self.get_arg("octopus_intelligent_slot_action_config", None, indirect=False)
service_name = entity_id.replace(".", "/")
result = self.call_service_wrapper(service_name, config_entry=config_entry, return_response=True)
if result and ("slots" in result):
planned = result["slots"]
else:
try:
completed = self.get_state_wrapper(entity_id=entity_id, attribute="completedDispatches") or self.get_state_wrapper(entity_id=entity_id, attribute="completed_dispatches")
planned = self.get_state_wrapper(entity_id=entity_id, attribute="plannedDispatches") or self.get_state_wrapper(entity_id=entity_id, attribute="planned_dispatches")
vehicle = self.get_state_wrapper(entity_id=entity_id, attribute="registeredKrakenflexDevice")
vehicle_pref = self.get_state_wrapper(entity_id=entity_id, attribute="vehicleChargingPreferences")
except (ValueError, TypeError):
self.log("Warn: Unable to get data from {} - octopus_intelligent_slot may not be set correctly".format(entity_id))
self.record_status(message="Error: octopus_intelligent_slot not set correctly", had_errors=True)
self.log("Warn: Unable to get data from {} - octopus_intelligent_slot using action config {} result was {}".format(entity_id, config_entry, result))
else:
try:
completed = self.get_state_wrapper(entity_id=entity_id, attribute="completed_dispatches") or self.get_state_wrapper(entity_id=entity_id, attribute="completedDispatches")
planned = self.get_state_wrapper(entity_id=entity_id, attribute="planned_dispatches") or self.get_state_wrapper(entity_id=entity_id, attribute="plannedDispatches")
except (ValueError, TypeError):
self.log("Warn: Unable to get data from {} - octopus_intelligent_slot may not be set correctly".format(entity_id))
self.record_status(message="Error: octopus_intelligent_slot not set correctly", had_errors=True)

# Completed and planned slots
if completed:
Expand All @@ -1186,33 +1161,18 @@ def fetch_sensor_data(self):

if self.num_cars >= 1:
# Extract vehicle data if we can get it
if vehicle or (octopus_api_direct and not ohme_automatic):
self.car_charging_battery_size[0] = float(vehicle.get("vehicleBatterySizeInKwh", self.car_charging_battery_size[0]))
self.car_charging_rate[0] = float(vehicle.get("chargePointPowerInKw", self.car_charging_rate[0]))
else:
size = self.get_state_wrapper(entity_id=entity_id, attribute="vehicle_battery_size_in_kwh")
rate = self.get_state_wrapper(entity_id=entity_id, attribute="charge_point_power_in_kw")
if size:
self.car_charging_battery_size[0] = size
if rate:
self.car_charging_rate[0] = rate
size = self.get_state_wrapper(entity_id=entity_id, attribute="vehicle_battery_size_in_kwh")
rate = self.get_state_wrapper(entity_id=entity_id, attribute="charge_point_power_in_kw")
if size:
self.car_charging_battery_size[0] = size
if rate:
self.car_charging_rate[0] = rate

# Get car charging limit again from car based on new battery size
self.car_charging_limit[0] = dp3((float(self.get_arg("car_charging_limit", 100.0, index=0)) * self.car_charging_battery_size[0]) / 100.0)

# Extract vehicle preference if we can get it
if (vehicle_pref or (octopus_api_direct and not ohme_automatic)) and self.octopus_intelligent_charging:
octopus_limit = max(float(vehicle_pref.get("weekdayTargetSoc", 100)), float(vehicle_pref.get("weekendTargetSoc", 100)))
octopus_ready_time = vehicle_pref.get("weekdayTargetTime", None)
if not octopus_ready_time:
octopus_ready_time = self.car_charging_plan_time[0]
else:
if isinstance(octopus_ready_time, str) and len(octopus_ready_time) == 5:
octopus_ready_time += ":00"
self.car_charging_plan_time[0] = octopus_ready_time
octopus_limit = dp3(octopus_limit * self.car_charging_battery_size[0] / 100.0)
self.car_charging_limit[0] = min(self.car_charging_limit[0], octopus_limit)
elif self.octopus_intelligent_charging:
if self.octopus_intelligent_charging:
octopus_ready_time = self.get_arg("octopus_ready_time", None)
if isinstance(octopus_ready_time, str) and len(octopus_ready_time) == 5:
octopus_ready_time += ":00"
Expand All @@ -1229,8 +1189,7 @@ def fetch_sensor_data(self):
if octopus_ready_time:
self.car_charging_plan_time[0] = octopus_ready_time

# Use octopus slots for charging?
if self.octopus_intelligent_charging:
# Use octopus slots for charging?
self.octopus_slots = self.add_now_to_octopus_slot(self.octopus_slots, self.now_utc)
if not self.octopus_intelligent_ignore_unplugged or self.car_charging_planned[0]:
self.car_charging_slots[0] = self.load_octopus_slots(self.octopus_slots, self.octopus_intelligent_consider_full)
Expand Down Expand Up @@ -1262,9 +1221,6 @@ def fetch_sensor_data(self):
# Fixed URL for rate export
self.log("Downloading export rates directly from URL {}".format(self.get_arg("rates_export_octopus_url", indirect=False)))
self.rate_export = self.download_octopus_rates(self.get_arg("rates_export_octopus_url", indirect=False))
elif octopus_api_direct:
self.log("Downloading export rates directly from Octopus API")
self.rate_export = self.get_octopus_direct("export")
elif "metric_octopus_export" in self.args:
# Octopus export rates
entity_id = self.get_arg("metric_octopus_export", None, indirect=False)
Expand All @@ -1287,12 +1243,8 @@ def fetch_sensor_data(self):
self.octopus_free_slots, self.octopus_saving_slots = self.fetch_octopus_sessions()

# Standing charge
if octopus_api_direct:
self.metric_standing_charge = self.get_standing_charge_direct()
self.log("Octopus Import standing charge is set to {} p".format(self.metric_standing_charge))
else:
self.metric_standing_charge = self.get_arg("metric_standing_charge", 0.0) * 100.0
self.log("Standing charge is set to {} p".format(self.metric_standing_charge))
self.metric_standing_charge = self.get_arg("metric_standing_charge", 0.0) * 100.0
self.log("Standing charge is set to {} p".format(self.metric_standing_charge))

# futurerate data
futurerate = FutureRate(self)
Expand Down
Loading