diff --git a/esphome/components/ct002/balancer.cpp b/esphome/components/ct002/balancer.cpp index 44155b6d..f739f80e 100644 --- a/esphome/components/ct002/balancer.cpp +++ b/esphome/components/ct002/balancer.cpp @@ -575,22 +575,39 @@ bool LoadBalancer::resolve_probe_state_(const ReportMap &reports, double now, return false; } -float LoadBalancer::compute_desired_contribution_( - const std::string &consumer_id, const ReportMap &reports, - const std::unordered_map &weights, float desired_total) { +// Element-to-key adapters so weighted_share can walk a ReportMap, a vector of +// ids, or a vector of id pointers without three copies of the loop. +static inline const std::string &key_of(const std::string &s) { return s; } +static inline const std::string &key_of(const std::string *s) { return *s; } +static inline const std::string &key_of(const std::pair &p) { + return p.first; +} + +// *total* split across *ids* in proportion to *weights*, falling back to an even +// split when the weights sum to zero, so a pool whose every member is saturated +// still shares rather than stalling. Mirrors balancer.py weighted_share. +template +static float weighted_share(float total, const std::unordered_map &weights, + const Ids &ids, size_t count, const std::string *consumer_id) { float total_weight = 0.0f; - for (const auto &r : reports) { - auto it = weights.find(r.first); + for (const auto &cid : ids) { + auto it = weights.find(key_of(cid)); if (it != weights.end()) total_weight += it->second; } - float fair_share; - if (total_weight > 0.0f) { - auto it = weights.find(consumer_id); - const float w = (it != weights.end()) ? it->second : 0.0f; - fair_share = desired_total * w / total_weight; - } else { - fair_share = desired_total / std::max(1, reports.size()); + if (total_weight <= 0.0f) return total / static_cast(std::max(1, count)); + float mine = 0.0f; + if (consumer_id != nullptr) { + auto it = weights.find(*consumer_id); + if (it != weights.end()) mine = it->second; } + return total * mine / total_weight; +} + +float LoadBalancer::compute_desired_contribution_( + const std::string &consumer_id, const ReportMap &reports, + const std::unordered_map &weights, float desired_total) { + const float fair_share = + weighted_share(desired_total, weights, reports, reports.size(), &consumer_id); const bool not_in_reports = reports.find(consumer_id) == reports.end(); if (!this->cfg_.fair_distribution || not_in_reports || (this->cfg_.balance_deadband > 0.0f && @@ -818,6 +835,32 @@ void LoadBalancer::log_steer_(const std::optional &consumer_id, this->steer_log_sink_(format_steer_log(entry)); } +// Score how well this consumer is following the commands it is sent. The +// detector keys off the *unpaced* command (last_intent_reading), not the paced +// last_target: pacing pins a stuck battery at the base step, which would look +// "idle" when pace_base_step < min_target (issue #522). The floor is compared +// against the unpaced intent for the same reason — a battery the pacing clamp +// is holding below its floor must still register as pushed, or a full/empty one +// would never be detected while clamped; #614's stall escape bounds that window. +// Skipped for manual and probing consumers, and for a deprioritized one, whose +// fade path still carries a transient non-zero command that would score as +// "cannot follow" and lock maybe_force_swap_saturated_ out of promoting it back. +// Mirrors balancer.py _track_saturation. +void LoadBalancer::track_saturation_(const std::string &consumer_id, + BalancerConsumerState &state, ConsumerMode mode, + ReportMap &reports) { + if (reports.find(consumer_id) == reports.end() || mode.kind == ConsumerModeKind::MANUAL) + return; + const auto probe_set = this->probe_participants_(); + if (probe_set.find(consumer_id) != probe_set.end() || + this->deprioritized_.find(consumer_id) != this->deprioritized_.end()) + return; + const ConsumerReport &report = reports[consumer_id]; + this->saturation_.update( + state, state.last_intent_reading, report.power, + saturation_floor(state, report, this->effective_min_dc_output_(consumer_id, reports))); +} + std::array LoadBalancer::compute_target( const std::optional &consumer_id, ConsumerMode mode, const ReportMap &all_reports, float grid_total, @@ -840,26 +883,9 @@ std::array LoadBalancer::compute_target( } BalancerConsumerState *state = nullptr; - if (consumer_id) state = &this->get_consumer_(*consumer_id); - // Saturation keys off the *unpaced* command (last_intent_reading), not the - // paced last_target: pacing pins a stuck battery at the base step, which would - // look "idle" when pace_base_step < min_target (issue #522). - std::optional last_intent_reading = state ? state->last_intent_reading : std::optional{}; - - if (consumer_id && state && active_reports.find(*consumer_id) != active_reports.end() && - mode.kind != ConsumerModeKind::MANUAL) { - const auto probe_set = this->probe_participants_(); - if (probe_set.find(*consumer_id) == probe_set.end() && - this->deprioritized_.find(*consumer_id) == this->deprioritized_.end()) { - const ConsumerReport &report = active_reports[*consumer_id]; - // The floor is compared against the *unpaced* intent, like the rest of - // this call (issue #522): a battery the pacing clamp is holding below its - // floor must still register as pushed, or a full/empty one would never be - // detected while clamped. #614's stall escape bounds that window. - this->saturation_.update( - *state, last_intent_reading, report.power, - saturation_floor(*state, report, this->effective_min_dc_output_(*consumer_id, active_reports))); - } + if (consumer_id) { + state = &this->get_consumer_(*consumer_id); + this->track_saturation_(*consumer_id, *state, mode, active_reports); } if (mode.kind == ConsumerModeKind::MANUAL && consumer_id && state) { @@ -1033,6 +1059,38 @@ BalancerSnapshot LoadBalancer::status_snapshot() const { // Auto-target pipeline // ------------------------------------------------------------------------- +// This consumer's slice of the grid imbalance, in W. Two terms, kept apart +// deliberately. The *tracking* term is its share of the grid error and carries +// the grid's sign by construction; the *balancing* term equalizes output across +// the same-phase pool and is zero-sum, so it is grid-neutral. Only the tracking +// term is clamped against the predicted grid direction — zeroing the balancing +// term too would make equalization one-sided near steady state (issue #523). +// The caller folds the result into what the consumer reports now to get an +// absolute net-output target. Mirrors balancer.py _residual_share. +float LoadBalancer::residual_share_(const std::optional &consumer_id, + const ReportMap &reports, float control_grid, + const std::unordered_map &eff_part, + const std::unordered_set &charge_blind) { + float fair_share = fair_share_(consumer_id, reports, control_grid, eff_part); + const auto concentrated = + this->concentrated_share_(consumer_id, reports, control_grid, eff_part, charge_blind); + if (concentrated) fair_share = *concentrated; + this->diag_fair_share_ = fair_share; + + float residual = fair_share; + if (this->cfg_.fair_distribution && !concentrated.has_value() && consumer_id && + reports.find(*consumer_id) != reports.end() && eff_part.count(*consumer_id)) { + residual = this->balance_correction_(*consumer_id, reports, eff_part, fair_share); + } + + float tracking = fair_share; + if ((control_grid < 0.0f && tracking > 0.0f) || + (control_grid > 0.0f && tracking < 0.0f)) { + tracking = 0.0f; + } + return tracking + (residual - fair_share); +} + std::array LoadBalancer::compute_auto_target_( const std::optional &consumer_id, const ReportMap &reports, float grid_total, const std::vector &sample_id) { @@ -1059,18 +1117,20 @@ std::array LoadBalancer::compute_auto_target_( this->predict_control_grid_(reports, grid_total, sample_id), trim_fresh); this->diag_control_grid_ = control_grid; + const auto blind = charge_blind_(reports, grid_total); + const std::unordered_set &charge_blind = blind.first; + const bool any_ac_chargeable = blind.second; + // Share weight per consumer: a saturated battery (one that stopped following + // its commands) earns a smaller slice, floored just above zero so it can + // recover; a charge-blind one earns nothing. std::unordered_map saturation; for (const auto &c : this->consumers_) saturation[c.first] = static_cast(c.second.saturation_score); std::unordered_map eff_part; for (const auto &r : reports) { - const float s = saturation.count(r.first) ? saturation[r.first] : 0.0f; - eff_part[r.first] = std::max(0.01f, 1.0f - s); + const float sat = saturation.count(r.first) ? saturation[r.first] : 0.0f; + eff_part[r.first] = std::max(0.01f, 1.0f - sat); } - - const auto blind = charge_blind_(reports, grid_total); - const std::unordered_set &charge_blind = blind.first; - const bool any_ac_chargeable = blind.second; for (const auto &cid : charge_blind) eff_part[cid] = 0.0f; auto efficiency_adjustments = @@ -1101,43 +1161,14 @@ std::array LoadBalancer::compute_auto_target_( for (const auto &kv : faded_adjustments) { if (eff_part.count(kv.first) && kv.second == 0.0f) eff_part[kv.first] = 0.0f; } - if (!faded_adjustments.empty() && consumer_id) { + if (consumer_id) { auto it = faded_adjustments.find(*consumer_id); - if (it != faded_adjustments.end() && it->second == 0.0f) { + if (it != faded_adjustments.end() && it->second == 0.0f) return this->steer_to_zero_(consumer_id, reports, /*paced=*/true); - } } - float fair_share = fair_share_(consumer_id, reports, control_grid, eff_part); - const auto concentrated = - this->concentrated_share_(consumer_id, reports, control_grid, eff_part, charge_blind); - if (concentrated) fair_share = *concentrated; - this->diag_fair_share_ = fair_share; - - // fair_share / balance_correction_ produce the residual: this consumer's - // slice of the grid imbalance to fold into its current output. The absolute - // net-output target is "what I report now plus my residual" (NetOutputW wrap - // below). - float residual; - if (!this->cfg_.fair_distribution || !consumer_id || - reports.find(*consumer_id) == reports.end() || concentrated.has_value()) { - residual = fair_share; - } else if (eff_part.count(*consumer_id)) { - residual = this->balance_correction_(*consumer_id, reports, eff_part, fair_share); - } else { - residual = fair_share; - } - // Clamp only the grid-tracking half (fair_share, which carries the grid's - // sign by construction) against the grid direction — never the - // balance-correction term, which is zero-sum across the same-phase pool and - // so grid-neutral; zeroing it too would make equalization one-sided near - // steady state (issue #523). - float tracking = fair_share; - if ((control_grid < 0.0f && tracking > 0.0f) || - (control_grid > 0.0f && tracking < 0.0f)) { - tracking = 0.0f; - } - residual = tracking + (residual - fair_share); + float residual = + this->residual_share_(consumer_id, reports, control_grid, eff_part, charge_blind); if (consumer_id) { residual = this->damp_oscillation_(*consumer_id, residual); } @@ -1442,39 +1473,22 @@ bool LoadBalancer::cannot_absorb_(const ReportMap &reports) const { // manual / inactive steer-to-zero bypass it (see balancer.py for the // rationale). Caps are W per PACE_REFERENCE_DT; the per-poll clamp scales // with the consumer's observed inter-poll time, clamped at 1.0. -float LoadBalancer::pace_reading_(const std::string &consumer_id, float reading, float reported, - const ReportMap &reports) { +// The learning half of pace_reading_: the cap tracks what the battery has +// *demonstrated* it can slew. It resets to the base step on a direction +// reversal, doubles per reference second while the battery visibly follows, and +// grows against a persistent stall so a device held under its minimum actionable +// command is not clamped there forever. Records what the device responded to +// (pace_responded_at) and how long it has been unresponsive (pace_stall_polls); +// the caller owns pace_cap itself. Mirrors balancer.py _pace_cap. +float LoadBalancer::pace_cap_(BalancerConsumerState &state, float reading, float reported, + int sign, float dt_ratio, bool can_stall, bool *stalled) { const float base = this->cfg_.pace_base_step; - if (base <= 0.0f) return reading; - auto &state = this->get_consumer_(consumer_id); - const double now = this->clock_(); - double dt = (state.pace_last_at > 0.0) ? now - state.pace_last_at : 0.0; - if (dt <= 0.0) { - // First paced poll, a non-advancing clock, or a backwards jump: assume - // one reference period rather than starving the clamp. - dt = PACE_REFERENCE_DT; - } - state.pace_last_at = now; - const float dt_ratio = static_cast(std::min(1.0, dt / PACE_REFERENCE_DT)); - // Reversals are paced too (bounds overshoot at zero crossings); consumers - // needing the unpaced control intent (issue #376 cross-talk attribution) - // read last_intent instead. - const int sign = (reading > 0.0f) ? 1 : (reading < 0.0f ? -1 : 0); float cap = (state.pace_cap > 0.0f) ? state.pace_cap : base; // Floored at the base step: hysteresis-regulator devices (B2500) need a // minimum reading to clear their input hold window at all; the cadence // scale still bounds the grown cap (mirrors balancer.py). - float limit = std::max(base, cap * dt_ratio); - // The stall escape and the response floor below only apply to devices that - // actually have a minimum actionable command — the DC-output family, whose - // channels are a hard on/off below their minimum. Every other battery can - // execute an arbitrarily small command, so it can never be deadlocked by the - // clamp; leaving it on the unmodified path keeps its behaviour bit-for-bit - // and confines the overshoot cost to the devices that need it. - const auto report_it = reports.find(consumer_id); - const bool can_stall = - report_it != reports.end() && needs_dc_output_floor(report_it->second.device_type); - bool stalled = false; + const float limit = std::max(base, cap * dt_ratio); + *stalled = false; if (sign == 0 || sign != state.pace_sign) { cap = base; state.pace_stall_polls = 0; @@ -1501,7 +1515,7 @@ float LoadBalancer::pace_reading_(const std::string &consumer_id, float reading, } else { // Held below what the device can act on: grow anyway once the stall has // persisted, or the clamp is self-sustaining (see PACE_STALL_ESCAPE_POLLS). - stalled = can_stall; + *stalled = can_stall; state.pace_stall_polls++; if (can_stall && state.pace_stall_polls >= PACE_STALL_ESCAPE_POLLS) { cap = std::min(cap * std::pow(PACE_GROWTH_FACTOR, dt_ratio), this->cfg_.pace_max_step); @@ -1516,7 +1530,40 @@ float LoadBalancer::pace_reading_(const std::string &consumer_id, float reading, // else branch back-computes cap as fabs(reading) / dt_ratio, which a fast poll // (small dt_ratio) can inflate past the max — and a later normal-cadence poll // would then slew beyond pace_max_step. - cap = std::min(cap, this->cfg_.pace_max_step); + return std::min(cap, this->cfg_.pace_max_step); +} + +float LoadBalancer::pace_reading_(const std::string &consumer_id, float reading, float reported, + const ReportMap &reports) { + const float base = this->cfg_.pace_base_step; + if (base <= 0.0f) return reading; + auto &state = this->get_consumer_(consumer_id); + const double now = this->clock_(); + double dt = (state.pace_last_at > 0.0) ? now - state.pace_last_at : 0.0; + if (dt <= 0.0) { + // First paced poll, a non-advancing clock, or a backwards jump: assume + // one reference period rather than starving the clamp. + dt = PACE_REFERENCE_DT; + } + state.pace_last_at = now; + const float dt_ratio = static_cast(std::min(1.0, dt / PACE_REFERENCE_DT)); + // Reversals are paced too (bounds overshoot at zero crossings); consumers + // needing the unpaced control intent (issue #376 cross-talk attribution) + // read last_intent instead. + const int sign = (reading > 0.0f) ? 1 : (reading < 0.0f ? -1 : 0); + // The stall escape and the response floor below only apply to devices that + // actually have a minimum actionable command — the DC-output family, whose + // channels are a hard on/off below their minimum. Every other battery can + // execute an arbitrarily small command, so it can never be deadlocked by the + // clamp; leaving it on the unmodified path keeps its behaviour bit-for-bit + // and confines the overshoot cost to the devices that need it. + const auto report_it = reports.find(consumer_id); + const bool can_stall = + report_it != reports.end() && needs_dc_output_floor(report_it->second.device_type); + + bool stalled = false; + const float cap = this->pace_cap_(state, reading, reported, sign, dt_ratio, can_stall, + &stalled); state.pace_cap = cap; state.pace_sign = sign; state.pace_prev_reported = reported; @@ -1526,7 +1573,7 @@ float LoadBalancer::pace_reading_(const std::string &consumer_id, float reading, // back down to base is precisely what keeps it stalled (max(base, cap * // dt_ratio) stays at base until cap reaches base / dt_ratio, so growing the // cap alone never frees a 0.3 s poller). - limit = std::max(base, stalled ? cap : cap * dt_ratio); + float limit = std::max(base, stalled ? cap : cap * dt_ratio); // Never clamp under a level this device has demonstrably responded to: for a // hysteresis regulator a smaller command is not a gentler one but an *off* // one, so the unit would switch off, stop moving, and need lifting again @@ -1557,18 +1604,16 @@ bool LoadBalancer::concentration_pool_balanced_(const ReportMap &reports, return false; } float actual_total = 0.0f; - float total_weight = 0.0f; + std::unordered_map weights; for (const auto *cid : conc_ids) { const auto &rep = reports.at(*cid); actual_total += rep.power; - total_weight += rep.weight; + weights[*cid] = rep.weight; } for (const auto *cid : conc_ids) { - const auto &rep = reports.at(*cid); - const float target_share = (total_weight > 0.0f) - ? actual_total * rep.weight / total_weight - : actual_total / static_cast(conc_ids.size()); - if (std::fabs(target_share - rep.power) >= deadband) return false; + const float target_share = + weighted_share(actual_total, weights, conc_ids, conc_ids.size(), cid); + if (std::fabs(target_share - reports.at(*cid).power) >= deadband) return false; } return true; } @@ -1596,22 +1641,13 @@ float LoadBalancer::balance_correction_(const std::string &consumer_id, // output rather than the plain average, so the configured ratio is the steady // state. Participation is decided by eff_part above, so a healthy battery with // a small weight is not dropped. Neutral weights reduce to the plain average. - float total_weight = 0.0f; std::unordered_map weights; for (const auto &cid : participating) { - float w = 1.0f; auto it = reports.find(cid); - if (it != reports.end()) w = it->second.weight; - weights[cid] = w; - total_weight += w; - } - float target_share; - if (total_weight > 0.0f) { - const float wself = weights.count(consumer_id) ? weights[consumer_id] : 0.0f; - target_share = actual_total * wself / total_weight; - } else { - target_share = actual_total / participating.size(); + weights[cid] = (it != reports.end()) ? it->second.weight : 1.0f; } + const float target_share = + weighted_share(actual_total, weights, participating, participating.size(), &consumer_id); const float error = target_share - actual_self; const float err_abs = std::fabs(error); if (cfg.balance_deadband > 0.0f && err_abs < cfg.balance_deadband) return fair_share; @@ -1641,6 +1677,35 @@ float LoadBalancer::balance_correction_(const std::string &consumer_id, // Efficiency deprioritization // ------------------------------------------------------------------------- +// Probe a swap the efficiency pass just made, before trusting it. Promoting a +// battery is a bet that it delivers more than the one it displaced; when this +// tick both promotes and demotes, the bet is tested against the battery it +// displaced rather than assumed (see begin_probe_). Nothing to test on the +// first pass, when there is no previous active set to compare against. +// Mirrors balancer.py _probe_active_set_change. +void LoadBalancer::probe_active_set_change_(const std::vector &previous_active, + size_t slots, double now) { + if (previous_active.empty()) return; + std::vector final_active(this->priority_.begin(), + this->priority_.begin() + + std::min(slots, this->priority_.size())); + std::vector promoted; + for (const auto &cid : final_active) { + if (std::find(previous_active.begin(), previous_active.end(), cid) == + previous_active.end()) { + promoted.push_back(cid); + } + } + std::vector backups; + for (const auto &cid : previous_active) { + if (std::find(final_active.begin(), final_active.end(), cid) == final_active.end()) + backups.push_back(cid); + } + if (!promoted.empty() && !backups.empty()) { + this->begin_probe_(promoted[0], final_active, backups, previous_active, now); + } +} + std::unordered_map LoadBalancer::compute_efficiency_deprioritized_( const ReportMap &reports, const std::vector &sample_id, float grid_total) { const auto &cfg = this->cfg_; @@ -1754,26 +1819,8 @@ std::unordered_map LoadBalancer::compute_efficiency_depriori } } - std::vector final_active(this->priority_.begin(), - this->priority_.begin() + - std::min(slots, this->priority_.size())); - if (!probe_active && !probe_resolved && !previous_active.empty()) { - std::vector promoted; - for (const auto &cid : final_active) { - if (std::find(previous_active.begin(), previous_active.end(), cid) == - previous_active.end()) { - promoted.push_back(cid); - } - } - std::vector backups; - for (const auto &cid : previous_active) { - if (std::find(final_active.begin(), final_active.end(), cid) == final_active.end()) - backups.push_back(cid); - } - if (!promoted.empty() && !backups.empty()) { - this->begin_probe_(promoted[0], final_active, backups, previous_active, now); - } - } + if (!probe_active && !probe_resolved) + this->probe_active_set_change_(previous_active, slots, now); for (const auto &cid : deprioritized) { if (this->deprioritized_.find(cid) == this->deprioritized_.end()) { diff --git a/esphome/components/ct002/balancer.h b/esphome/components/ct002/balancer.h index 9cbf6ecc..4784aed7 100644 --- a/esphome/components/ct002/balancer.h +++ b/esphome/components/ct002/balancer.h @@ -547,6 +547,10 @@ class LoadBalancer { protected: BalancerConsumerState &get_consumer_(const std::string &consumer_id); + // Score how well a consumer is following its commands. Mirrors balancer.py + // _track_saturation. + void track_saturation_(const std::string &consumer_id, BalancerConsumerState &state, + ConsumerMode mode, ReportMap &reports); void invalidate_efficiency_cache_(); std::unordered_set probe_participants_() const; float next_probe_requested_abs_(float current_requested_abs, float ceiling) const; @@ -614,6 +618,13 @@ class LoadBalancer { std::array fading_target_(const std::string &consumer_id, const ReportMap &reports, float grid_total, const std::unordered_map &eff_part); + // This consumer's slice of the grid imbalance: a grid-tracking term clamped + // against the grid direction, plus a grid-neutral balancing term that is not. + // Mirrors balancer.py _residual_share. + float residual_share_(const std::optional &consumer_id, + const ReportMap &reports, float control_grid, + const std::unordered_map &eff_part, + const std::unordered_set &charge_blind); // This consumer's weight-proportional slice of the grid error. Mirrors // balancer.py _fair_share. static float fair_share_(const std::optional &consumer_id, @@ -630,6 +641,8 @@ class LoadBalancer { float fair_share); bool concentration_pool_balanced_(const ReportMap &reports, const std::vector &conc_ids); + float pace_cap_(BalancerConsumerState &state, float reading, float reported, int sign, + float dt_ratio, bool can_stall, bool *stalled); float pace_reading_(const std::string &consumer_id, float reading, float reported, const ReportMap &reports); float damp_oscillation_(const std::string &consumer_id, float residual); @@ -646,6 +659,10 @@ class LoadBalancer { std::unordered_map compute_efficiency_deprioritized_( const ReportMap &reports, const std::vector &sample_id, float grid_total); + // Probe a swap the efficiency pass just made, before trusting it. Mirrors + // balancer.py _probe_active_set_change. + void probe_active_set_change_(const std::vector &previous_active, + size_t slots, double now); // Reconcile the rotation order with the reporting pool. Mirrors balancer.py // _sync_pool. void sync_pool_(const ReportMap &reports, double grace); diff --git a/src/astrameter/ct002/balancer.py b/src/astrameter/ct002/balancer.py index 24b6b461..6a990312 100644 --- a/src/astrameter/ct002/balancer.py +++ b/src/astrameter/ct002/balancer.py @@ -5,7 +5,7 @@ import dataclasses import logging import time -from collections.abc import Callable, Mapping +from collections.abc import Callable, Collection, Mapping from typing import Any, Literal, NamedTuple, NewType, get_args from astrameter.config.logger import logger @@ -98,6 +98,24 @@ def __post_init__(self) -> None: """Stand-in for a consumer that did not report this tick (unknown or just removed).""" +def weighted_share( + total: float, + weights: Mapping[str, float], + ids: Collection[str], + consumer_id: str | None, +) -> float: + """*total* split across *ids* in proportion to *weights*. + + Falls back to an even split when the weights sum to zero, so a pool whose + every member is saturated still shares rather than stalling. + """ + total_weight = sum(weights.get(cid, 0.0) for cid in ids) + if total_weight <= 0: + return total / max(1, len(ids)) + mine = 0.0 if consumer_id is None else weights.get(consumer_id, 0.0) + return total * mine / total_weight + + def _report_of(reports: Reports, consumer_id: str | None) -> ConsumerReport: """*consumer_id*'s report, or :data:`NO_REPORT` when it did not report. @@ -1228,11 +1246,7 @@ def _compute_desired_contribution( weights: dict[str, float], desired_total: float, ) -> float: - total_weight = sum(weights.get(cid, 0.0) for cid in reports) - if total_weight > 0: - fair_share = desired_total * weights.get(consumer_id, 0.0) / total_weight - else: - fair_share = desired_total / max(1, len(reports)) + fair_share = weighted_share(desired_total, weights, reports, consumer_id) if ( not self._cfg.fair_distribution or consumer_id not in reports @@ -1369,6 +1383,45 @@ def _log_steer( state.saturation_score if state else 0.0, ) + def _track_saturation( + self, + consumer_id: str, + state: BalancerConsumerState, + consumer_mode: ConsumerMode, + reports: Reports, + ) -> None: + """Score how well this consumer is following the commands it is sent. + + The detector keys off ``last_intent_reading`` — the *unpaced* command — + because pacing pins a battery that can't follow at the base step, and + the paced reading would make a full/empty battery look idle (issue + #522). The floor is compared against the unpaced intent for the same + reason: a battery the clamp holds below its floor must still register + as pushed; the stall escape bounds how long that lasts. + + Skipped for manual and probing consumers, and for a deprioritized one — + its fade path still carries a transient non-zero command that would + score as "cannot follow" and lock ``_maybe_force_swap_saturated`` out of + promoting it back. Its score stays at the zero the symmetric clear in + ``_compute_efficiency_deprioritized`` set. + """ + if ( + consumer_id not in reports + or consumer_mode.mode == "manual" + or consumer_id in self._probe_participants() + or consumer_id in self._deprioritized + ): + return + report = _report_of(reports, consumer_id) + self._saturation.update( + state, + state.last_intent_reading, + report.power, + saturation_floor( + state, report, self._effective_min_dc_output(consumer_id, reports) + ), + ) + def compute_target( self, consumer_id: str | None, @@ -1401,38 +1454,10 @@ def compute_target( cid: r for cid, r in all_reports.items() if cid not in inactive } - # Update saturation (skip manual, probe, and deprioritized consumers). - # The detector keys off ``last_intent_reading`` — the *unpaced* command - # — because pacing pins a battery that can't follow at the base step, - # and the paced reading would make a full/empty battery look idle - # (issue #522). A deprioritized consumer is skipped because its fade - # path still carries a transient non-zero command that would score as - # "cannot follow" and lock ``_maybe_force_swap_saturated`` out of - # promoting it back; its score stays at the zero the symmetric clear in - # ``_compute_efficiency_deprioritized`` set. - state = self._get_consumer(consumer_id) if consumer_id else None - if ( - state is not None - and consumer_id in active_reports - and consumer_mode.mode != "manual" - and consumer_id not in self._probe_participants() - and consumer_id not in self._deprioritized - ): - report = _report_of(active_reports, consumer_id) - actual = report.power - # The floor is compared against the unpaced intent too: a battery - # the clamp holds below its floor must still register as pushed - # (issue #522); the stall escape bounds how long that lasts. - self._saturation.update( - state, - state.last_intent_reading, - actual, - saturation_floor( - state, - report, - self._effective_min_dc_output(consumer_id, active_reports), - ), - ) + state = None + if consumer_id: + state = self._get_consumer(consumer_id) + self._track_saturation(consumer_id, state, consumer_mode, active_reports) if consumer_mode.mode == "manual" and state is not None: reported = _report_of(active_reports, consumer_id).power @@ -1870,10 +1895,12 @@ def _compute_auto_target( control_grid = self._apply_import_trim(control_grid, trim_fresh) self._diag_control_grid = control_grid + charge_blind, any_ac_chargeable = self._charge_blind(reports, grid_total) + # Share weight per consumer: a saturated battery (one that stopped + # following its commands) earns a smaller slice, floored just above + # zero so it can recover; a charge-blind one earns nothing. saturation = {cid: s.saturation_score for cid, s in self._consumers.items()} eff_part = {cid: max(0.01, 1.0 - saturation.get(cid, 0.0)) for cid in reports} - - charge_blind, any_ac_chargeable = self._charge_blind(reports, grid_total) for cid in charge_blind: eff_part[cid] = 0.0 @@ -1905,13 +1932,43 @@ def _compute_auto_target( for cid, fade_w in faded_adjustments.items(): if cid in eff_part and fade_w == 0.0: eff_part[cid] = 0.0 - if ( - faded_adjustments - and consumer_id - and faded_adjustments.get(consumer_id) == 0.0 - ): + if consumer_id and faded_adjustments.get(consumer_id) == 0.0: return self._steer_to_zero(consumer_id, reports, paced=True) + residual = self._residual_share( + consumer_id, reports, control_grid, eff_part, charge_blind + ) + + if consumer_id: + residual = self._damp_oscillation(consumer_id, residual) + + reported = _report_of(reports, consumer_id).power if consumer_id else 0 + return self._emit( + consumer_id, + NetOutputW(reported + residual), + reported, + reports, + eff_part, + pace=True, + ) + + def _residual_share( + self, + consumer_id: str | None, + reports: Reports, + control_grid: float, + eff_part: dict[str, float], + charge_blind: set[str], + ) -> float: + """This consumer's slice of the grid imbalance, in W. + + Two terms, kept apart deliberately. The *tracking* term is its share + of the grid error, and carries the grid's sign by construction; the + *balancing* term equalizes output across the same-phase pool and is + zero-sum, so it is grid-neutral. Only the tracking term is clamped + against the predicted grid direction — zeroing the balancing term too + would make equalization one-sided near steady state (issue #523). + """ fair_share = self._fair_share(consumer_id, reports, control_grid, eff_part) concentrated = self._concentrated_share( consumer_id, reports, control_grid, eff_part, charge_blind @@ -1920,48 +1977,22 @@ def _compute_auto_target( fair_share = concentrated self._diag_fair_share = fair_share - cfg = self._cfg - - # ``fair_share`` / ``_balance_correction`` produce the residual: this - # consumer's slice of the grid imbalance to fold into its current - # output. The absolute net-output target is therefore "what I report - # now plus my residual share" — see the NetOutputW wrap below. + residual = fair_share if ( - not cfg.fair_distribution - or consumer_id is None - or consumer_id not in reports - or concentrated is not None + self._cfg.fair_distribution + and concentrated is None + and consumer_id is not None + and consumer_id in reports + and consumer_id in eff_part ): - residual = fair_share - elif consumer_id in eff_part: residual = self._balance_correction( consumer_id, reports, eff_part, fair_share ) - else: - residual = fair_share - # Clamp only the grid-tracking half (``fair_share``, which carries the - # grid's sign by construction) against the predicted grid direction — - # never the balance-correction term, which is zero-sum across the - # same-phase pool and so grid-neutral; zeroing it too would make - # equalization one-sided near steady state (issue #523). tracking = fair_share if (control_grid < 0 and tracking > 0) or (control_grid > 0 and tracking < 0): tracking = 0.0 - residual = tracking + (residual - fair_share) - - if consumer_id: - residual = self._damp_oscillation(consumer_id, residual) - - reported = _report_of(reports, consumer_id).power if consumer_id else 0 - return self._emit( - consumer_id, - NetOutputW(reported + residual), - reported, - reports, - eff_part, - pace=True, - ) + return tracking + (residual - fair_share) @staticmethod def _charge_blind(reports: Reports, grid_total: float) -> tuple[set[str], bool]: @@ -2211,50 +2242,32 @@ def _damp_oscillation(self, consumer_id: str, residual: float) -> float: state.osc_last_sign = sign return residual * (1.0 - cfg.osc_damp_max * state.osc_score) - def _pace_reading( - self, consumer_id: str, reading: float, reported: float, reports: Reports - ) -> float: - """Clamp the auto-path *reading* to the consumer's ramp-pacing cap. - - The battery integrates the reading with its own accelerating ramp, so - the reading we send is the only bound on its per-poll movement. The - cap starts at ``pace_base_step``, doubles per reference second toward - ``pace_max_step`` only while the battery demonstrably tracks the - command, follows the error back down, and resets to the base step on - direction reversal — bounding stale-feedback overshoot to the battery's - *demonstrated* slew. Caps are W per :data:`PACE_REFERENCE_DT`, scaled - by the observed inter-poll time (clamped at 1.0). - - Paced: the regulation loop, the fade transition and the deprioritized / - charge-blind wind-down (the firmware applies a charge-direction reading - in full in one cycle, so an unpaced wind-down is a one-poll step - disturbance on the rest of the pool). Not paced: probe targets, the - MIN_DC_OUTPUT floor, manual targets and the inactive steer-to-zero. - Callers needing the unpaced intent (issue #376) read ``last_intent``. + def _pace_cap( + self, + state: BalancerConsumerState, + reading: float, + reported: float, + sign: int, + dt_ratio: float, + can_stall: bool, + ) -> tuple[float, bool]: + """Return this consumer's ramp cap in W, and whether it is stalled. + + The learning half of :meth:`_pace_reading`: the cap tracks what the + battery has *demonstrated* it can slew. It resets to the base step on + a direction reversal, doubles per reference second while the battery + visibly follows, and grows against a persistent stall so a device held + under its minimum actionable command is not clamped there forever. + Records what the device responded to (``pace_responded_at``) and how + long it has been unresponsive (``pace_stall_polls``); the caller owns + ``pace_cap`` itself. """ base = self._cfg.pace_base_step - if base <= 0: - return reading - state = self._get_consumer(consumer_id) - now = self._clock() - dt = now - state.pace_last_at if state.pace_last_at > 0.0 else 0.0 - if dt <= 0.0: - # First paced poll, a non-advancing clock, or a backwards jump: - # assume one reference period rather than starving the clamp. - dt = PACE_REFERENCE_DT - state.pace_last_at = now - dt_ratio = min(1.0, dt / PACE_REFERENCE_DT) - sign = 1 if reading > 0 else -1 if reading < 0 else 0 cap = state.pace_cap if state.pace_cap > 0 else base # Never below the base step: hysteresis-style regulators (B2500) need a # minimum reading to clear their input hold window at all. The cadence # scale still bounds the grown cap. limit = max(base, cap * dt_ratio) - # The stall escape and the response floor below apply only to devices - # with a minimum actionable command (the DC-output family); any other - # battery can execute an arbitrarily small command and can never be - # deadlocked by the clamp, so it stays on the unmodified path. - can_stall = _needs_dc_output_floor(_report_of(reports, consumer_id).device_type) stalled = False if sign == 0 or sign != state.pace_sign: cap = base @@ -2303,7 +2316,49 @@ def _pace_reading( # but the else branch back-computes cap as abs(reading) / dt_ratio, # which a fast poll (small dt_ratio) can inflate past the max — and a # later normal-cadence poll would then slew beyond pace_max_step. - cap = min(cap, self._cfg.pace_max_step) + return min(cap, self._cfg.pace_max_step), stalled + + def _pace_reading( + self, consumer_id: str, reading: float, reported: float, reports: Reports + ) -> float: + """Clamp the auto-path *reading* to the consumer's ramp-pacing cap. + + The battery integrates the reading with its own accelerating ramp, so + the reading we send is the only bound on its per-poll movement. + :meth:`_pace_cap` decides how much movement this battery has earned; + this method measures the poll interval, applies that cap, and records + what was sent. Caps are W per :data:`PACE_REFERENCE_DT`, scaled by the + observed inter-poll time (clamped at 1.0). + + Paced: the regulation loop, the fade transition and the deprioritized / + charge-blind wind-down (the firmware applies a charge-direction reading + in full in one cycle, so an unpaced wind-down is a one-poll step + disturbance on the rest of the pool). Not paced: probe targets, the + MIN_DC_OUTPUT floor, manual targets and the inactive steer-to-zero. + Callers needing the unpaced intent (issue #376) read ``last_intent``. + """ + base = self._cfg.pace_base_step + if base <= 0: + return reading + state = self._get_consumer(consumer_id) + now = self._clock() + dt = now - state.pace_last_at if state.pace_last_at > 0.0 else 0.0 + if dt <= 0.0: + # First paced poll, a non-advancing clock, or a backwards jump: + # assume one reference period rather than starving the clamp. + dt = PACE_REFERENCE_DT + state.pace_last_at = now + dt_ratio = min(1.0, dt / PACE_REFERENCE_DT) + sign = 1 if reading > 0 else -1 if reading < 0 else 0 + # The stall escape and the response floor below apply only to devices + # with a minimum actionable command (the DC-output family); any other + # battery can execute an arbitrarily small command and can never be + # deadlocked by the clamp, so it stays on the unmodified path. + can_stall = _needs_dc_output_floor(_report_of(reports, consumer_id).device_type) + + cap, stalled = self._pace_cap( + state, reading, reported, sign, dt_ratio, can_stall + ) state.pace_cap = cap state.pace_sign = sign state.pace_prev_reported = reported @@ -2344,13 +2399,9 @@ def _concentration_pool_balanced( return False actual_total = sum(_report_of(reports, cid).power for cid in conc_ids) weights = {cid: _report_of(reports, cid).weight for cid in conc_ids} - total_weight = sum(weights.values()) for cid in conc_ids: actual_self = _report_of(reports, cid).power - if total_weight > 0: - target_share = actual_total * weights[cid] / total_weight - else: - target_share = actual_total / len(conc_ids) + target_share = weighted_share(actual_total, weights, conc_ids, cid) if abs(target_share - actual_self) >= deadband: return False return True @@ -2376,11 +2427,7 @@ def _balance_correction( # decided by ``eff_part`` above, so a small weight never drops a # healthy battery from the pool. weights = {cid: _report_of(reports, cid).weight for cid in participating} - total_weight = sum(weights.values()) - if total_weight > 0: - target_share = actual_total * weights.get(consumer_id, 0.0) / total_weight - else: - target_share = actual_total / len(participating) + target_share = weighted_share(actual_total, weights, participating, consumer_id) error = target_share - actual_self err_abs = abs(error) if cfg.balance_deadband > 0 and err_abs < cfg.balance_deadband: @@ -2543,14 +2590,8 @@ def _compute_efficiency_deprioritized( deprioritized = set(self._priority[slots:]) result: dict[str, float] = {cid: 0.0 for cid in deprioritized} - final_active = tuple(self._priority[:slots]) - if not probing and previous_active: - promoted = [cid for cid in final_active if cid not in previous_active] - backups = [cid for cid in previous_active if cid not in final_active] - if promoted and backups: - self._begin_probe( - promoted[0], final_active, tuple(backups), previous_active, now - ) + if not probing: + self._probe_active_set_change(previous_active, slots, now) for cid in deprioritized - self._deprioritized: # Symmetric with the promotion clear above: the score is a memory @@ -2559,6 +2600,38 @@ def _compute_efficiency_deprioritized( # the fading window would bar ``_maybe_force_swap_saturated`` from # ever promoting it back. self._forget_saturation(cid) + self._log_role_changes(deprioritized, abs_target, slots) + + self._deprioritized = deprioritized + self._cache_sample = cache_key + self._cache_result = result + return result + + def _probe_active_set_change( + self, previous_active: tuple[str, ...], slots: int, now: float + ) -> None: + """Probe a swap the efficiency pass just made, before trusting it. + + Promoting a battery is a bet that it delivers more than the one it + displaced. When this tick both promotes and demotes, the bet is tested + against the battery it displaced rather than assumed: see + :meth:`_begin_probe`. Nothing to test on the first pass, when there is + no previous active set to compare against. + """ + if not previous_active: + return + final_active = tuple(self._priority[:slots]) + promoted = [cid for cid in final_active if cid not in previous_active] + backups = [cid for cid in previous_active if cid not in final_active] + if promoted and backups: + self._begin_probe( + promoted[0], final_active, tuple(backups), previous_active, now + ) + + def _log_role_changes( + self, deprioritized: set[str], abs_target: float, slots: int + ) -> None: + """Record consumers entering or leaving the deprioritized set.""" for cid, verb in ( *((cid, "deprioritizing") for cid in deprioritized - self._deprioritized), *((cid, "activating") for cid in self._deprioritized - deprioritized), @@ -2571,11 +2644,6 @@ def _compute_efficiency_deprioritized( slots, ) - self._deprioritized = deprioritized - self._cache_sample = cache_key - self._cache_result = result - return result - def _rotate_priority_head( self, reports: Reports, now: float, active_slots: int ) -> None: