99from __future__ import annotations
1010
1111import asyncio
12+ from asyncio import CancelledError
13+ from collections .abc import Callable
14+ from dataclasses import dataclass
1215import re
1316
14- from ..models import CapabilityChoice , WriteCapability
15- from ..payload .pi30 import Pi30Error
17+ from ..models import CapabilityChoice , DetectedInverter , WriteCapability
18+ from ..payload .pi30 import Pi30Error , parse_qpiri
1619
1720
1821_CONTROLS = (
1922 ("max_charging_current" , "QMCHGCR" , "MCHGC" , "Max Total Charge Current" ),
2023 ("max_ac_charging_current" , "QMUCHGCR" , "MUCHGC" , "Max Utility Charge Current" ),
2124)
2225
26+ _ACQUISITION_STATE = "pi30_charge_current_acquisition"
27+ _ACQUISITION_TIMEOUT = 3.0
28+ _ACQUISITION_MAX_ATTEMPTS = 3
29+
30+
31+ async def _request_parsed (
32+ session , command : str , parser : Callable , * , timeout : float ,
33+ command_timings : list [tuple [str , int , str ]] | None ,
34+ ):
35+ """Measure request AND parse; an invalid payload is never a success."""
36+
37+ loop = asyncio .get_running_loop ()
38+ started = loop .time ()
39+ outcome = "error"
40+ try :
41+ value = parser (await asyncio .wait_for (session .request (command ), timeout = timeout ))
42+ outcome = "ok"
43+ return value
44+ except TimeoutError :
45+ outcome = "timeout"
46+ raise
47+ except Pi30Error as exc :
48+ # Pi30Session normalizes a transport timeout into this protocol error.
49+ outcome = "timeout" if str (exc ) == "request_timeout" else "error"
50+ raise
51+ except CancelledError :
52+ outcome = "cancelled"
53+ raise
54+ finally :
55+ if command_timings is not None :
56+ command_timings .append ((command , int (round ((loop .time () - started ) * 1000 )), outcome ))
57+
58+
59+ @dataclass (slots = True )
60+ class _Acquisition :
61+ binding : tuple
62+ attempts : int = 0
63+ complete : bool = False
64+ in_progress : bool = False
65+
66+
67+ def _binding (inverter : DetectedInverter ) -> tuple :
68+ return (
69+ inverter .driver_key , inverter .protocol_family , inverter .variant_key ,
70+ inverter .serial_number , inverter .probe_target ,
71+ inverter .profile_name , inverter .register_schema_name ,
72+ )
73+
74+
75+ def charge_current_acquisition_diagnostics (runtime_state : dict | None ) -> dict :
76+ state = (runtime_state or {}).get (_ACQUISITION_STATE )
77+ if not isinstance (state , _Acquisition ):
78+ return {}
79+ return {
80+ "pi30_charge_current_acquisition_attempts" : state .attempts ,
81+ "pi30_charge_current_acquisition_status" : (
82+ "complete" if state .complete else
83+ "exhausted" if state .attempts >= _ACQUISITION_MAX_ATTEMPTS else "pending"
84+ ),
85+ }
86+
87+
88+ async def async_acquire_charge_current_controls (
89+ session , inverter : DetectedInverter , * , runtime_state : dict | None ,
90+ command_timings : list [tuple [str , int , str ]] | None = None ,
91+ ) -> dict :
92+ """Finish optional metadata acquisition on the bound runtime session.
93+
94+ Detection is allowed to run out of optional budget. Saved bindings also
95+ arrive without advertised choices. Neither case is a negative capability
96+ verdict: use up to three separate, bounded acquisition windows, then stop
97+ for this session. The hub clears runtime_state on session changes. Never
98+ persist attempts or infer current readback from cached identity details.
99+ """
100+
101+ if runtime_state is None or inverter .driver_key != "pi30" :
102+ return {}
103+ binding = _binding (inverter )
104+ state = runtime_state .get (_ACQUISITION_STATE )
105+ if not isinstance (state , _Acquisition ) or state .binding != binding :
106+ state = _Acquisition (binding )
107+ runtime_state [_ACQUISITION_STATE ] = state
108+ if state .complete or state .in_progress or state .attempts >= _ACQUISITION_MAX_ATTEMPTS :
109+ return {}
110+ capabilities = inverter .capabilities
111+ existing = {cap .key for cap in capabilities }
112+ if all (key in existing for key , * _ in _CONTROLS ):
113+ state .complete = True
114+ return {}
115+
116+ state .attempts += 1
117+ state .in_progress = True
118+ loop = asyncio .get_running_loop ()
119+ deadline = loop .time () + _ACQUISITION_TIMEOUT
120+
121+ def is_current () -> bool :
122+ return (
123+ runtime_state .get (_ACQUISITION_STATE ) is state
124+ and _binding (inverter ) == binding
125+ and inverter .capabilities is capabilities
126+ )
127+
128+ try :
129+ try :
130+ current = await _request_parsed (
131+ session , "QPIRI" , parse_qpiri , timeout = min (1.5 , _ACQUISITION_TIMEOUT ),
132+ command_timings = command_timings ,
133+ )
134+ except Exception : # Optional metadata failure; cancellation propagates.
135+ return {}
136+ if not is_current ():
137+ return {}
138+ enriched = await async_enrich_charge_current_controls (
139+ session , current , capabilities , timeout = max (0.0 , deadline - loop .time ()),
140+ is_current = is_current ,
141+ command_timings = command_timings ,
142+ )
143+ # No stale completion may publish across a reset/rebind or overwrite a
144+ # concurrently changed capability set. There is no await after this gate.
145+ if not is_current ():
146+ return {}
147+ values = {
148+ cap .value_key : cap .enum_value_map [current [cap .value_key ]]
149+ for cap in enriched if cap .key not in existing
150+ }
151+ inverter .capabilities = enriched
152+ inverter .details .update (values )
153+ state .complete = all (key in {cap .key for cap in enriched } for key , * _ in _CONTROLS )
154+ return values
155+ finally :
156+ state .in_progress = False
157+
23158
24159def parse_charge_current_choices (payload : str ) -> tuple [int , ...]:
25160 """Accept a bounded list of distinct three-digit positive ampere values."""
@@ -34,28 +169,41 @@ def parse_charge_current_choices(payload: str) -> tuple[int, ...]:
34169
35170async def async_enrich_charge_current_controls (
36171 session , values : dict , capabilities : tuple [WriteCapability , ...], * , timeout : float ,
172+ is_current : Callable [[], bool ] | None = None ,
173+ command_timings : list [tuple [str , int , str ]] | None = None ,
37174) -> tuple [WriteCapability , ...]:
38175 """Read at most two missing lists within a shared optional time budget.
39176
40177 A timeout/NAK/malformed list never invalidates an already detected inverter.
41- This is not part of the signature scan or regular polling. Writes remain
42- untested/Full-Control-only and accept only the device's advertised values.
178+ Used by best-effort detection and finite post-binding acquisition, not the
179+ recurring telemetry command set. Writes remain untested/Full-Control-only
180+ and accept only the device's advertised values.
43181 """
44182
45183 result = list (capabilities )
46184 existing = {cap .key for cap in capabilities }
47185 loop = asyncio .get_running_loop ()
48186 deadline = loop .time () + max (0.0 , timeout )
49- for index , (key , query , command , title ) in enumerate (_CONTROLS ):
50- current = values .get (key )
51- if key in existing or type (current ) is not int or current <= 0 :
52- continue
187+ missing = [
188+ (index , control ) for index , control in enumerate (_CONTROLS )
189+ if control [0 ] not in existing
190+ and type (values .get (control [0 ])) is int and values [control [0 ]] > 0
191+ ]
192+ for position , (index , (key , query , command , title )) in enumerate (missing ):
193+ if is_current is not None and not is_current ():
194+ break
195+ current = values [key ]
53196 remaining = deadline - loop .time ()
54197 if remaining <= 0 :
55198 break
56199 try :
57- payload = await asyncio .wait_for (session .request (query ), timeout = min (1.5 , remaining ))
58- choices = parse_charge_current_choices (payload )
200+ # Reserve a fair share for every missing list: a slow/unsupported
201+ # total-current query must not starve the utility-current query.
202+ choices = await _request_parsed (
203+ session , query , parse_charge_current_choices ,
204+ timeout = min (1.5 , remaining / (len (missing ) - position )),
205+ command_timings = command_timings ,
206+ )
59207 except Exception : # Optional query/parser failure; cancellation is BaseException.
60208 continue
61209 if current not in choices :
0 commit comments