diff --git a/apps/predbat/config.py b/apps/predbat/config.py index 663c391f5..99e51c0ee 100644 --- a/apps/predbat/config.py +++ b/apps/predbat/config.py @@ -937,6 +937,50 @@ "enable": "num_cars", "enable_condition": "num_cars > 0", }, + { + "name": "car_charging_solar", + "friendly_name": "Charge car on excess solar", + "type": "switch", + "default": False, + "enable": "num_cars", + "enable_condition": "num_cars > 0", + }, + { + "name": "car_charging_solar_excess", + "friendly_name": "Car solar charging surplus threshold", + "type": "input_number", + "min": 0.1, + "max": 20.0, + "step": 0.1, + "unit": "kW", + "icon": "mdi:solar-power", + "default": 1.0, + "enable": "car_charging_solar", + }, + { + "name": "car_charging_plan_min_soc", + "friendly_name": "Car minimum charge from any source", + "type": "input_number", + "min": 0, + "max": 100, + "step": 1, + "unit": "%", + "icon": "mdi:battery-charging-low", + "default": 100, + "enable": "car_charging_solar", + }, + { + "name": "car_charging_rate_threshold_export", + "friendly_name": "Car solar charging export rate threshold", + "type": "input_number", + "min": -10, + "max": 99, + "step": 1, + "unit": "p/kWh", + "icon": "mdi:currency-usd", + "default": 99, + "enable": "car_charging_solar", + }, { "name": "car_charging_plan_max_price", "friendly_name": "Car Charging Plan max price", @@ -1362,6 +1406,40 @@ "restore": False, "manual": True, }, + { + "name": "manual_car_away", + "friendly_name": "Manual car away", + "type": "select", + "options": ["off"], + "icon": "mdi:car-off", + "default": "off", + "restore": False, + "manual": True, + "enable": "num_cars", + }, + { + "name": "manual_car_deadline", + "friendly_name": "Manual car ready by", + "type": "select", + "options": ["off"], + "icon": "mdi:car-clock", + "default": "off", + "restore": False, + "manual_rate": True, + "enable": "num_cars", + }, + { + "name": "manual_car_deadline_value", + "friendly_name": "Manual car ready by level", + "type": "input_number", + "min": 0, + "max": 100, + "step": 1, + "unit": "%", + "icon": "mdi:car-electric", + "default": 80, + "enable": "num_cars", + }, { "name": "manual_import_rates", "friendly_name": "Manual import rates", diff --git a/apps/predbat/fetch.py b/apps/predbat/fetch.py index 5c58a0a54..d0c993d01 100644 --- a/apps/predbat/fetch.py +++ b/apps/predbat/fetch.py @@ -1312,12 +1312,6 @@ def fetch_sensor_data(self, save=True): # "everything but the most expensive" threshold, which the scan then replaces with the real # low-rate band. Rendering on the provisional value painted most of the day as cheap. self.log_dispatch_timelines() - - # Work out car plan? - self.fetch_sensor_data_car_planning() - # Publish the car plan - self.publish_car_plan() - # Work out iBoost plan if self.iboost_enable and (((not self.iboost_solar) and (not self.iboost_charging)) or self.iboost_smart): self.iboost_plan = self.plan_iboost_smart() @@ -1372,6 +1366,15 @@ def fetch_sensor_data(self, save=True): else: self.load_inday_adjustment = 1.0 + # Work out car plan? Runs after the PV forecast so car_charging_solar can place slots on predicted + # sunshine, and after the load forecast is finished - the modal filter above rewrites load_minutes, + # the history forecast adds to load_forecast, and load_inday_adjustment is only computed here, all + # three of which feed the house load that solar surplus is measured against. Nothing between the + # PV forecast and this point uses the car plan. + self.fetch_sensor_data_car_planning() + # Publish the car plan + self.publish_car_plan() + force_replan = dynamic_load_car_changed # Compare on the change-detection signature, not the raw slots, so the per-cycle re-clocking # of an in-progress dispatch (start advanced to now, energy scaled to remaining time) does not @@ -1409,12 +1412,17 @@ def fetch_sensor_data_car_planning(self): self.log("Car {} on Octopus Intelligent, no active plan".format(car_n)) elif self.car_charging_planned[car_n] or self.car_charging_now[car_n]: limit_percent = dp1(self.car_charging_limit[car_n] / self.car_charging_battery_size[car_n] * 100) if self.car_charging_battery_size[car_n] else 0 + # Report the bought-energy target as well as the overall limit: with car_charging_plan_min_soc + # lowered they differ, and a log showing only the limit makes a correct plan look wrong + min_soc_kwh = min(dp3(self.car_charging_plan_min_soc * self.car_charging_battery_size[car_n] / 100.0), self.car_charging_limit[car_n]) self.log( - "Car {} plan charging from {}kWh to {}% ({}kWh), with slots {}, ready by {}".format( + "Car {} plan charging from {}kWh to {}% ({}kWh), buying only up to {}% ({}kWh), with slots {}, ready by {}".format( car_n, self.car_charging_soc[car_n], limit_percent, self.car_charging_limit[car_n], + self.car_charging_plan_min_soc, + min_soc_kwh, self.low_rates, self.car_charging_plan_time[car_n], ) @@ -2613,6 +2621,10 @@ def get_car_charging_planned(self): self.car_charging_planned_response = [str(response).lower() for response in self.get_arg("car_charging_planned_response", ["yes", "on", "enable", "true"])] self.car_charging_now_response = [str(response).lower() for response in self.get_arg("car_charging_now_response", ["yes", "on", "enable", "true", "charging"])] self.car_charging_from_battery = self.get_arg("car_charging_from_battery") + self.car_charging_solar = self.get_arg("car_charging_solar") + self.car_charging_solar_excess = self.get_arg("car_charging_solar_excess") + self.car_charging_rate_threshold_export = self.get_arg("car_charging_rate_threshold_export") + self.car_charging_plan_min_soc = self.get_arg("car_charging_plan_min_soc") # Car charging planned sensor for car_n in range(self.num_cars): @@ -3392,6 +3404,11 @@ def fetch_config_options(self): self.manual_freeze_charge_times = self.manual_times("manual_freeze_charge") self.manual_freeze_export_times = self.manual_times("manual_freeze_export") self.manual_demand_times = self.manual_times("manual_demand") + # Deliberately not folded into manual_all_times below: that set forces charge and export + # windows into existence, and this does the opposite - a slot the car is away for should not + # have a window manufactured for it. + self.manual_car_away_times = self.manual_times("manual_car_away") + self.manual_car_deadline_keep = self.manual_rates("manual_car_deadline", default_rate=self.get_arg("manual_car_deadline_value")) self.manual_all_times = self.manual_charge_times + self.manual_export_times + self.manual_demand_times + self.manual_freeze_charge_times + self.manual_freeze_export_times self.manual_api = self.api_select_update("manual_api") self.manual_import_rates = self.manual_rates("manual_import_rates", default_rate=self.get_arg("manual_import_value")) diff --git a/apps/predbat/output.py b/apps/predbat/output.py index 581000e08..107d19a43 100644 --- a/apps/predbat/output.py +++ b/apps/predbat/output.py @@ -198,6 +198,7 @@ def publish_car_plan(self): show["cost"] = cost if window.get("kwh_cancelled"): show["kwh_cancelled"] = dp2(window["kwh_cancelled"]) + show["solar"] = window.get("solar", False) total_cost += cost total_kwh += kwh plan.append(show) diff --git a/apps/predbat/plan.py b/apps/predbat/plan.py index 2a1e25a38..1e32e4e12 100644 --- a/apps/predbat/plan.py +++ b/apps/predbat/plan.py @@ -5903,6 +5903,372 @@ def plan_iboost_smart(self): plan = self.sort_window_by_time(plan) return plan + def car_slot_is_away(self, start, end): + """ + Is the car marked as away for any part of this slot? + + manual_car_away is a per-slot override saying the car will not be plugged in - out at the + shops, at work, anywhere. Predbat cannot tell that in advance: car_charging_planned reports + whether the car is plugged in now, so the plan happily schedules an afternoon charge for a car + that will not be there, and only discovers otherwise when the afternoon arrives. + + Overlap rather than an exact start match, so an away marker suppresses any window it touches + rather than only one that happens to begin on the same minute. + + Args: + - start, end: absolute minutes of the slot, half-open + + Returns: + - bool: True when the car is away for any part of it + """ + if not self.manual_car_away_times: + return False + return any((start < away + self.plan_interval_minutes) and (end > away) for away in self.manual_car_away_times) + + def car_solar_load_forecast(self): + """ + Build the central-case house load forecast used to size solar car slots + + This is the same step_data_history call calculate_plan() makes for load_minutes_step, and it has + to be repeated here rather than borrowed: car slots are planned during the fetch, before the + first prediction of the cycle exists, so the plan's copy is either absent or a cycle stale. + + The historical load it is built from already has car charging subtracted out of it (car_charging_hold + in get_filtered_load_window), so this is house load without the car - which is what the surplus + calculation needs, and why feeding the car's own draw back in as load cannot happen here. + + metric_load_divergence is the one input not yet computed this cycle, so the previous cycle's value + is used and the first cycle after a restart runs without it. It only moves load between adjacent + five minute buckets and leaves the total alone, so a half-hour surplus barely notices. + + Returns: + - dict: kWh of house load per PREDICT_STEP bucket, keyed by minutes from now + """ + return self.step_data_history( + self.load_minutes, + self.minutes_now, + forward=False, + scale_today=self.load_inday_adjustment, + scale_fixed=self.load_scaling, + type_load=True, + load_forecast=self.load_forecast, + load_scaling_dynamic=self.load_scaling_dynamic, + cloud_factor=getattr(self, "metric_load_divergence", None), + load_adjust=self.manual_load_adjust, + load_baseline=self.dynamic_load_baseline, + ) + + def car_solar_surplus_kwh(self, start, end, load_step, rate_kw=None): + """ + Forecast solar surplus over an absolute minute range, in kWh + + Surplus is forecast PV less forecast house load, which is what a charge-on-solar charger actually + gets to draw. It is floored at zero in every PREDICT_STEP bucket rather than once over the whole + range, because a car cannot charge on an average: a sunny first half followed by a dark second + half yields the sunny half, not their sum. + + rate_kw, when given, caps each bucket at what the charger can physically take. Without it a 10kW + midday surplus would be credited in full to a 7kW charger. + + Load buckets are anchored on minutes_now while the slots sit on the plan interval grid, so the two + are generally out of phase; each bucket contributes only the fraction of itself the range covers. + + Args: + - start, end: absolute minutes, half-open + - load_step: house load forecast from car_solar_load_forecast() + - rate_kw: charger limit in kW, or None for the raw surplus + + Returns: + - float: kWh of surplus available in the range + """ + step = PREDICT_STEP + total = 0.0 + # Start at the bucket containing the range start, never before now - surplus already in the past + # cannot be charged, and load_step has no buckets there either + first = max(0, int((start - self.minutes_now) // step) * step) + for offset in range(first, self.forecast_minutes + self.plan_interval_minutes, step): + bucket_start = self.minutes_now + offset + if bucket_start >= end: + break + overlap = min(end, bucket_start + step) - max(start, bucket_start) + if overlap <= 0: + continue + pv_kwh = 0.0 + for minute in range(bucket_start, bucket_start + step): + pv_kwh += self.pv_forecast_minute.get(minute, 0.0) + surplus = max(0.0, pv_kwh - load_step.get(offset, 0.0)) + if rate_kw is not None: + surplus = min(surplus, rate_kw * step / 60.0) + total += surplus * overlap / step + return total + + def car_slot_has_sun(self, start, end, load_step): + """ + Whether forecast solar exceeds the house load anywhere in a slot + + A car charging in such a slot is load the inverter feeds from that surplus before it charges the + battery, so even a slot Predbat buys from the grid takes the battery's sun. + + Args: + - start, end: absolute plan minutes + - load_step: house load forecast, as for car_solar_surplus_kwh() + + Returns: + - bool: True when there is surplus to lose + """ + start = max(start, self.minutes_now) + return end > start and self.car_solar_surplus_kwh(start, end, load_step) > 0.05 + + def car_solar_reserved_for_car(self, load_step, car_n=0): + """ + Surplus the car has first claim on ahead of the house battery, in kWh + + The battery-priority hold assumes the car can catch up later, which stops being true in two cases. + Mark the afternoon away and the morning becomes the car's last chance, so banking it for a battery + that has the whole day to fill is the wrong way round - with the battery taking surplus first the car + would never see a solar window at all. And a one-off deadline means the car needs a level by a time the battery + has no stake in, so the sun before it should go to the car rather than force a purchase the battery + then saves itself from. + + Each case claims what the car still needs, capped by what the slots it will be present for can + deliver before that case's cut-off - the whole forecast for away time, the deadline for a deadline - + so the hold gives up exactly that much and no more. The larger claim wins, since both are drawn from + the same earliest windows. Zero when neither is set, which leaves the everyday behaviour untouched. + + Args: + - load_step: house load forecast, as used for the surplus calculation + - car_n: which car + + Returns: + - float: kWh to keep back for the car, 0.0 when there is nothing to reserve + """ + if car_n >= self.num_cars: + return 0.0 + claims = [] + if self.manual_car_away_times: + claims.append((self.car_charging_limit[car_n], None)) + for deadline in self.car_one_off_deadlines(car_n): + claims.append((deadline["kwh"], deadline["minute"])) + + reserved = 0.0 + for target_kwh, until in claims: + needed = target_kwh - self.car_charging_soc[car_n] + if needed > 0: + reserved = max(reserved, self.car_solar_available_kwh(load_step, car_n, needed, until)) + return reserved + + def car_solar_available_kwh(self, load_step, car_n, needed, until=None): + """ + Surplus the car could take from the slots it is present for, up to a cut-off, capped at needed + + Args: + - load_step: house load forecast, as used for the surplus calculation + - car_n: which car + - needed: stop counting once this much is found + - until: absolute plan minute the energy must land by, or None for the whole forecast + + Returns: + - float: kWh available, no more than needed + """ + available = 0.0 + rate_limit = self.car_charging_rate[car_n] * self.plan_interval_minutes / 60.0 + start_minute = int(self.minutes_now / self.plan_interval_minutes) * self.plan_interval_minutes + for minute in range(start_minute, self.minutes_now + self.forecast_minutes, self.plan_interval_minutes): + slot_start = max(minute, self.minutes_now) + slot_end = minute + self.plan_interval_minutes + if until is not None and slot_end > until: + break + if slot_end <= slot_start or self.car_slot_is_away(slot_start, slot_end): + continue + available += min(self.car_solar_surplus_kwh(slot_start, slot_end, load_step), rate_limit) + if available >= needed: + return needed + return min(needed, available) + + def plan_car_charging_solar_windows(self, load_step=None): + """ + Find the slots where forecast solar is worth diverting to the car + + A slot qualifies when the forecast surplus - PV less house load, not raw PV - is at least + car_charging_solar_excess, and the export rate is no higher than car_charging_rate_threshold_export. + That second test is what "it doesn't make sense to export it" means in practice. Testing the surplus + rather than raw generation is what the setting's name has always claimed: 4kW of sun against a 4kW + house leaves the car nothing, and used to qualify anyway. + + Unlike the paid-import windows these run to the end of the forecast rather than stopping at the + ready time. Solar is opportunistic: it tops the car up above the guaranteed minimum whenever the + sun is free, and bounding it by a morning ready time would exclude every daylight hour. + + The house battery gets the surplus first, and the car is offered a window once the pack is predicted + full. A kWh in the pack displaces the evening peak; one in a car that has been promised nothing + displaces at most a cheap overnight top-up, so the pack is the better home for it. The exceptions are + a car that needs the sun more urgently - a one-off deadline, or away time later in the day - and + car_solar_reserved_for_car releases exactly what those need before the hold applies. + + Args: + - load_step: house load forecast, built here when the caller has not already done so + + Returns: + - list: candidate windows, each marked with solar=True + """ + if not self.car_charging_solar: + return [] + if load_step is None: + load_step = self.car_solar_load_forecast() + + windows = [] + slot_count = 0 + rejected_sun = 0 + rejected_export = 0 + rejected_battery = 0 + # Level the battery has to reach before the car is offered anything, and a running estimate of + # the pack as the held surplus fills it. Walking it forward means "full enough" is judged at the + # time of each slot rather than from the SoC right now, so a pack that gets there mid-morning + # releases the car mid-morning rather than holding all day. + # Surplus the car can only get from the slots it will be present for. The hold below gives up + # exactly this much, so away time moves the car's charge earlier instead of removing it: the + # battery still has the rest of the day, the car does not. + reserved_for_car = self.car_solar_reserved_for_car(load_step) + given_to_car = 0.0 + battery_priority_kwh = self.soc_max + battery_estimate = self.soc_kw + start_minute = int(self.minutes_now / self.plan_interval_minutes) * self.plan_interval_minutes + end_minute = self.minutes_now + self.forecast_minutes + for minute in range(start_minute, end_minute, self.plan_interval_minutes): + slot_count += 1 + # The slot the clock is currently inside started in the past; only its remainder is available + slot_start = max(minute, self.minutes_now) + slot_end = minute + self.plan_interval_minutes + if slot_end <= slot_start: + continue + # The threshold is a power, so the slot's energy is converted rather than compared directly: + # on a 30 minute plan interval 1.25kWh is 2.5kW. Keeping it a power means the setting does + # not silently change meaning if plan_interval_minutes is not 30, and lets a part slot be + # judged on the same footing as a whole one. + surplus_kwh = self.car_solar_surplus_kwh(slot_start, slot_end, load_step) + # The battery gets the surplus until it is predicted to reach the configured level. What it + # cannot physically take in the slot is not held back - that would strand surplus that the + # car could have used and the grid will otherwise buy at the midday rate. + if battery_estimate < battery_priority_kwh and given_to_car >= reserved_for_car: + to_battery = min( + surplus_kwh, + battery_priority_kwh - battery_estimate, + self.battery_rate_max_charge * self.battery_rate_max_scaling * (slot_end - slot_start), + ) + battery_estimate += max(to_battery, 0.0) + rejected_battery += 1 + continue + power_kw = surplus_kwh * 60.0 / (slot_end - slot_start) + if power_kw < self.car_charging_solar_excess: + rejected_sun += 1 + continue + export_rate = self.rate_export.get(minute, 0.0) + if export_rate > self.car_charging_rate_threshold_export: + rejected_export += 1 + continue + # Price the slot at the export rate: solar sent to the car is not bought, it is export + # given up, so that is its real cost and what the plan should show + # Count what this window hands over, so the hold resumes once the car's reserved share is met + given_to_car += min(surplus_kwh, self.car_charging_rate[0] * (slot_end - slot_start) / 60.0) if self.num_cars else 0.0 + windows.append({"start": minute, "end": slot_end, "average": export_rate, "solar": True, "power_kw": dp2(power_kw)}) + + if slot_count: + accepted = ", ".join("{}={}kW".format(self.time_abs_str(window["start"]), window["power_kw"]) for window in windows) + self.log( + "Car solar windows: {} of {} slots qualify (need forecast surplus >= {}kW and export rate <= {}), rejected {} for low surplus, {} for export rate and {} held for the battery{}".format( + len(windows), + slot_count, + self.car_charging_solar_excess, + self.car_charging_rate_threshold_export, + rejected_sun, + rejected_export, + rejected_battery, + " - accepted: " + accepted if accepted else "", + ) + ) + return windows + + def car_one_off_deadlines(self, car_n=0): + """ + The charge levels promised by manual_car_deadline, as absolute plan minutes and kWh + + manual_car_deadline is a one-off "have the car at this level by this time", set from the plan page for + a trip that does not fit the everyday ready time. The level is a percentage of the car's battery, so it + is applied to the first car only - one percentage across several cars of different sizes would promise + something nobody asked for. It is capped at car_charging_limit: the car will not charge past its own + limit whatever the plan says, and planning energy it will refuse would put phantom car load into the + forecast the battery is planned against. + + Args: + - car_n: which car + + Returns: + - list: {"minute", "kwh", "percent"} dicts in time order, empty when no deadline is set + """ + keep = self.manual_car_deadline_keep + if not keep or car_n != 0 or car_n >= self.num_cars: + return [] + deadlines = [] + for minute in sorted(keep): + # manual_rates() fills every minute of the slot; the deadline is the slot's first minute + if (minute - 1) in keep: + continue + percent = keep[minute] + kwh = dp3(percent * self.car_charging_battery_size[car_n] / 100.0) + if kwh > self.car_charging_limit[car_n]: + self.log("Warn: Car {} ready-by level {}% at {} is above its charge limit of {}kWh, planning to the limit instead - raise the car's own limit to reach it".format(car_n, percent, self.time_abs_str(minute), self.car_charging_limit[car_n])) + kwh = self.car_charging_limit[car_n] + deadlines.append({"minute": minute, "kwh": kwh, "percent": percent}) + return deadlines + + def car_deadlines(self, car_n, ready_minutes, min_soc_kwh): + """ + Every charge level the car has been promised, in time order + + The everyday one is car_charging_plan_min_soc by car_charging_plan_time. manual_car_deadline adds a one-off + on top. Both are guarantees rather than hopes: the free sun that lands before each one is counted first, + and whatever it leaves is bought before the deadline passes. + + Args: + - car_n: which car + - ready_minutes: the everyday ready time, as an absolute plan minute + - min_soc_kwh: the everyday minimum, in kWh + + Returns: + - list: {"minute", "kwh", "one_off"} dicts sorted by minute + """ + deadlines = [{"minute": ready_minutes, "kwh": min_soc_kwh, "one_off": False}] + for deadline in self.car_one_off_deadlines(car_n): + deadlines.append({"minute": deadline["minute"], "kwh": deadline["kwh"], "one_off": True}) + deadlines.sort(key=lambda deadline: deadline["minute"]) + return deadlines + + def car_deadline_purchase_windows(self, car_n, deadline_minute): + """ + Every import slot between now and a one-off deadline, in the order the car should buy them + + The everyday minimum buys from low_rates, which only holds the windows under the import threshold. A + one-off deadline is an explicit "I need this", so it may buy from any slot before it: restricting it to + the cheap band would let a midday deadline fail on a tariff whose cheap band is overnight and has + already passed. Cheapest first in smart mode, soonest first otherwise, matching the everyday pass. + + Args: + - car_n: which car + - deadline_minute: absolute plan minute the level is promised by + + Returns: + - list: import windows as {"start", "end", "average"} dicts + """ + windows = [] + start_minute = int(self.minutes_now / self.plan_interval_minutes) * self.plan_interval_minutes + for minute in range(start_minute, deadline_minute, self.plan_interval_minutes): + windows.append({"start": minute, "end": minute + self.plan_interval_minutes, "average": self.rate_import.get(minute, self.rate_min)}) + if windows and self.car_charging_plan_smart[car_n]: + order = self.sort_window_by_price(windows, reverse_time=True) + order.reverse() + windows = [windows[window_n] for window_n in order] + return windows + def plan_car_charging(self, car_n, low_rates): """ Plan when the car will charge, taking into account ready time and pricing @@ -5932,35 +6298,107 @@ def plan_car_charging(self, car_n, low_rates): # car_charging_now never adds a slot here: this plan drives binary_sensor.predbat_car_charging_slot, # which starts the charger, so a slot for "charging now" kept a charge going on its own. The hold # for a car charging outside the plan is execute_plan()'s, and dynamic load models its load. - for window_n in price_sorted: - window = low_rates[window_n] + + # Solar surplus windows are considered first, so free sunshine is used before any paid import and + # the price pass below only has to cover whatever solar cannot deliver. The house load forecast is + # built once and handed down rather than rebuilt per window - step_data_history walks every + # previous day for every bucket and is far too expensive to call inside the loop. + load_step = self.car_solar_load_forecast() if self.car_charging_solar else {} + solar_windows = self.plan_car_charging_solar_windows(load_step) + bought_windows = [low_rates[window_n] for window_n in price_sorted] + + # A car charging while the sun is up is load the inverter feeds from the surplus before it charges the + # battery, so a "bought" slot in the sun takes the battery's solar too. With equal import prices that makes + # a sunny slot the worse choice - reported from a live system buying at 14:00 in full sun when midnight cost + # the same - so among equally priced slots the sunless ones go first. Only reorders ties, so it cannot make + # a plan dearer. + if load_step and self.car_charging_plan_smart[car_n]: + sunny = {id(window): self.car_slot_has_sun(window["start"], window["end"], load_step) for window in bought_windows} + bought_windows.sort(key=lambda window: (window["average"], sunny[id(window)])) + + # Energy that must be there by the ready time, whatever the weather. Bought slots stop here; + # solar carries on to the full limit, which is how "minimum from any source, the rest from sun" + # is expressed. Left at 100% (the default) both targets are the same and nothing changes. + min_soc_kwh = min(dp3(self.car_charging_plan_min_soc * self.car_charging_battery_size[car_n] / 100.0), self.car_charging_limit[car_n]) + deadlines = self.car_deadlines(car_n, ready_minutes, min_soc_kwh) + horizon = self.minutes_now + self.forecast_minutes + + # Each deadline in turn gets the free sun that lands before it, then buys whatever that leaves. Only + # solar that fully lands before a deadline can count towards it: a single running car_soc is shared by + # every pass, so offering later sun first let tomorrow's sunshine satisfy "60% by 07:30" and break the + # loop before one overnight slot was bought - reported from a live system as a car left short all night + # against 39p import it was never offered. Solar reaches for the full limit wherever it appears, since + # it is free; only bought energy stops at the level promised. The everyday minimum buys from low_rates + # and honours car_charging_plan_max_price as it always has; a one-off deadline may buy any slot before + # it and ignores the price cap, because it is an explicit instruction that the charge is needed. + # Whatever sun is left then tops the car up towards the limit, whenever it arrives. + # Entries are (window, target_kwh, end_limit, capped_by_max_price). + candidates = [] + for deadline in deadlines: + for window in solar_windows: + if window["end"] <= deadline["minute"]: + candidates.append((window, self.car_charging_limit[car_n], horizon, False)) + purchase = self.car_deadline_purchase_windows(car_n, deadline["minute"]) if deadline["one_off"] else bought_windows + for window in purchase: + candidates.append((window, deadline["kwh"], deadline["minute"], not deadline["one_off"])) + if deadline["one_off"]: + self.log("Car {} promised {}kWh by {}".format(car_n, deadline["kwh"], self.time_abs_str(deadline["minute"]))) + for window in solar_windows: + candidates.append((window, self.car_charging_limit[car_n], horizon, False)) + + for window, window_target, end_limit, capped_by_max_price in candidates: + is_solar = window.get("solar", False) start = max(window["start"], self.minutes_now) - end = min(window["end"], ready_minutes) + end = min(window["end"], end_limit) price = window["average"] length = 0 kwh = 0 + # The car is not here, so nothing can go into it - not cheap import, not surplus solar. + # Tested before the have-enough check below so a skipped slot cannot end the loop: the + # charge it would have taken has to move to a slot the car is actually present for. + if self.car_slot_is_away(start, end): + continue + # Stop once we have enough charge, allow small margin for rounding if (car_soc + 0.1) >= self.car_charging_limit[car_n]: break + # Enough charge for what this window is allowed to deliver. Skip rather than stop: a later + # deadline may promise a higher level, and later sun still tops up towards the limit + if (car_soc + 0.1) >= window_target: + continue + # Skip past windows if end <= start: continue - # Skip over prices when they are too high - if (max_price != 0) and price > max_price: + # A solar window and a cheap-import window can cover the same time; only plan one of them + if any((start < slot["end"]) and (end > slot["start"]) for slot in plan): + continue + + # Skip over prices when they are too high. Solar windows are exempt: their energy is not + # being bought, so the import price of that time of day says nothing about them. So are a + # one-off deadline's purchases, which the user has said are needed whatever they cost. + if capped_by_max_price and (max_price != 0) and price > max_price and not is_solar: continue - # Compute amount of charge + # Compute amount of charge. A bought slot draws the charger's full rate because that is what + # Predbat is asking for. A solar slot draws only what the sun leaves over the house, because + # the charger modulates - assuming the full rate there put a 7kW draw in the forecast against + # a 2kW surplus, inflating predicted load and making the battery look like it had to cover + # the difference. length = end - start hours = length / 60 - kwh = self.car_charging_rate[car_n] * hours + if is_solar: + kwh = self.car_solar_surplus_kwh(start, end, load_step, rate_kw=self.car_charging_rate[car_n]) + else: + kwh = self.car_charging_rate[car_n] * hours kwh_add = kwh * self.car_charging_loss - kwh_left = max(self.car_charging_limit[car_n] - car_soc, 0) + kwh_left = max(window_target - car_soc, 0) # Clamp length to required amount (shorten the window) if kwh_add > kwh_left: @@ -5968,7 +6406,14 @@ def plan_car_charging(self, car_n, low_rates): length = int(min(round(((length * percent) / 5) + 0.5, 0) * 5, end - start)) end = start + length hours = length / 60 - kwh = self.car_charging_rate[car_n] * hours + # Recompute rather than scale: solar is not flat across the window, so the proportional + # length above is only a first guess at where to cut and the surplus in the part that + # survives has to be measured. It can come out under kwh_left, which is fine - a later + # window picks up the remainder. + if is_solar: + kwh = self.car_solar_surplus_kwh(start, end, load_step, rate_kw=self.car_charging_rate[car_n]) + else: + kwh = self.car_charging_rate[car_n] * hours kwh_add = min(kwh * self.car_charging_loss, kwh_left) kwh = kwh_add / self.car_charging_loss @@ -5982,8 +6427,18 @@ def plan_car_charging(self, car_n, low_rates): new_slot["average"] = window["average"] new_slot["cost"] = dp2(new_slot["average"] * kwh) new_slot["octopus"] = False + new_slot["solar"] = window.get("solar", False) plan.append(new_slot) + # Say so when a promise cannot be kept. Away time can leave too few slots before a deadline, and the + # charger rate caps what each one delivers - better a warning now than a car found short at the time. + for deadline in deadlines: + if not deadline["one_off"]: + continue + reached = self.car_charging_soc[car_n] + sum(slot["kwh"] * self.car_charging_loss for slot in plan if slot["end"] <= deadline["minute"]) + if reached + 0.1 < deadline["kwh"]: + self.log("Warn: Car {} can only reach {}kWh of the {}kWh promised by {} - too few slots the car is present for before then, at its charge rate".format(car_n, dp2(reached), deadline["kwh"], self.time_abs_str(deadline["minute"]))) + # Return sorted back in time order plan = self.sort_window_by_time(plan) return plan diff --git a/apps/predbat/predbat.py b/apps/predbat/predbat.py index 318429f9f..a466836b7 100644 --- a/apps/predbat/predbat.py +++ b/apps/predbat/predbat.py @@ -352,6 +352,8 @@ def reset(self): self.manual_freeze_charge_times = [] self.manual_freeze_export_times = [] self.manual_demand_times = [] + self.manual_car_away_times = [] + self.manual_car_deadline_keep = {} self.manual_all_times = [] self.manual_api = [] self.manual_import_rates = {} diff --git a/apps/predbat/tests/test_car_away.py b/apps/predbat/tests/test_car_away.py new file mode 100644 index 000000000..1e4f22b16 --- /dev/null +++ b/apps/predbat/tests/test_car_away.py @@ -0,0 +1,253 @@ +# ----------------------------------------------------------------------------- +# Predbat Home Battery System +# Copyright Trefor Southwell 2026 - All Rights Reserved +# This application maybe used for personal use only and not for commercial use +# ----------------------------------------------------------------------------- +# fmt off +# pylint: disable=consider-using-f-string +# pylint: disable=line-too-long +# pylint: disable=attribute-defined-outside-init + +"""Tests for manual_car_away, the per-slot override saying the car will not be plugged in. + +car_charging_planned tells Predbat whether the car is plugged in *now*, so a plan made this morning +happily schedules an afternoon charge for a car that will be out, and only finds out when the +afternoon comes. This override is how the user says so in advance. + +A slot the car is away for is skipped outright rather than shortened or repriced: if the car is not +there, nothing can go into it from any source. +""" + +from tests.test_infra import reset_rates, update_rates_import + + +def ready_time_str(my_predbat, minutes_ahead): + """An HH:MM:SS ready time the given number of minutes after the test clock.""" + target = (my_predbat.minutes_now + minutes_ahead) % (24 * 60) + return "{:02d}:{:02d}:00".format(target // 60, target % 60) + + +def setup_car(my_predbat, car_kwh=10.0, ready_ahead=720, rate=7.0): + """A single car needing car_kwh within ready_ahead minutes, with no away slots.""" + my_predbat.num_cars = 1 + my_predbat.car_charging_soc = [0.0] + my_predbat.car_charging_limit = [car_kwh] + my_predbat.car_charging_battery_size = [50.0] + my_predbat.car_charging_rate = [rate] + my_predbat.car_charging_loss = 1.0 + my_predbat.car_charging_slots = [[]] + my_predbat.car_charging_plan_time = [ready_time_str(my_predbat, ready_ahead)] + my_predbat.car_charging_plan_smart = [True] + my_predbat.car_charging_plan_max_price = [0.0] + my_predbat.car_charging_now = [False] + my_predbat.manual_car_away_times = [] + + +def build_low_rates(my_predbat, count=24): + """Half-hour import windows, all the same price so slot choice is by time, not price.""" + return [{"start": my_predbat.minutes_now + 30 * n, "end": my_predbat.minutes_now + 30 * (n + 1), "average": 5.0} for n in range(count)] + + +def test_away_slots_are_skipped(my_predbat): + """A slot marked away yields no charging, and the charge moves to slots the car is present for.""" + print(" - test_away_slots_are_skipped") + failed = False + setup_car(my_predbat) + reset_rates(my_predbat, 5.0, 1.0) + low_rates = build_low_rates(my_predbat) + update_rates_import(my_predbat, low_rates) + + baseline = my_predbat.plan_car_charging(0, low_rates) + if not baseline: + print("ERROR: expected a baseline plan with no away slots") + return True + baseline_kwh = sum(slot["kwh"] for slot in baseline) + + # Mark the first four half-hours away + away = [my_predbat.minutes_now + 30 * n for n in range(4)] + my_predbat.manual_car_away_times = away + plan = my_predbat.plan_car_charging(0, low_rates) + + for slot in plan: + for marker in away: + if slot["start"] < marker + my_predbat.plan_interval_minutes and slot["end"] > marker: + print("ERROR: slot {}-{} overlaps an away marker at {}".format(slot["start"], slot["end"], marker)) + failed = True + + # The car still needs the same energy, it just has to come from later slots + if abs(sum(slot["kwh"] for slot in plan) - baseline_kwh) > 0.2: + print("ERROR: the charge should move, not vanish - baseline {} got {}".format(baseline_kwh, sum(slot["kwh"] for slot in plan))) + failed = True + + return failed + + +def test_away_does_not_end_the_search(my_predbat): + """An away slot must not stop the loop, or every later slot is lost with it. + + The have-enough test breaks out of the loop; an away slot has to be skipped before reaching it, + or the first away marker silently truncates the whole plan. + """ + print(" - test_away_does_not_end_the_search") + failed = False + setup_car(my_predbat) + reset_rates(my_predbat, 5.0, 1.0) + low_rates = build_low_rates(my_predbat) + update_rates_import(my_predbat, low_rates) + + # Away for the very first slot only - everything after it must still be planned + my_predbat.manual_car_away_times = [my_predbat.minutes_now] + plan = my_predbat.plan_car_charging(0, low_rates) + + if not plan: + print("ERROR: an away marker on the first slot wiped out the whole plan") + return True + if sum(slot["kwh"] for slot in plan) < (my_predbat.car_charging_limit[0] - 0.5): + print("ERROR: the car should still reach its limit from later slots, got {}".format(sum(slot["kwh"] for slot in plan))) + failed = True + return failed + + +def test_away_matches_on_overlap(my_predbat): + """Suppression is by overlap, not by an exact start match.""" + print(" - test_away_matches_on_overlap") + failed = False + setup_car(my_predbat) + + interval = my_predbat.plan_interval_minutes + marker = my_predbat.minutes_now + 60 + my_predbat.manual_car_away_times = [marker] + + # A window starting inside the away slot overlaps it + if not my_predbat.car_slot_is_away(marker + 5, marker + interval + 5): + print("ERROR: a window starting inside an away slot should be suppressed") + failed = True + # One ending inside it also overlaps + if not my_predbat.car_slot_is_away(marker - 5, marker + 5): + print("ERROR: a window ending inside an away slot should be suppressed") + failed = True + # One entirely before it does not + if my_predbat.car_slot_is_away(marker - interval, marker): + print("ERROR: a window ending exactly as the away slot starts should not be suppressed") + failed = True + # Nor one entirely after + if my_predbat.car_slot_is_away(marker + interval, marker + interval * 2): + print("ERROR: a window starting after the away slot ends should not be suppressed") + failed = True + + # With nothing configured, nothing is ever away + my_predbat.manual_car_away_times = [] + if my_predbat.car_slot_is_away(marker, marker + interval): + print("ERROR: with no away times configured nothing should be suppressed") + failed = True + return failed + + +def test_away_survives_an_unset_config_value(my_predbat): + """A registered-but-unset selection must decode to nothing rather than crashing. + + manual_car_away is gated behind an "enable", unlike the other manual selects, so before the UI + has published it the stored value reads back as None rather than "". manual_times() called + .replace() on that directly, which took fetch_config_options() down with an AttributeError - + seen as the headless annual bootstrap failing to start at all. + """ + print(" - test_away_survives_an_unset_config_value") + failed = False + item = my_predbat.config_index.get("manual_car_away") + if item is None: + print("ERROR: manual_car_away is not registered") + return True + saved = item.get("value", "") + try: + item["value"] = None + try: + result = my_predbat.manual_times("manual_car_away", update=False) + except Exception as e: + print("ERROR: an unset value raised {}: {}".format(type(e).__name__, e)) + return True + if result: + print("ERROR: an unset value should decode to no slots, got {}".format(result)) + failed = True + # And the same for an empty string, which is what a cleared selection looks like + item["value"] = "" + if my_predbat.manual_times("manual_car_away", update=False): + print("ERROR: an empty value should decode to no slots") + failed = True + finally: + item["value"] = saved + return failed + + +def test_away_is_reachable_from_the_plan_ui(my_predbat): + """The plan table must offer Car Away, send it, and clear it again. + + The config item and the planner logic are not enough on their own: the plan's time cells build + their dropdown from an explicit list of actions in web_helper.py and post them to a handler in + web.py that matches on the action string. Miss either and the setting exists but there is no way + to reach it from the plan, which is where it is actually useful - the first version of this + feature shipped exactly that way. + + Reads the sources rather than driving a browser, which is what the surrounding web tests do. + """ + print(" - test_away_is_reachable_from_the_plan_ui") + import os + + failed = False + here = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + helper = open(os.path.join(here, "web_helper.py")).read() + web = open(os.path.join(here, "web.py")).read() + + if "'Car Away'" not in helper: + print("ERROR: the plan time-cell dropdown does not offer a Car Away action") + failed = True + if "manual_car_away_times" not in helper: + print("ERROR: the plan renderer never reads manual_car_away_times, so a set slot cannot show as set") + failed = True + if helper.count("manual_car_away_times") < 4: + print("ERROR: expected the away times in the action union, both highlight blocks and the menu test, got {}".format(helper.count("manual_car_away_times"))) + failed = True + + if 'action == "Car Away"' not in web: + print("ERROR: the plan_override handler does not accept the Car Away action") + failed = True + if web.count('"manual_car_away_times": manual_car_away_times') < 2: + print("ERROR: both plan payloads must carry the away times or one of the two views cannot show them") + failed = True + + # Clear has to reach this select too, or a marker set from the plan can never be removed there + clear_block = web.split('if action == "Clear":')[1].split("else:")[0] if 'if action == "Clear":' in web else "" + if "manual_car_away" not in clear_block: + print("ERROR: Clear does not clear manual_car_away, so an away marker cannot be undone from the plan") + failed = True + + return failed + + +def run_car_away_tests(my_predbat): + """Run every manual_car_away test.""" + print("**** Running car away tests ****\n") + carried = ( + "num_cars", + "car_charging_soc", + "car_charging_limit", + "car_charging_battery_size", + "car_charging_rate", + "car_charging_loss", + "car_charging_slots", + "car_charging_plan_time", + "car_charging_plan_smart", + "car_charging_plan_max_price", + "car_charging_now", + "manual_car_away_times", + ) + saved = {name: getattr(my_predbat, name, None) for name in carried} + try: + failed = test_away_slots_are_skipped(my_predbat) + failed |= test_away_does_not_end_the_search(my_predbat) + failed |= test_away_matches_on_overlap(my_predbat) + failed |= test_away_survives_an_unset_config_value(my_predbat) + failed |= test_away_is_reachable_from_the_plan_ui(my_predbat) + finally: + for name, value in saved.items(): + setattr(my_predbat, name, value) + return failed diff --git a/apps/predbat/tests/test_car_deadline.py b/apps/predbat/tests/test_car_deadline.py new file mode 100644 index 000000000..9994b75d3 --- /dev/null +++ b/apps/predbat/tests/test_car_deadline.py @@ -0,0 +1,571 @@ +# ----------------------------------------------------------------------------- +# Predbat Home Battery System +# Copyright Trefor Southwell 2026 - All Rights Reserved +# This application maybe used for personal use only and not for commercial use +# ----------------------------------------------------------------------------- +# fmt off +# pylint: disable=consider-using-f-string +# pylint: disable=line-too-long +# pylint: disable=attribute-defined-outside-init + +"""Tests for manual_car_deadline, the one-off "have the car at this level by this time". + +The everyday guarantee is car_charging_plan_min_soc by car_charging_plan_time. That covers a routine, +not an event: a trip at 15:00 that needs 80% does not fit a 07:30 ready time, and editing the standing +settings for it means remembering to put them back. The deadline is set from the plan page instead and +expires on its own once its slot has passed. + +It is a guarantee rather than a hope, in the same way the everyday minimum is. Free sun that lands +before the deadline is counted first, and whatever it leaves is bought before the deadline passes - from +any import slot, cheapest first, and regardless of car_charging_plan_max_price, because the user has +said the charge is needed. +""" + +from datetime import timedelta + +from tests.test_car_solar import set_pv, setup_car +from tests.test_infra import reset_rates, update_rates_import + + +def setup_deadline_car(my_predbat, soc=10.0, limit=40.0, min_soc=0, ready_ahead=720): + """A 50kWh car at soc kWh with a limit of limit kWh, and no standing minimum unless asked for. + + min_soc defaults to 0 so the everyday guarantee asks for nothing, which isolates what the deadline + itself causes: anything bought in these tests is bought because of it. + """ + setup_car(my_predbat, car_kwh=limit, ready_ahead=ready_ahead) + my_predbat.car_charging_soc = [soc] + my_predbat.car_charging_plan_min_soc = min_soc + my_predbat.manual_car_away_times = [] + my_predbat.manual_car_deadline_keep = {} + reset_rates(my_predbat, 30.0, 5.0) + + +def set_deadline(my_predbat, minutes_ahead, percent): + """Promise percent by the slot minutes_ahead from now, stored the way manual_rates() decodes it.""" + start = int((my_predbat.minutes_now + minutes_ahead) / my_predbat.plan_interval_minutes) * my_predbat.plan_interval_minutes + my_predbat.manual_car_deadline_keep = {minute: percent for minute in range(start, start + my_predbat.plan_interval_minutes)} + return start + + +def bought_by(plan, minute): + """kWh bought from the grid in slots that finish by minute.""" + return sum(slot["kwh"] for slot in plan if not slot.get("solar") and slot["end"] <= minute) + + +def solar_by(plan, minute): + """kWh taken from the sun in slots that finish by minute.""" + return sum(slot["kwh"] for slot in plan if slot.get("solar") and slot["end"] <= minute) + + +def test_deadline_buys_before_it(my_predbat): + """With no sun and no standing minimum, the deadline's shortfall is bought before it passes.""" + print(" - test_deadline_buys_before_it") + failed = False + setup_deadline_car(my_predbat) + + if my_predbat.plan_car_charging(0, []): + print("ERROR: with no deadline and no minimum nothing should be planned, so this test proves nothing") + return True + + # 10kWh in a 50kWh car, 60% promised = 30kWh, so 20kWh has to be bought within four hours + deadline = set_deadline(my_predbat, 240, 60) + plan = my_predbat.plan_car_charging(0, []) + bought = bought_by(plan, deadline) + if abs(bought - 20.0) > 0.1: + print("ERROR: expected 20kWh bought before the deadline, got {} from {}".format(bought, plan)) + failed = True + if any(slot["end"] > deadline for slot in plan): + print("ERROR: with no sun, nothing should be planned after the deadline: {}".format(plan)) + failed = True + return failed + + +def test_deadline_ignores_the_price_cap(my_predbat): + """car_charging_plan_max_price still binds the everyday minimum, but not a one-off deadline.""" + print(" - test_deadline_ignores_the_price_cap") + failed = False + # Every slot costs 30p against a 10p cap + setup_deadline_car(my_predbat, min_soc=60, ready_ahead=240) + my_predbat.car_charging_plan_max_price = [10.0] + low_rates = [{"start": my_predbat.minutes_now + 30 * n, "end": my_predbat.minutes_now + 30 * (n + 1), "average": 30.0} for n in range(8)] + + everyday = my_predbat.plan_car_charging(0, low_rates) + if everyday: + print("ERROR: the everyday minimum should still respect the price cap, got {}".format(everyday)) + failed = True + + deadline = set_deadline(my_predbat, 240, 60) + plan = my_predbat.plan_car_charging(0, low_rates) + if bought_by(plan, deadline) < 19.9: + print("ERROR: a one-off deadline is an explicit need and should buy past the price cap, got {}".format(plan)) + failed = True + return failed + + +def test_deadline_buys_outside_the_cheap_band(my_predbat): + """A deadline may buy any slot before it, not only the ones in low_rates. + + low_rates holds the windows under the import threshold. On a tariff whose cheap band is overnight, a + midday deadline would otherwise find nothing it is allowed to buy and fail silently. + """ + print(" - test_deadline_buys_outside_the_cheap_band") + failed = False + setup_deadline_car(my_predbat) + # The only cheap windows start after the deadline + low_rates = [{"start": my_predbat.minutes_now + 300 + 30 * n, "end": my_predbat.minutes_now + 330 + 30 * n, "average": 5.0} for n in range(8)] + update_rates_import(my_predbat, low_rates) + + deadline = set_deadline(my_predbat, 240, 60) + plan = my_predbat.plan_car_charging(0, low_rates) + if bought_by(plan, deadline) < 19.9: + print("ERROR: the deadline should buy at the day rate when the cheap band is too late, got {}".format(plan)) + failed = True + return failed + + +def test_deadline_buys_the_cheapest_slots(my_predbat): + """In smart mode the deadline's purchases are the cheapest slots before it, not the soonest.""" + print(" - test_deadline_buys_the_cheapest_slots") + failed = False + setup_deadline_car(my_predbat) + # Two expensive slots up front, then six cheap ones: 6 x 3.7kWh = 22.2kWh covers the 20kWh needed + windows = [{"start": my_predbat.minutes_now + 30 * n, "end": my_predbat.minutes_now + 30 * (n + 1), "average": 50.0 if n < 2 else 10.0} for n in range(8)] + update_rates_import(my_predbat, windows) + + deadline = set_deadline(my_predbat, 240, 60) + plan = my_predbat.plan_car_charging(0, []) + bought = [slot for slot in plan if not slot.get("solar") and slot["end"] <= deadline] + if not bought: + print("ERROR: nothing was bought before the deadline") + failed = True + elif any(slot["average"] != 10.0 for slot in bought): + print("ERROR: expected only the 10p slots to be bought, got {}".format(bought)) + failed = True + return failed + + +def test_deadline_uses_sun_first(my_predbat): + """Sun that lands before the deadline is used before anything is bought.""" + print(" - test_deadline_uses_sun_first") + failed = False + setup_deadline_car(my_predbat) + my_predbat.car_charging_solar = True + # 7kW for three hours ending an hour before the deadline: 21kWh, more than the 20kWh needed + set_pv(my_predbat, 7.0, start_offset=60, length=180) + + deadline = set_deadline(my_predbat, 300, 60) + plan = my_predbat.plan_car_charging(0, []) + if bought_by(plan, deadline) > 0.1: + print("ERROR: the sun before the deadline covers it, so nothing should be bought: {}".format(plan)) + failed = True + if solar_by(plan, deadline) < 19.9: + print("ERROR: expected the deadline to be met from the sun, got {}".format(plan)) + failed = True + return failed + + +def test_deadline_does_not_count_later_sun(my_predbat): + """Sun after the deadline cannot meet it, so the shortfall is bought - and the sun still tops up after. + + This is the failure that left a car short overnight against its everyday minimum: sun arriving after + the deadline was counted towards it, and the purchase that should have covered it was never made. + """ + print(" - test_deadline_does_not_count_later_sun") + failed = False + setup_deadline_car(my_predbat) + my_predbat.car_charging_solar = True + set_pv(my_predbat, 7.0, start_offset=300, length=300) + + deadline = set_deadline(my_predbat, 240, 60) + plan = my_predbat.plan_car_charging(0, []) + if bought_by(plan, deadline) < 19.9: + print("ERROR: the sun comes after the deadline, so its shortfall must be bought, got {}".format(plan)) + failed = True + if not any(slot.get("solar") and slot["start"] >= deadline for slot in plan): + print("ERROR: the later sun should still top the car up towards its limit: {}".format(plan)) + failed = True + return failed + + +def test_deadline_is_capped_at_the_limit(my_predbat): + """A level above the car's charge limit plans to the limit - the car will not take more.""" + print(" - test_deadline_is_capped_at_the_limit") + failed = False + setup_deadline_car(my_predbat, limit=40.0) + deadline = set_deadline(my_predbat, 360, 100) + + promised = my_predbat.car_one_off_deadlines(0) + if len(promised) != 1 or abs(promised[0]["kwh"] - 40.0) > 0.001: + print("ERROR: a 100% deadline on a 40kWh limit should be capped to 40kWh, got {}".format(promised)) + failed = True + total = sum(slot["kwh"] for slot in my_predbat.plan_car_charging(0, [])) + if total > 30.0 + 0.1: + print("ERROR: planned {}kWh into a car with only 30kWh of room below its limit".format(total)) + failed = True + if bought_by(my_predbat.plan_car_charging(0, []), deadline) < 29.9: + print("ERROR: the capped level should still be bought in full before the deadline") + failed = True + return failed + + +def test_both_deadlines_are_kept(my_predbat): + """The everyday minimum and a later one-off deadline are each met by their own time.""" + print(" - test_both_deadlines_are_kept") + failed = False + # Everyday: 40% (20kWh) in two hours. One-off: 80% (40kWh) in six. + setup_deadline_car(my_predbat, soc=10.0, limit=40.0, min_soc=40, ready_ahead=120) + low_rates = [{"start": my_predbat.minutes_now + 30 * n, "end": my_predbat.minutes_now + 30 * (n + 1), "average": 5.0} for n in range(24)] + update_rates_import(my_predbat, low_rates) + ready = my_predbat.minutes_now + 120 + + deadline = set_deadline(my_predbat, 360, 80) + plan = my_predbat.plan_car_charging(0, low_rates) + if bought_by(plan, ready) < 9.9: + print("ERROR: the everyday minimum needs 10kWh by its ready time, got {} from {}".format(bought_by(plan, ready), plan)) + failed = True + if bought_by(plan, deadline) < 29.9: + print("ERROR: the one-off deadline needs 30kWh by its time, got {} from {}".format(bought_by(plan, deadline), plan)) + failed = True + return failed + + +def test_deadline_claims_sun_ahead_of_the_battery(my_predbat): + """A battery-priority hold must not bank the sun a deadline is counting on. + + With the battery first, the morning surplus would fill the pack while the car's deadline bought from + the grid - paying for energy the sun was providing for free, to spare a battery that has the rest of + the day to fill. The deadline claims the sun before it, exactly as away time claims the sun the car + will be present for. + """ + print(" - test_deadline_claims_sun_ahead_of_the_battery") + failed = False + setup_deadline_car(my_predbat) + my_predbat.car_charging_solar = True + my_predbat.soc_max = 10.0 + my_predbat.soc_kw = 0.0 + my_predbat.battery_rate_max_charge = 5.0 / 60.0 + my_predbat.battery_rate_max_scaling = 1.0 + # 21kWh of sun ending an hour before the deadline; the empty 10kWh pack would take the first 4 slots + set_pv(my_predbat, 7.0, start_offset=60, length=180) + load_step = my_predbat.car_solar_load_forecast() + + if my_predbat.car_solar_reserved_for_car(load_step) != 0.0: + print("ERROR: with no deadline and no away time nothing should be reserved for the car") + failed = True + + deadline = set_deadline(my_predbat, 300, 60) + reserved = my_predbat.car_solar_reserved_for_car(load_step) + if abs(reserved - 20.0) > 0.1: + print("ERROR: the deadline needs 20kWh and the sun before it can deliver that, so 20kWh should be reserved, got {}".format(reserved)) + failed = True + + plan = my_predbat.plan_car_charging(0, []) + if bought_by(plan, deadline) > 0.1: + print("ERROR: the sun before the deadline should go to the car rather than the battery, but {}kWh was bought: {}".format(bought_by(plan, deadline), plan)) + failed = True + return failed + + +def mark_away(my_predbat, start_ahead, end_ahead): + """Mark the car away for the slots between start_ahead and end_ahead minutes from now.""" + base = my_predbat.minutes_now + my_predbat.manual_car_away_times = list(range(base + start_ahead, base + end_ahead, my_predbat.plan_interval_minutes)) + + +def overlaps_away(my_predbat, plan): + """Planned slots that touch a slot the car is marked away for.""" + return [slot for slot in plan if my_predbat.car_slot_is_away(slot["start"], slot["end"])] + + +def test_deadline_buys_around_away_time(my_predbat): + """Away slots before a deadline are skipped, so its purchases move to slots the car is there for. + + The cheapest slots here are exactly the ones the car is away for. Buying them would plan a charge into + a car that is not plugged in, and the deadline would be missed for real. + """ + print(" - test_deadline_buys_around_away_time") + failed = False + setup_deadline_car(my_predbat) + windows = [{"start": my_predbat.minutes_now + 30 * n, "end": my_predbat.minutes_now + 30 * (n + 1), "average": 5.0 if n < 4 else 30.0} for n in range(10)] + update_rates_import(my_predbat, windows) + mark_away(my_predbat, 0, 120) + + deadline = set_deadline(my_predbat, 300, 60) + plan = my_predbat.plan_car_charging(0, []) + if overlaps_away(my_predbat, plan): + print("ERROR: charge planned while the car is away: {}".format(overlaps_away(my_predbat, plan))) + failed = True + if bought_by(plan, deadline) < 19.9: + print("ERROR: the six present slots before the deadline can deliver the 20kWh, got {} from {}".format(bought_by(plan, deadline), plan)) + failed = True + return failed + + +def test_deadline_ready_for_a_trip(my_predbat): + """Ready by the time you leave, away from then on: the whole promise lands before departure. + + The usual shape of a trip. The sun after departure is no use to a car that is not there, so none of the + promise may be left to it and nothing may be planned after the deadline at all. + """ + print(" - test_deadline_ready_for_a_trip") + failed = False + setup_deadline_car(my_predbat) + my_predbat.car_charging_solar = True + set_pv(my_predbat, 7.0, start_offset=300, length=300) + mark_away(my_predbat, 240, 720) + + deadline = set_deadline(my_predbat, 240, 60) + plan = my_predbat.plan_car_charging(0, []) + if bought_by(plan, deadline) < 19.9: + print("ERROR: the car leaves at the deadline, so the promise has to be bought before it, got {}".format(plan)) + failed = True + if overlaps_away(my_predbat, plan): + print("ERROR: the sun after departure was planned into a car that is away: {}".format(overlaps_away(my_predbat, plan))) + failed = True + return failed + + +def test_deadline_does_not_claim_sun_while_away(my_predbat): + """The battery hold only releases sun the car is present for; sun while it is away stays with the battery.""" + print(" - test_deadline_does_not_claim_sun_while_away") + failed = False + setup_deadline_car(my_predbat) + my_predbat.car_charging_solar = True + my_predbat.soc_max = 10.0 + my_predbat.soc_kw = 0.0 + my_predbat.battery_rate_max_charge = 5.0 / 60.0 + my_predbat.battery_rate_max_scaling = 1.0 + # Six sunny slots before the deadline; the car is away for the first four of them + set_pv(my_predbat, 7.0, start_offset=60, length=180) + mark_away(my_predbat, 60, 180) + load_step = my_predbat.car_solar_load_forecast() + + deadline = set_deadline(my_predbat, 300, 60) + reserved = my_predbat.car_solar_reserved_for_car(load_step) + if abs(reserved - 7.0) > 0.1: + print("ERROR: only the two present sunny slots (7kWh) can be claimed for the car, got {}".format(reserved)) + failed = True + + plan = my_predbat.plan_car_charging(0, []) + if overlaps_away(my_predbat, plan): + print("ERROR: charge planned while the car is away: {}".format(overlaps_away(my_predbat, plan))) + failed = True + if bought_by(plan, deadline) + solar_by(plan, deadline) < 19.9: + print("ERROR: sun and purchases together should still meet the deadline, got {}".format(plan)) + failed = True + return failed + + +def test_deadline_warns_when_away_leaves_too_little(my_predbat): + """A promise that cannot be kept is said out loud, and what can be delivered still is.""" + print(" - test_deadline_warns_when_away_leaves_too_little") + failed = False + setup_deadline_car(my_predbat) + # Away for six of the eight slots before the deadline: two slots at 3.7kWh cannot deliver 20kWh + mark_away(my_predbat, 0, 180) + deadline = set_deadline(my_predbat, 240, 60) + + messages = [] + original_log = my_predbat.log + my_predbat.log = lambda message, *args, **kwargs: messages.append(message) + try: + plan = my_predbat.plan_car_charging(0, []) + finally: + my_predbat.log = original_log + + if not any("can only reach" in message for message in messages): + print("ERROR: an unreachable deadline should be warned about, logged {}".format(messages)) + failed = True + if abs(bought_by(plan, deadline) - 7.4) > 0.1: + print("ERROR: the two present slots should still be bought in full, got {} from {}".format(bought_by(plan, deadline), plan)) + failed = True + return failed + + +def test_deadline_applies_to_the_first_car_only(my_predbat): + """The level is a percentage of one car's battery, so it is not applied to every car.""" + print(" - test_deadline_applies_to_the_first_car_only") + failed = False + setup_deadline_car(my_predbat) + my_predbat.num_cars = 2 + my_predbat.car_charging_soc = [10.0, 10.0] + my_predbat.car_charging_limit = [40.0, 40.0] + my_predbat.car_charging_battery_size = [50.0, 50.0] + set_deadline(my_predbat, 240, 60) + if not my_predbat.car_one_off_deadlines(0): + print("ERROR: the first car should carry the deadline") + failed = True + if my_predbat.car_one_off_deadlines(1): + print("ERROR: the second car should not carry the first car's deadline") + failed = True + return failed + + +def test_deadline_survives_an_unset_config_value(my_predbat): + """A gated valued select reads back None before it is enabled, and must decode to nothing. + + manual_car_deadline is gated on num_cars. manual_rates() called .replace() on the stored value + directly - the same shape that took headless startup down through manual_times() for manual_car_away. + """ + print(" - test_deadline_survives_an_unset_config_value") + failed = False + item = my_predbat.config_index.get("manual_car_deadline") + if item is None: + print("ERROR: manual_car_deadline is not registered") + return True + saved = item.get("value", "") + try: + item["value"] = None + try: + result = my_predbat.manual_rates("manual_car_deadline", update=False) + except Exception as e: + print("ERROR: an unset value raised {}: {}".format(type(e).__name__, e)) + return True + if result: + print("ERROR: an unset value should decode to no deadline, got {}".format(result)) + failed = True + finally: + item["value"] = saved + return failed + + +def test_deadline_select_takes_its_own_default(my_predbat): + """A slot picked without a level takes manual_car_deadline_value, and an explicit level is kept. + + Valued selects are routed on a substring of their name, and an unrecognised one is dropped with a + warning - so without its own branch the deadline could not be set at all. + """ + print(" - test_deadline_select_takes_its_own_default") + failed = False + item = my_predbat.config_index.get("manual_car_deadline") + saved = item.get("value", "") + slot = int(my_predbat.minutes_now / my_predbat.plan_interval_minutes) * my_predbat.plan_interval_minutes + 120 + label = (my_predbat.midnight_utc + timedelta(minutes=slot)).strftime("%a %H:%M") + default = my_predbat.get_arg("manual_car_deadline_value") + try: + my_predbat.manual_select("manual_car_deadline", "off") + my_predbat.manual_select("manual_car_deadline", label) + decoded = my_predbat.manual_rates("manual_car_deadline", default_rate=default, update=False) + if decoded.get(slot) != default: + print("ERROR: a slot with no level should take the deadline default {}, got {}".format(default, decoded.get(slot))) + failed = True + + my_predbat.manual_select("manual_car_deadline", "off") + my_predbat.manual_select("manual_car_deadline", "{}=70".format(label)) + decoded = my_predbat.manual_rates("manual_car_deadline", default_rate=default, update=False) + if decoded.get(slot) != 70: + print("ERROR: an explicit level of 70 should be kept, got {}".format(decoded.get(slot))) + failed = True + finally: + my_predbat.manual_select("manual_car_deadline", "off") + item["value"] = saved + return failed + + +def test_deadline_is_reachable_from_the_plan_ui(my_predbat): + """The plan's time cells must offer the deadline, send it, show it, and clear it. + + Car Away first shipped with its config and planner logic but no way to reach it from the plan, which + is where it is useful. Reads the sources rather than driving a browser, as the web tests around it do. + """ + print(" - test_deadline_is_reachable_from_the_plan_ui") + import os + + failed = False + here = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + helper = open(os.path.join(here, "web_helper.py")).read() + web = open(os.path.join(here, "web.py")).read() + + if "handleCarDeadline(" not in helper: + print("ERROR: the plan time-cell dropdown does not offer a car ready-by action") + failed = True + if "overrides.manual_car_deadline" not in helper: + print("ERROR: the plan renderer never reads the deadline, so a set one cannot show as set") + failed = True + if "override-car-deadline" not in helper: + print("ERROR: a slot carrying the deadline is not highlighted") + failed = True + for action in ("'Set Car Deadline'", "'Clear Car Deadline'"): + if action not in helper: + print("ERROR: the plan page never sends {}".format(action)) + failed = True + if 'action == "Set Car Deadline"' not in web or 'action == "Clear Car Deadline"' not in web: + print("ERROR: the rate_override handler does not accept the car deadline actions") + failed = True + if web.count('"manual_car_deadline": manual_car_deadline_list') < 2: + print("ERROR: both plan payloads must carry the deadline or one of the two views cannot show it") + failed = True + + # Only one deadline at a time: setting a new one has to clear the old before adding + set_block = web.split('action == "Set Car Deadline":')[1].split("elif action")[0] if 'action == "Set Car Deadline":' in web else "" + if 'async_manual_select("manual_car_deadline", "off")' not in set_block: + print("ERROR: setting a deadline must replace any existing one rather than add a second") + failed = True + return failed + + +def run_car_deadline_tests(my_predbat): + """Run every manual_car_deadline test. + + The car, battery, PV and load settings live on the shared my_predbat instance, so everything these + tests change is snapshotted and put back - leaking a deadline or a lowered minimum into a later test + only shows up in some orderings (#5079). + """ + print("**** Running car deadline tests ****\n") + carried = ( + "minutes_now", + "load_forecast_only", + "load_forecast", + "load_scaling", + "load_inday_adjustment", + "load_scaling_dynamic", + "manual_load_adjust", + "metric_load_divergence", + "dynamic_load_baseline", + "pv_forecast_minute", + "num_cars", + "car_charging_soc", + "car_charging_limit", + "car_charging_battery_size", + "car_charging_rate", + "car_charging_loss", + "car_charging_slots", + "car_charging_plan_time", + "car_charging_plan_smart", + "car_charging_plan_max_price", + "car_charging_now", + "car_charging_solar", + "car_charging_solar_excess", + "car_charging_rate_threshold_export", + "car_charging_plan_min_soc", + "manual_car_away_times", + "manual_car_deadline_keep", + "battery_rate_max_charge", + "battery_rate_max_scaling", + "soc_kw", + "soc_max", + ) + saved = {name: getattr(my_predbat, name, None) for name in carried} + try: + failed = test_deadline_buys_before_it(my_predbat) + failed |= test_deadline_ignores_the_price_cap(my_predbat) + failed |= test_deadline_buys_outside_the_cheap_band(my_predbat) + failed |= test_deadline_buys_the_cheapest_slots(my_predbat) + failed |= test_deadline_uses_sun_first(my_predbat) + failed |= test_deadline_does_not_count_later_sun(my_predbat) + failed |= test_deadline_is_capped_at_the_limit(my_predbat) + failed |= test_both_deadlines_are_kept(my_predbat) + failed |= test_deadline_claims_sun_ahead_of_the_battery(my_predbat) + failed |= test_deadline_buys_around_away_time(my_predbat) + failed |= test_deadline_ready_for_a_trip(my_predbat) + failed |= test_deadline_does_not_claim_sun_while_away(my_predbat) + failed |= test_deadline_warns_when_away_leaves_too_little(my_predbat) + failed |= test_deadline_applies_to_the_first_car_only(my_predbat) + failed |= test_deadline_survives_an_unset_config_value(my_predbat) + failed |= test_deadline_select_takes_its_own_default(my_predbat) + failed |= test_deadline_is_reachable_from_the_plan_ui(my_predbat) + finally: + for name, value in saved.items(): + setattr(my_predbat, name, value) + return failed diff --git a/apps/predbat/tests/test_car_solar.py b/apps/predbat/tests/test_car_solar.py new file mode 100644 index 000000000..ca20e4afc --- /dev/null +++ b/apps/predbat/tests/test_car_solar.py @@ -0,0 +1,751 @@ +# ----------------------------------------------------------------------------- +# Predbat Home Battery System +# Copyright Trefor Southwell 2026 - All Rights Reserved +# This application maybe used for personal use only and not for commercial use +# ----------------------------------------------------------------------------- +# fmt off +# pylint: disable=consider-using-f-string +# pylint: disable=line-too-long +# pylint: disable=attribute-defined-outside-init + +"""Tests for solar-aware car charging and the battery-versus-export trade-off. + +Two separate things are covered here. car_charging_solar places car slots on forecast sunshine so the +price-based pass only has to cover the shortfall. The trade-off between exporting the battery and +saving it for the car needs no new machinery - it already falls out of the cost model whenever +car_charging_from_battery is on - so the tests here pin that behaviour rather than add to it. + +Solar slots are sized from the forecast surplus (PV less house load) rather than the charger's rated +power, because a charge-on-solar charger modulates to whatever is spare. set_house_load below is what +makes that testable: it drives the load forecast directly instead of through the fixture's history. +""" + +from const import PREDICT_STEP, EXPORT_MODE_TARGET +from utils import dp2, export_mode_of +from tests.test_infra import reset_inverter, reset_rates, update_rates_import, update_rates_export +from prediction import Prediction + + +def ready_time_str(my_predbat, minutes_ahead): + """Return an HH:MM:SS ready time the given number of minutes after the test clock. + + The ready time is absolute wall-clock in the config, so a fixed string like "23:30:00" makes a test + pass or fail depending on the hour it runs at. Deriving it from minutes_now keeps these deterministic. + """ + target = (my_predbat.minutes_now + minutes_ahead) % (24 * 60) + return "{:02d}:{:02d}:00".format(target // 60, target % 60) + + +def set_house_load(my_predbat, kw): + """Force the house load forecast to a flat kw, whatever history the shared fixture is carrying. + + step_data_history normally builds load from days_previous, which no test can predict. Turning on + load_forecast_only zeroes that historical term, and dynamic_load_baseline then sets each bucket's + floor - so with everything else neutralised each bucket comes out at exactly the requested power. + """ + my_predbat.load_forecast_only = True + my_predbat.load_forecast = {} + my_predbat.load_scaling = 1.0 + my_predbat.load_inday_adjustment = 1.0 + my_predbat.load_scaling_dynamic = {} + my_predbat.manual_load_adjust = {} + my_predbat.metric_load_divergence = None + kwh_per_step = kw * PREDICT_STEP / 60.0 + my_predbat.dynamic_load_baseline = {my_predbat.minutes_now + offset: kwh_per_step for offset in range(0, my_predbat.forecast_minutes + my_predbat.plan_interval_minutes, PREDICT_STEP)} + + +def setup_car(my_predbat, car_kwh=8.0, ready_ahead=720, rate=7.4, house_kw=0.0): + """Configure a single car needing car_kwh within ready_ahead minutes, and clear any existing plan. + + minutes_now is snapped down onto the plan-interval grid first. Solar windows sit on that grid while + set_pv places the sunny block at an offset from minutes_now, so unless the two are aligned the first + and last windows only partly overlap the sun and legitimately carry less than the full surplus - which + made the per-slot power assertions depend on the wall-clock minute the suite happened to start at. + """ + my_predbat.minutes_now = int(my_predbat.minutes_now / my_predbat.plan_interval_minutes) * my_predbat.plan_interval_minutes + set_house_load(my_predbat, house_kw) + my_predbat.num_cars = 1 + my_predbat.car_charging_soc = [0.0] + my_predbat.car_charging_limit = [car_kwh] + my_predbat.car_charging_battery_size = [50.0] + my_predbat.car_charging_rate = [rate] + my_predbat.car_charging_loss = 1.0 + my_predbat.car_charging_slots = [[]] + my_predbat.car_charging_plan_time = [ready_time_str(my_predbat, ready_ahead)] + my_predbat.car_charging_plan_smart = [True] + my_predbat.car_charging_plan_max_price = [0.0] + my_predbat.car_charging_now = [False] + my_predbat.car_charging_solar = False + my_predbat.car_charging_solar_excess = 1.0 + my_predbat.car_charging_rate_threshold_export = 99 + my_predbat.car_charging_plan_min_soc = 100 + # A full house battery: surplus goes to the pack first, and these tests are about how the car's share is + # sized rather than who gets it. The ones about the hold set the battery themselves. + my_predbat.soc_max = 10.0 + my_predbat.soc_kw = 10.0 + + +def set_pv(my_predbat, midday_kw, start_offset=240, length=240): + """Publish a flat block of forecast PV of midday_kw starting start_offset minutes from now.""" + my_predbat.pv_forecast_minute = {} + for minute in range(my_predbat.minutes_now, my_predbat.minutes_now + my_predbat.forecast_minutes): + in_sun = start_offset <= (minute - my_predbat.minutes_now) < (start_offset + length) + my_predbat.pv_forecast_minute[minute] = (midday_kw / 60.0) if in_sun else 0.0 + + +def build_low_rates(my_predbat, count=48): + """Build half-hour import windows, cheap overnight and expensive by day.""" + low_rates = [] + for n in range(0, count): + price = 5.0 if (n % 24) > 12 else 30.0 + low_rates.append({"start": my_predbat.minutes_now + 30 * n, "end": my_predbat.minutes_now + 30 * (n + 1), "average": price}) + return low_rates + + +def test_solar_windows_selection(my_predbat): + """plan_car_charging_solar_windows gates on forecast power and on the export rate.""" + print(" - test_solar_windows_selection") + failed = False + setup_car(my_predbat) + reset_rates(my_predbat, 30.0, 5.0) + + # Off by default: no solar windows at all, whatever the sunshine + set_pv(my_predbat, 5.0) + if my_predbat.plan_car_charging_solar_windows(): + print("ERROR: no solar windows should be produced while car_charging_solar is off") + failed = True + + # On, with plenty of sun and a low export rate + my_predbat.car_charging_solar = True + windows = my_predbat.plan_car_charging_solar_windows() + if not windows: + print("ERROR: expected solar windows with 5kW of forecast PV") + failed = True + for window in windows: + if not window.get("solar"): + print("ERROR: solar windows must be marked solar=True") + failed = True + # Windows sit on the plan-interval grid, so one may straddle the edge of the sunny block; + # what matters is that every window overlaps it rather than starting inside it + start_offset = window["start"] - my_predbat.minutes_now + end_offset = window["end"] - my_predbat.minutes_now + if not (start_offset < 480 and end_offset > 240): + print("ERROR: solar window at offset {}-{} does not overlap the sunny block".format(start_offset, end_offset)) + failed = True + + # Sun below the excess threshold produces nothing + set_pv(my_predbat, 0.5) + if my_predbat.plan_car_charging_solar_windows(): + print("ERROR: 0.5kW is below the 1.0kW threshold and should produce no windows") + failed = True + + # Plenty of sun, but export pays better than the threshold allows + set_pv(my_predbat, 5.0) + reset_rates(my_predbat, 30.0, 40.0) + my_predbat.car_charging_rate_threshold_export = 20 + if my_predbat.plan_car_charging_solar_windows(): + print("ERROR: an export rate of 40 above the threshold of 20 should suppress solar charging") + failed = True + + return failed + + +def test_solar_reduces_paid_import(my_predbat): + """With solar available the price pass only tops up the shortfall, so less energy is bought.""" + print(" - test_solar_reduces_paid_import") + failed = False + setup_car(my_predbat, car_kwh=8.0, ready_ahead=720) + reset_rates(my_predbat, 30.0, 5.0) + low_rates = build_low_rates(my_predbat) + update_rates_import(my_predbat, low_rates) + set_pv(my_predbat, 7.0) + + # Solar off: the whole 8kWh is bought + plan_off = my_predbat.plan_car_charging(0, low_rates) + bought_off = sum(slot["kwh"] for slot in plan_off if not slot.get("solar")) + if any(slot.get("solar") for slot in plan_off): + print("ERROR: no slot should be marked solar while car_charging_solar is off") + failed = True + + # Solar on: sunshine covers part of it, so strictly less has to be bought + my_predbat.car_charging_solar = True + plan_on = my_predbat.plan_car_charging(0, low_rates) + solar_kwh = sum(slot["kwh"] for slot in plan_on if slot.get("solar")) + bought_on = sum(slot["kwh"] for slot in plan_on if not slot.get("solar")) + + if solar_kwh <= 0: + print("ERROR: expected some of the car charge to come from solar, plan {}".format(plan_on)) + failed = True + if bought_on >= (bought_off - 0.1): + print("ERROR: solar should reduce paid import, bought off={} on={}".format(bought_off, bought_on)) + failed = True + + # Both plans must still meet the car's requirement + for name, plan in (("off", plan_off), ("on", plan_on)): + total = sum(slot["kwh"] for slot in plan) + if total < (my_predbat.car_charging_limit[0] - 0.2): + print("ERROR: plan {} only delivers {} of {}kWh".format(name, total, my_predbat.car_charging_limit[0])) + failed = True + + return failed + + +def test_solar_slots_do_not_overlap(my_predbat): + """A solar window and a cheap-import window covering the same time yield only one slot.""" + print(" - test_solar_slots_do_not_overlap") + failed = False + setup_car(my_predbat, car_kwh=20.0, ready_ahead=720) + reset_rates(my_predbat, 5.0, 1.0) + low_rates = build_low_rates(my_predbat) + update_rates_import(my_predbat, low_rates) + set_pv(my_predbat, 7.0) + my_predbat.car_charging_solar = True + + plan = my_predbat.plan_car_charging(0, low_rates) + if not any(slot.get("solar") for slot in plan): + print("ERROR: expected at least one solar slot in the plan to make this test meaningful") + failed = True + for first in range(len(plan)): + for second in range(first + 1, len(plan)): + if (plan[first]["start"] < plan[second]["end"]) and (plan[first]["end"] > plan[second]["start"]): + print("ERROR: overlapping car slots {} and {}".format(plan[first], plan[second])) + failed = True + return failed + + +def test_min_soc_splits_bought_from_solar(my_predbat): + """Bought slots stop at the guaranteed minimum; solar carries on to the full limit.""" + print(" - test_min_soc_splits_bought_from_solar") + failed = False + # 71kWh car, limit 80% (56.8kWh), minimum 30% (21.3kWh) - the Tesla sun-slider arrangement + setup_car(my_predbat, car_kwh=56.8, ready_ahead=720) + my_predbat.car_charging_battery_size = [71.0] + reset_rates(my_predbat, 30.0, 5.0) + low_rates = build_low_rates(my_predbat) + update_rates_import(my_predbat, low_rates) + set_pv(my_predbat, 7.0) + my_predbat.car_charging_solar = True + + # Default (100%): everything solar does not cover is bought, up to the full limit + plan_full = my_predbat.plan_car_charging(0, low_rates) + bought_full = sum(slot["kwh"] for slot in plan_full if not slot.get("solar")) + + # With a 30% minimum, bought energy stops there and solar keeps going + my_predbat.car_charging_plan_min_soc = 30 + plan_split = my_predbat.plan_car_charging(0, low_rates) + bought_split = sum(slot["kwh"] for slot in plan_split if not slot.get("solar")) + solar_split = sum(slot["kwh"] for slot in plan_split if slot.get("solar")) + + if bought_split >= bought_full: + print("ERROR: a 30% minimum should buy less ({}) than the default ({})".format(bought_split, bought_full)) + failed = True + # Solar already delivers past 30%, so nothing should need buying here + if solar_split > 21.3 and bought_split > 0.1: + print("ERROR: solar covers the minimum, so nothing should be bought, got {}".format(bought_split)) + failed = True + if solar_split <= 21.3: + print("ERROR: expected solar to carry past the 21.3kWh minimum, got {}".format(solar_split)) + failed = True + + # No sun: the minimum must still be met from the grid, and no more + set_pv(my_predbat, 0.0) + plan_dark = my_predbat.plan_car_charging(0, low_rates) + bought_dark = sum(slot["kwh"] for slot in plan_dark if not slot.get("solar")) + if any(slot.get("solar") for slot in plan_dark): + print("ERROR: no solar slots should be planned with no sun") + failed = True + if not (20.0 <= bought_dark <= 23.0): + print("ERROR: with no sun expected roughly the 21.3kWh minimum bought, got {}".format(bought_dark)) + failed = True + + return failed + + +def solar_kw_in_slot(slot): + """Average power a planned slot draws, in kW.""" + return slot["kwh"] * 60.0 / (slot["end"] - slot["start"]) + + +def test_solar_slot_size_follows_surplus(my_predbat): + """A solar slot is sized from PV minus house load, not from the charger's rated power.""" + print(" - test_solar_slot_size_follows_surplus") + failed = False + # 7kW of sun against a 3kW house leaves 4kW for a 7kW charger + setup_car(my_predbat, car_kwh=50.0, ready_ahead=720, rate=7.0, house_kw=3.0) + reset_rates(my_predbat, 30.0, 5.0) + set_pv(my_predbat, 7.0) + my_predbat.car_charging_solar = True + + plan = my_predbat.plan_car_charging(0, []) + solar_slots = [slot for slot in plan if slot.get("solar")] + if not solar_slots: + print("ERROR: expected solar slots with 7kW of sun against a 3kW house") + return True + + for slot in solar_slots: + power = solar_kw_in_slot(slot) + if abs(power - 4.0) > 0.3: + print("ERROR: slot should draw the 4kW surplus, got {}kW ({})".format(dp2(power), slot)) + failed = True + + # The old behaviour was to assume the full charger rate regardless, so pin that it is gone + if any(solar_kw_in_slot(slot) > 6.0 for slot in solar_slots): + print("ERROR: a solar slot is still being sized at the charger's rated power") + failed = True + + # Raising the house load must reduce what the car is predicted to take, from the same sunshine + total_at_3kw = sum(slot["kwh"] for slot in solar_slots) + set_house_load(my_predbat, 5.0) + total_at_5kw = sum(slot["kwh"] for slot in my_predbat.plan_car_charging(0, []) if slot.get("solar")) + if not (total_at_5kw < total_at_3kw): + print("ERROR: a bigger house load should leave the car less, got {} vs {}".format(total_at_5kw, total_at_3kw)) + failed = True + + return failed + + +def test_solar_slot_capped_by_charger(my_predbat): + """Surplus beyond what the charger can take is not credited to the car.""" + print(" - test_solar_slot_capped_by_charger") + failed = False + # 12kW of sun and a 1kW house leaves 11kW spare, but the charger tops out at 7kW + setup_car(my_predbat, car_kwh=50.0, ready_ahead=720, rate=7.0, house_kw=1.0) + reset_rates(my_predbat, 30.0, 5.0) + set_pv(my_predbat, 12.0) + my_predbat.car_charging_solar = True + + plan = my_predbat.plan_car_charging(0, []) + solar_slots = [slot for slot in plan if slot.get("solar")] + if not solar_slots: + print("ERROR: expected solar slots with 12kW of sun") + return True + for slot in solar_slots: + power = solar_kw_in_slot(slot) + if power > 7.05: + print("ERROR: slot draws {}kW, above the 7kW charger limit ({})".format(dp2(power), slot)) + failed = True + if abs(power - 7.0) > 0.3: + print("ERROR: with 11kW spare the charger should run flat out at 7kW, got {}kW".format(dp2(power))) + failed = True + return failed + + +def test_solar_window_needs_real_surplus(my_predbat): + """The excess threshold is tested against surplus, so sun the house is already eating does not qualify.""" + print(" - test_solar_window_needs_real_surplus") + failed = False + setup_car(my_predbat, rate=7.0, house_kw=0.0) + reset_rates(my_predbat, 30.0, 5.0) + my_predbat.car_charging_solar = True + my_predbat.car_charging_solar_excess = 1.0 + set_pv(my_predbat, 4.0) + + # Sanity: with no house load 4kW of sun is 4kW of surplus and qualifies + if not my_predbat.plan_car_charging_solar_windows(): + print("ERROR: 4kW of sun against an idle house should qualify") + failed = True + + # The same 4kW of sun against a 4kW house leaves nothing, and must not qualify + set_house_load(my_predbat, 4.0) + windows = my_predbat.plan_car_charging_solar_windows() + if windows: + print("ERROR: 4kW of sun fully consumed by a 4kW house should produce no windows, got {}".format(windows[:3])) + failed = True + + # A house drawing 3.5kW leaves 0.5kW, still under the 1kW threshold + set_house_load(my_predbat, 3.5) + if my_predbat.plan_car_charging_solar_windows(): + print("ERROR: a 0.5kW surplus is below the 1kW threshold and should produce no windows") + failed = True + + # Drop the house to 2kW and the remaining 2kW surplus clears it + set_house_load(my_predbat, 2.0) + if not my_predbat.plan_car_charging_solar_windows(): + print("ERROR: a 2kW surplus should clear the 1kW threshold") + failed = True + return failed + + +def test_solar_surplus_floored_per_bucket(my_predbat): + """Surplus is floored at zero per bucket, so a dark half slot cannot cancel a sunny one.""" + print(" - test_solar_surplus_floored_per_bucket") + failed = False + setup_car(my_predbat, rate=7.0, house_kw=2.0) + reset_rates(my_predbat, 30.0, 5.0) + + # An hour of sun starting two hours out, against a house that draws 2kW around the clock + set_pv(my_predbat, 6.0, start_offset=120, length=60) + load_step = my_predbat.car_solar_load_forecast() + + now = my_predbat.minutes_now + # The sunny hour on its own: 6kW of sun less 2kW of house is 4kW, so 4kWh over the hour + sunny = my_predbat.car_solar_surplus_kwh(now + 120, now + 180, load_step) + if abs(sunny - 4.0) > 0.2: + print("ERROR: expected 4kWh of surplus over the sunny hour, got {}".format(dp2(sunny))) + failed = True + + # A dark hour is a 2kW deficit, but a deficit is not a negative surplus - it is simply nothing + dark = my_predbat.car_solar_surplus_kwh(now + 240, now + 300, load_step) + if abs(dark) > 0.01: + print("ERROR: a dark hour should yield no surplus, got {}".format(dp2(dark))) + failed = True + + # Spanning both, the dark half must not eat into the sunny half's 4kWh + spanning = my_predbat.car_solar_surplus_kwh(now + 120, now + 300, load_step) + if abs(spanning - 4.0) > 0.2: + print("ERROR: the dark hours should not cancel the sunny one, expected 4kWh got {}".format(dp2(spanning))) + failed = True + return failed + + +def bought_starts(plan): + """Offsets from now of the grid slots in a car plan.""" + return [slot["start"] for slot in plan if not slot.get("solar")] + + +def test_bought_slots_prefer_no_sun(my_predbat): + """Among equally priced grid slots the car takes the sunless ones: in the sun it would eat the battery's solar. + + Reported from a live system: off-peak cost the same at 14:00 as at midnight, ties went to the earliest slot, + and the car charged at 14:00 in full sun - load the inverter fed from the surplus before the battery. + """ + print(" - test_bought_slots_prefer_no_sun") + failed = False + setup_car(my_predbat, car_kwh=3.5, ready_ahead=1200, rate=7.0, house_kw=1.0) + reset_rates(my_predbat, 34.0, 5.0) + # Solar windows stay closed (export pays more than the threshold), so only the grid ordering is under test + my_predbat.car_charging_rate_threshold_export = 1 + set_pv(my_predbat, 7.0, start_offset=0, length=240) + low_rates = [{"start": my_predbat.minutes_now + 30 * n, "end": my_predbat.minutes_now + 30 * (n + 1), "average": 34.0} for n in range(30)] + sun_end = my_predbat.minutes_now + 240 + + # Without solar car charging nothing changes: the tie still goes to the earliest slot, in the sun + plain = bought_starts(my_predbat.plan_car_charging(0, low_rates)) + if not plain or plain[0] >= sun_end: + print("ERROR: with solar car charging off the earliest (sunny) slot should still win, so this test proves nothing: {}".format(plain)) + return True + + my_predbat.car_charging_solar = True + starts = bought_starts(my_predbat.plan_car_charging(0, low_rates)) + my_predbat.car_charging_solar = False + if not starts or any(start < sun_end for start in starts): + print("ERROR: at the same price the car should charge after the sun, got starts {} with sun until {}".format(starts, sun_end)) + failed = True + return failed + + +def test_solar_battery_takes_surplus_first(my_predbat): + """The house battery gets surplus solar first; the car is offered windows once it is predicted full. + + A kWh in the pack displaces the evening peak, while one in a car with nothing promised displaces at most + a cheap overnight top-up. The fullness is walked forward slot by slot as the held surplus fills the pack, + so a battery that gets there mid-morning releases the car mid-morning rather than holding all day. + """ + print(" - test_solar_battery_takes_surplus_first") + failed = False + setup_car(my_predbat, rate=7.0, house_kw=1.0) + reset_rates(my_predbat, 30.0, 5.0) + my_predbat.car_charging_solar = True + my_predbat.battery_rate_max_charge = 2.0 / 30.0 # 2kWh per 30 minute slot, pinned for determinism + my_predbat.battery_rate_max_scaling = 1.0 + my_predbat.soc_max = 10.0 + set_pv(my_predbat, 7.0, start_offset=240, length=240) + + # Full battery: nothing to bank, so every sunny slot goes to the car + my_predbat.soc_kw = 10.0 + full = my_predbat.plan_car_charging_solar_windows() + if len(full) != 8: + print("ERROR: a full battery should release all 8 sunny slots, got {}".format(len(full))) + return True + + # Empty battery at 2kWh a slot: the first 5 sunny slots fill it, the last 3 go to the car + my_predbat.soc_kw = 0.0 + empty = my_predbat.plan_car_charging_solar_windows() + if len(empty) != 3: + print("ERROR: an empty 10kWh pack filling at 2kWh a slot should hold 5 of 8 slots, leaving 3, got {}".format(len(empty))) + failed = True + if empty and empty[0]["start"] != full[5]["start"]: + print("ERROR: the car should be released once the pack is full, from the sixth sunny slot, got {}".format(empty[0]["start"])) + failed = True + + # Half full: fewer slots held + my_predbat.soc_kw = 5.0 + half = my_predbat.plan_car_charging_solar_windows() + if len(half) != 5: + print("ERROR: a half-full pack should hold 3 slots, leaving 5, got {}".format(len(half))) + failed = True + return failed + + +def test_away_moves_solar_earlier(my_predbat): + """Away time makes the battery-priority hold yield, so the car charges while it is still here. + + The hold assumes the car can catch up later, which stops being true the moment the afternoon is + marked away - and with the battery taking surplus first the car would otherwise never see a solar + window at all. Reported from a live system: marking the afternoon away made the car charge from the grid + at 30p instead of moving to the morning sun. + """ + print(" - test_away_moves_solar_earlier") + failed = False + setup_car(my_predbat, car_kwh=30.0, ready_ahead=1200, rate=7.0, house_kw=1.0) + reset_rates(my_predbat, 30.0, 5.0) + my_predbat.car_charging_solar = True + my_predbat.soc_max = 27.0 + my_predbat.soc_kw = 0.0 + my_predbat.battery_rate_max_charge = 10.0 / 60.0 + my_predbat.battery_rate_max_scaling = 1.0 + set_pv(my_predbat, 8.0, start_offset=120, length=480) + low_rates = [{"start": my_predbat.minutes_now + 30 * n, "end": my_predbat.minutes_now + 30 * (n + 1), "average": 30.0} for n in range(40)] + update_rates_import(my_predbat, low_rates) + + my_predbat.manual_car_away_times = [] + baseline = [slot for slot in my_predbat.plan_car_charging(0, low_rates) if slot.get("solar")] + if not baseline: + print("ERROR: expected some solar charging with no away markers") + return True + baseline_first = baseline[0]["start"] - my_predbat.minutes_now + + # Mark the afternoon away - the car's remaining chances are all in the morning + my_predbat.manual_car_away_times = [my_predbat.minutes_now + m for m in range(300, 600, 30)] + plan = my_predbat.plan_car_charging(0, low_rates) + solar = [slot for slot in plan if slot.get("solar")] + + if not solar: + print("ERROR: away time removed solar charging entirely instead of moving it earlier") + failed = True + else: + moved_to = solar[0]["start"] - my_predbat.minutes_now + if moved_to >= baseline_first: + print("ERROR: solar should start earlier than {} once the afternoon is away, got {}".format(baseline_first, moved_to)) + failed = True + + for slot in plan: + if my_predbat.car_slot_is_away(slot["start"], slot["end"]): + print("ERROR: planned a slot the car is away for: {}-{}".format(slot["start"], slot["end"])) + failed = True + + # With no away time set the hold is untouched, which is the everyday case + my_predbat.manual_car_away_times = [] + if my_predbat.car_solar_reserved_for_car(my_predbat.car_solar_load_forecast()) != 0.0: + print("ERROR: nothing should be reserved for the car when no away time is set") + failed = True + + return failed + + +def test_solar_windows_ignore_ready_time(my_predbat): + """Solar windows run to the forecast horizon, so a morning ready time does not exclude daylight.""" + print(" - test_solar_windows_ignore_ready_time") + failed = False + setup_car(my_predbat) + reset_rates(my_predbat, 30.0, 5.0) + my_predbat.car_charging_solar = True + # Sun starts 4 hours out; a ready time only 1 hour out must not suppress it + set_pv(my_predbat, 5.0, start_offset=240) + + windows = my_predbat.plan_car_charging_solar_windows() + if not windows: + print("ERROR: solar windows should be offered regardless of the ready time") + failed = True + if not any((window["start"] - my_predbat.minutes_now) >= 240 for window in windows): + print("ERROR: expected solar windows beyond the sunny block start, got {}".format(windows[:3])) + failed = True + return failed + + +def run_car_export_tradeoff(my_predbat, export_rate, from_battery, car_kwh=8.0): + """Plan a battery against a fixed evening car slot and one export window at the given rate.""" + end_record = my_predbat.forecast_minutes + my_predbat.calculate_best_charge = True + my_predbat.calculate_best_export = True + my_predbat.soc_max = 20.0 + my_predbat.soc_kw = 18.0 + my_predbat.reserve = 0.5 + my_predbat.set_charge_freeze = True + my_predbat.set_export_freeze = True + my_predbat.best_soc_keep = 0.0 + my_predbat.debug_enable = False + my_predbat.battery_charging_from_grid = True + my_predbat.car_charging_from_battery = from_battery + my_predbat.car_energy_reported_load = True + my_predbat.num_cars = 1 + my_predbat.car_charging_soc = [0.0] + my_predbat.car_charging_limit = [car_kwh] + my_predbat.car_charging_battery_size = [50.0] + my_predbat.car_charging_loss = 1.0 + + now = my_predbat.minutes_now + my_predbat.car_charging_slots = [[{"start": now + 120, "end": now + 240, "kwh": car_kwh, "average": 30.0, "octopus": False}]] + + charge_window_best = [{"start": now + 30 * n, "end": now + 30 * (n + 1), "average": 30.0} for n in range(0, 48)] + export_window_best = [{"start": now, "end": now + 60, "average": export_rate}] + + reset_rates(my_predbat, 30.0, export_rate) + update_rates_import(my_predbat, charge_window_best) + update_rates_export(my_predbat, export_window_best) + + pv_step = {minute: 0.0 for minute in range(0, my_predbat.forecast_minutes, 5)} + load_step = {minute: 0.2 / (60 / 5) for minute in range(0, my_predbat.forecast_minutes, 5)} + my_predbat.load_minutes_step = load_step + my_predbat.load_minutes_step10 = load_step + my_predbat.pv_forecast_minute_step = pv_step + my_predbat.pv_forecast_minute10_step = pv_step + my_predbat.prediction = Prediction(my_predbat, pv_step, pv_step, load_step, load_step) + + charge_limit_best = [0 for _ in range(len(charge_window_best))] + export_limits_best = [100 for _ in range(len(export_window_best))] + result = my_predbat.run_prediction(charge_limit_best, charge_window_best, export_window_best, export_limits_best, False, end_record=end_record) + + my_predbat.charge_limit_best = charge_limit_best + my_predbat.export_limits_best = export_limits_best + my_predbat.charge_window_best = charge_window_best + my_predbat.export_window_best = export_window_best + my_predbat.optimise_all_windows(result[0], result[8]) + + final = my_predbat.run_prediction(my_predbat.charge_limit_best, my_predbat.charge_window_best, my_predbat.export_window_best, my_predbat.export_limits_best, False, end_record=end_record, save="best") + # Export limits are a packed structure since #5047, so ask the accessor rather than comparing the + # raw value - a tuple is not less than 100, and reading it as one silently reports "did not export" + exported = bool(my_predbat.export_limits_best) and export_mode_of(my_predbat.export_limits_best[0]) == EXPORT_MODE_TARGET + return exported, final[1] + final[2] + + +def test_minimum_is_bought_before_the_ready_time(my_predbat): + """Solar after the deadline must not satisfy car_charging_plan_min_soc. + + Solar windows reach for the full limit and run to the forecast horizon, and a single running + car_soc is shared with the bought pass. Planned first, tomorrow's sunshine could carry the car + past its limit and break the loop before one overnight slot was bought - so a car sitting below + the minimum at 01:30 was offered nothing against cheap import, and the "60% by 07:30" guarantee + was met hours late. Reported from a live system. + """ + print(" - test_minimum_is_bought_before_the_ready_time") + failed = False + # Ready in 4 hours, sun not until 6 hours out: nothing solar can count towards the guarantee. + # The deadline has to be wide enough that the charger is not the binding constraint - 20kWh at + # 7.4kW needs most of three hours, so a tighter window would fail on physics, not on the bug. + setup_car(my_predbat, car_kwh=40.0, ready_ahead=240) + my_predbat.car_charging_soc = [10.0] + my_predbat.car_charging_plan_min_soc = 60 # of a 50kWh pack = 30kWh, so 20kWh must be bought + my_predbat.car_charging_solar = True + reset_rates(my_predbat, 30.0, 5.0) + low_rates = build_low_rates(my_predbat) + update_rates_import(my_predbat, low_rates) + set_pv(my_predbat, 7.0, start_offset=360, length=600) + + plan = my_predbat.plan_car_charging(0, low_rates) + ready_minutes = my_predbat.minutes_now + 240 + bought_by_ready = sum(slot["kwh"] for slot in plan if not slot.get("solar") and slot["end"] <= ready_minutes) + + if bought_by_ready <= 0: + print("ERROR: nothing was bought before the ready time, so the minimum is met only by later solar: {}".format(plan)) + failed = True + # 10kWh in the car, 30kWh minimum: the shortfall has to be bought, not left to the sun + elif bought_by_ready < 19.0: + print("ERROR: expected about 20kWh bought before the ready time, got {} from {}".format(bought_by_ready, plan)) + failed = True + + # The solar that follows should still take it beyond the minimum towards the full limit + if not any(slot.get("solar") for slot in plan): + print("ERROR: solar should still top the car up above the minimum, got {}".format(plan)) + failed = True + + return failed + + +def test_car_export_tradeoff(my_predbat): + """Exporting versus saving the battery for the car is already decided by the cost model. + + This needs no dedicated mechanism: the car's load is part of the prediction, so an export that + drains the battery is priced against the import the car then needs. The test pins that, because + the behaviour is easy to break and not obvious from reading either module alone. + """ + print(" - test_car_export_tradeoff") + failed = False + reset_inverter(my_predbat) + + # Export pays far less than the import the car would otherwise need: keep the charge + exported_cheap, import_cheap = run_car_export_tradeoff(my_predbat, export_rate=5.0, from_battery=True) + if exported_cheap: + print("ERROR: should not export at 5p when the car will need import at 30p") + failed = True + + # Export pays far more than that import: sell it and buy the car's energy back + exported_rich, _ = run_car_export_tradeoff(my_predbat, export_rate=60.0, from_battery=True) + if not exported_rich: + print("ERROR: should export at 60p even though the car will then import at 30p") + failed = True + + # Letting the battery serve the car reduces what is bought from the grid + _, import_blocked = run_car_export_tradeoff(my_predbat, export_rate=5.0, from_battery=False) + if import_cheap >= import_blocked: + print("ERROR: car_charging_from_battery on should import less ({}) than off ({})".format(import_cheap, import_blocked)) + failed = True + + return failed + + +def run_car_solar_tests(my_predbat): + """Run every solar car-charging and car/export trade-off test. + + The car settings live on the shared my_predbat instance, so they are snapshotted and put back + afterwards. Without this a lowered car_charging_plan_min_soc leaks into whichever test runs next + and silently halves its expected charge - which only shows up in some test orderings. The load + forecast inputs set_house_load overwrites are carried for the same reason: leaving + load_forecast_only on would zero the historical load of every test that follows. + """ + print("**** Running car solar tests ****\n") + carried = ( + "minutes_now", + "load_forecast_only", + "load_forecast", + "load_scaling", + "load_inday_adjustment", + "load_scaling_dynamic", + "manual_load_adjust", + "metric_load_divergence", + "dynamic_load_baseline", + "num_cars", + "car_charging_soc", + "car_charging_limit", + "car_charging_battery_size", + "car_charging_rate", + "car_charging_loss", + "car_charging_slots", + "car_charging_plan_time", + "car_charging_plan_smart", + "car_charging_plan_max_price", + "car_charging_now", + "car_charging_solar", + "car_charging_solar_excess", + "manual_car_away_times", + "battery_rate_max_charge", + "battery_rate_max_scaling", + "soc_kw", + "soc_max", + "car_charging_rate_threshold_export", + "car_charging_plan_min_soc", + "car_charging_from_battery", + ) + saved = {name: getattr(my_predbat, name, None) for name in carried} + + try: + failed = test_solar_windows_selection(my_predbat) + failed |= test_solar_reduces_paid_import(my_predbat) + failed |= test_solar_slots_do_not_overlap(my_predbat) + failed |= test_solar_windows_ignore_ready_time(my_predbat) + failed |= test_minimum_is_bought_before_the_ready_time(my_predbat) + failed |= test_min_soc_splits_bought_from_solar(my_predbat) + failed |= test_solar_surplus_floored_per_bucket(my_predbat) + failed |= test_solar_window_needs_real_surplus(my_predbat) + failed |= test_solar_slot_size_follows_surplus(my_predbat) + failed |= test_solar_slot_capped_by_charger(my_predbat) + failed |= test_solar_battery_takes_surplus_first(my_predbat) + failed |= test_bought_slots_prefer_no_sun(my_predbat) + failed |= test_away_moves_solar_earlier(my_predbat) + if failed: + return failed + failed |= test_car_export_tradeoff(my_predbat) + finally: + for name, value in saved.items(): + setattr(my_predbat, name, value) + return failed diff --git a/apps/predbat/tests/test_infra.py b/apps/predbat/tests/test_infra.py index 17d63634b..8e732b969 100644 --- a/apps/predbat/tests/test_infra.py +++ b/apps/predbat/tests/test_infra.py @@ -504,6 +504,10 @@ def get_default_config(self): "car_charging_limit": 100.0, "car_charging_exclusive": False, "car_charging_from_battery": False, + "car_charging_solar": False, + "car_charging_solar_excess": 1.0, + "car_charging_rate_threshold_export": 99, + "car_charging_plan_min_soc": 100, "car_charging_planned_response": ["yes", "on", "enable", "true"], "car_charging_now_response": ["yes", "on", "enable", "true", "charging"], "combine_rate_threshold": 1.0, diff --git a/apps/predbat/unit_test.py b/apps/predbat/unit_test.py index f076cb42d..b64ae507d 100644 --- a/apps/predbat/unit_test.py +++ b/apps/predbat/unit_test.py @@ -41,6 +41,7 @@ from tests.test_predheat import test_predheat from tests.test_debug_enable_auto_scope import test_debug_enable_auto_scope from tests.test_charge_hold import run_charge_hold_tests +from tests.test_car_solar import run_car_solar_tests from tests.test_octopus_slots import run_load_octopus_slots_tests, run_octopus_slot_max_default_tests from tests.test_multi_car_iog import run_multi_car_iog_tests from tests.test_fetch_config_options import test_fetch_config_options @@ -156,6 +157,8 @@ from tests.test_manual_soc import run_test_manual_soc from tests.test_manual_soc_max import run_test_manual_soc_max from tests.test_manual_times import run_test_manual_times +from tests.test_car_away import run_car_away_tests +from tests.test_car_deadline import run_car_deadline_tests from tests.test_manual_select import run_test_manual_select from tests.test_minute_array import test_minute_array from tests.test_minute_data import test_minute_data, test_minute_data_load, test_minute_data_no_smoothing_backwards, test_minute_data_no_smoothing_forward @@ -453,6 +456,7 @@ def main(): ("predheat", test_predheat, "Predheat scheduler and predheat_enable gate tests (#4670)", False), ("debug_enable_auto_scope", test_debug_enable_auto_scope, "debug_enable auto-disable-after-N-hours tests (#4438 review)", False), ("charge_hold", run_charge_hold_tests, "Charge freeze hold modelling tests", False), + ("car_solar", run_car_solar_tests, "Solar car charging and car/export trade-off tests", False), ("basic_rates", test_basic_rates, "Basic rates tests", False), ("clock", run_clock_tests, "update_time clock tests", False), ("rate_min_forward_calc", test_rate_min_forward_calc, "Rate min forward calc tests", False), @@ -526,6 +530,8 @@ def main(): ("manual_soc", run_test_manual_soc, "Manual SOC target tests", False), ("manual_soc_max", run_test_manual_soc_max, "Manual SOC maximum (ceiling) target tests (issue #1578)", False), ("manual_times", run_test_manual_times, "Manual times tests", False), + ("car_away", run_car_away_tests, "manual_car_away per-slot override tests", False), + ("car_deadline", run_car_deadline_tests, "manual_car_deadline one-off ready-by level tests", False), ("manual_select", run_test_manual_select, "Manual select tests", False), ("web_if", run_test_web_if, "Web interface tests", False), ("web_apps_edit", run_web_apps_edit_tests, "Apps.yaml editor add/delete tests (issue #4714)", False), diff --git a/apps/predbat/userinterface.py b/apps/predbat/userinterface.py index f33fbd125..1fa785f34 100644 --- a/apps/predbat/userinterface.py +++ b/apps/predbat/userinterface.py @@ -1192,6 +1192,9 @@ def load_user_config(self, quiet=True, register=False, load_config=False): if "_import" in item["name"]: # Manual import rate self.manual_rates(name, new_value=ha_value, default_rate=self.get_arg("manual_import_value")) + elif "_car_deadline" in item["name"]: + # A slot picked from the HA dropdown carries no "=level", so it takes the deadline default + self.manual_rates(name, new_value=ha_value, default_rate=self.get_arg("manual_car_deadline_value")) else: self.manual_rates(name, new_value=ha_value, default_rate=self.get_arg("manual_export_value")) elif item.get("api"): @@ -1457,6 +1460,9 @@ def manual_select(self, config_item, value): elif "_soc" in item["name"]: # Manual soc rate self.manual_rates(config_item, new_value=item_value, default_rate=self.get_arg("manual_soc_value")) + elif "_car_deadline" in item["name"]: + # Car charge level promised by a one-off deadline + self.manual_rates(config_item, new_value=item_value, default_rate=self.get_arg("manual_car_deadline_value")) else: self.log("Warn: Manual rate sensor {} not recognised".format(config_item)) else: @@ -1601,6 +1607,9 @@ def manual_rates(self, config_item, exclude=None, new_value=None, default_rate=0 values = new_value else: values = item.get("value", "") + # A select gated on "enable" (manual_car_deadline on num_cars) reads back None rather than "" until it + # is enabled - the same shape that crashed headless startup through manual_times() for manual_car_away + values = values or "" values = values.replace("+", "") values_list = [] if values: @@ -1693,6 +1702,11 @@ def manual_times(self, config_item, exclude=None, new_value=None, update=True): values = new_value else: values = item.get("value", "") + # An item that is registered but has never been given a value reads back as None rather than + # "", which is what a config item gated behind an "enable" looks like before the UI has + # published it - and is what the headless annual bootstrap sees. Treat it as no selection + # rather than letting .replace() raise out of fetch_config_options(). + values = values or "" values = values.replace("+", "") values_list = [] if values: diff --git a/apps/predbat/web.py b/apps/predbat/web.py index 613c33289..05b866da9 100644 --- a/apps/predbat/web.py +++ b/apps/predbat/web.py @@ -2586,6 +2586,7 @@ async def html_api_plan_data(self, request): manual_freeze_charge_times = self.base.manual_times("manual_freeze_charge", update=False) manual_freeze_export_times = self.base.manual_times("manual_freeze_export", update=False) manual_demand_times = self.base.manual_times("manual_demand", update=False) + manual_car_away_times = self.base.manual_times("manual_car_away", update=False) manual_import_rates = self.base.manual_rates("manual_import_rates", update=False) manual_export_rates = self.base.manual_rates("manual_export_rates", update=False) manual_load_adjust = self.base.manual_rates("manual_load_adjust", update=False) @@ -2598,6 +2599,8 @@ async def html_api_plan_data(self, request): manual_load_adjust_list = [{"minutes": k, "adjustment": v} for k, v in manual_load_adjust.items()] manual_soc_list = [{"minutes": k, "target": v} for k, v in manual_soc_keep.items()] manual_soc_max_list = [{"minutes": k, "target": v} for k, v in manual_soc_max_keep.items()] + manual_car_deadline_keep = self.base.manual_rates("manual_car_deadline", default_rate=self.base.get_arg("manual_car_deadline_value"), update=False) + manual_car_deadline_list = [{"minutes": k, "target": v} for k, v in manual_car_deadline_keep.items()] # Build overrides object overrides = { @@ -2606,11 +2609,13 @@ async def html_api_plan_data(self, request): "manual_freeze_charge_times": manual_freeze_charge_times, "manual_freeze_export_times": manual_freeze_export_times, "manual_demand_times": manual_demand_times, + "manual_car_away_times": manual_car_away_times, "manual_import_rates": manual_import_rates_list, "manual_export_rates": manual_export_rates_list, "manual_load_adjust": manual_load_adjust_list, "manual_soc": manual_soc_list, "manual_soc_max": manual_soc_max_list, + "manual_car_deadline": manual_car_deadline_list, } # Calculate hash of overrides for change detection @@ -2685,6 +2690,7 @@ async def html_plan(self, request): manual_freeze_charge_times = self.base.manual_times("manual_freeze_charge", update=False) manual_freeze_export_times = self.base.manual_times("manual_freeze_export", update=False) manual_demand_times = self.base.manual_times("manual_demand", update=False) + manual_car_away_times = self.base.manual_times("manual_car_away", update=False) manual_import_rates = self.base.manual_rates("manual_import_rates", update=False) manual_export_rates = self.base.manual_rates("manual_export_rates", update=False) manual_load_adjust = self.base.manual_rates("manual_load_adjust", update=False) @@ -2697,6 +2703,8 @@ async def html_plan(self, request): manual_load_adjust_list = [{"minutes": k, "adjustment": v} for k, v in manual_load_adjust.items()] manual_soc_list = [{"minutes": k, "target": v} for k, v in manual_soc_keep.items()] manual_soc_max_list = [{"minutes": k, "target": v} for k, v in manual_soc_max_keep.items()] + manual_car_deadline_keep = self.base.manual_rates("manual_car_deadline", default_rate=self.base.get_arg("manual_car_deadline_value"), update=False) + manual_car_deadline_list = [{"minutes": k, "target": v} for k, v in manual_car_deadline_keep.items()] # Build overrides object overrides = { @@ -2705,11 +2713,13 @@ async def html_plan(self, request): "manual_freeze_charge_times": manual_freeze_charge_times, "manual_freeze_export_times": manual_freeze_export_times, "manual_demand_times": manual_demand_times, + "manual_car_away_times": manual_car_away_times, "manual_import_rates": manual_import_rates_list, "manual_export_rates": manual_export_rates_list, "manual_load_adjust": manual_load_adjust_list, "manual_soc": manual_soc_list, "manual_soc_max": manual_soc_max_list, + "manual_car_deadline": manual_car_deadline_list, } # Calculate hash of overrides for change detection @@ -4926,6 +4936,17 @@ async def html_rate_override(self, request): actual_rate = manual_soc_max.get(minutes_from_midnight, rate) clear_option = "[{}={}]".format(override_time.strftime("%a %H:%M"), actual_rate) await self.base.async_manual_select("manual_soc_max", clear_option) + elif action == "Set Car Deadline": + # One deadline at a time: a new one replaces the old rather than promising two levels + rate = max(0.0, min(100.0, rate)) + item = self.base.config_index.get("manual_car_deadline_value", {}) + await self.set_state_external(item.get("entity", None), rate) + await self.base.async_manual_select("manual_car_deadline", "off") + await self.base.async_manual_select("manual_car_deadline", "{}={}".format(override_time.strftime("%a %H:%M"), rate)) + elif action == "Clear Car Deadline": + # There is only ever one, and a slot picked from the HA dropdown is stored without its level, + # so rebuilding the exact "[time=level]" string to toggle it off would not always match + await self.base.async_manual_select("manual_car_deadline", "off") else: self.log("ERROR: Unknown action for rate override") return web.json_response({"success": False, "message": "Unknown action"}, status=400) @@ -4974,6 +4995,9 @@ async def html_plan_override(self, request): if action == "Clear": await self.base.async_manual_select("manual_demand", selection_option) await self.base.async_manual_select("manual_demand", clear_option) + # Clear has to reach every select the slot could be set on, not just demand, or a + # Car Away marker can be set from the plan and never removed from it + await self.base.async_manual_select("manual_car_away", clear_option) else: if action == "Manual Demand": await self.base.async_manual_select("manual_demand", selection_option) @@ -4985,6 +5009,8 @@ async def html_plan_override(self, request): await self.base.async_manual_select("manual_freeze_charge", selection_option) elif action == "Manual Freeze Export": await self.base.async_manual_select("manual_freeze_export", selection_option) + elif action == "Car Away": + await self.base.async_manual_select("manual_car_away", selection_option) else: return web.json_response({"success": False, "message": "Unknown action"}, status=400) diff --git a/apps/predbat/web_helper.py b/apps/predbat/web_helper.py index 3e2f41cdb..27485a887 100644 --- a/apps/predbat/web_helper.py +++ b/apps/predbat/web_helper.py @@ -6940,6 +6940,57 @@ def get_plan_css(): closeDropdowns(); } + // Handle the car ready-by deadline: one level promised by one time, replacing any earlier one + function handleCarDeadline(time, dropdownId, isClear) { + let value = null; + if (!isClear && dropdownId) { + const inputElement = document.getElementById('car-deadline-' + dropdownId); + if (inputElement) { + value = inputElement.value; + } + } + + const formData = new FormData(); + formData.append('time', time); + formData.append('action', isClear ? 'Clear Car Deadline' : 'Set Car Deadline'); + formData.append('rate', value || '0'); + + fetch('./rate_override', { + method: 'POST', + body: formData + }) + .then(response => response.json()) + .then(data => { + if (data.success) { + const messageElement = document.createElement('div'); + messageElement.textContent = isClear ? `Car ready-by cleared` : `Car ready by ${time} at ${value}%`; + messageElement.style.position = 'fixed'; + messageElement.style.top = '65px'; + messageElement.style.right = '10px'; + messageElement.style.padding = '10px'; + messageElement.style.backgroundColor = '#4CAF50'; + messageElement.style.color = 'white'; + messageElement.style.borderRadius = '4px'; + messageElement.style.zIndex = '1000'; + document.body.appendChild(messageElement); + setTimeout(() => { + messageElement.style.opacity = '0'; + messageElement.style.transition = 'opacity 0.5s'; + setTimeout(() => messageElement.remove(), 500); + }, 3000); + setTimeout(() => location.reload(), 1000); + } else { + showErrorMessage(data.message || 'Unknown error'); + } + }) + .catch(error => { + console.error('Error:', error); + showErrorMessage(error.message); + }); + + closeDropdowns(); + } + // Close dropdowns when clicking outside document.addEventListener("click", function(event) { if (!event.target.matches('.clickable-time-cell') && !event.target.matches('.clickable-state-cell') && !event.target.closest('.dropdown-content')) { @@ -7372,9 +7423,13 @@ def get_plan_renderer_js(): overrides.manual_export_times, overrides.manual_freeze_charge_times, overrides.manual_freeze_export_times, - overrides.manual_demand_times + overrides.manual_demand_times, + overrides.manual_car_away_times || [] ); + // The car ready-by deadline carries a level, so it is shown with it rather than as a bare highlight + const carDeadline = (overrides.manual_car_deadline || []).find(r => r.minutes === minutesFromMidnight); + // Determine highlight color based on override type let bgColor = '#FFFFFF'; let overrideClass = ''; @@ -7394,10 +7449,19 @@ def get_plan_renderer_js(): } else if (overrides.manual_freeze_export_times.includes(minutesFromMidnight)) { bgColor = '#D8D8D8'; // Darker gray hint overrideClass = 'override-freeze-export'; + } else if ((overrides.manual_car_away_times || []).includes(minutesFromMidnight)) { + bgColor = '#E0E0FF'; // Light blue hint + overrideClass = 'override-car-away'; + } else if (carDeadline) { + bgColor = '#D0E8FF'; // Sky blue hint + overrideClass = 'override-car-deadline'; } let html = ``; html += timeDisplay; + if (carDeadline) { + html += ` ⏱${carDeadline.target}%`; + } html += '