Repository navigation
Kostal Piko Bat - #3298
Kostal Piko Bat#3298
Conversation
There was a problem hiding this comment.
Pull request overview
Adds a new battery (“Speichermodul”) component for the Kostal Piko device module so the system can read battery power/SoC via the Piko JSON API and expose it as a bat component alongside existing counter and inverter components.
Changes:
- Register a new
batcomponent type in the Kostal Piko device factory. - Extend Kostal Piko config with
KostalPikoBatSetup/ configuration. - Introduce
KostalPikoBatimplementation fetching values from/api/dxs.jsonand writingBatState.
Reviewed changes
Copilot reviewed 3 out of 3 changed files in this pull request and generated 2 comments.
| File | Description |
|---|---|
packages/modules/devices/kostal/kostal_piko/device.py |
Wires bat into ComponentFactoryByType so the device can instantiate/update the new battery component. |
packages/modules/devices/kostal/kostal_piko/config.py |
Adds configuration/setup classes for the new battery component type. |
packages/modules/devices/kostal/kostal_piko/bat.py |
Implements battery polling, sim-counting, peak filtering, and state storage for Kostal Piko battery metrics. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| def get_values(self) -> Tuple[float, List[float]]: | ||
| # Bat Current, Bat Voltage, Bat SoC | ||
| params = (('dxsEntries', ['33556225', '33556226', '33556229']),) | ||
| resp = req.get_http_session().get('http://'+self.ip_address+'/api/dxs.json', | ||
| params=params, | ||
| timeout=3).json()["dxsEntries"] | ||
| power = float(resp[0]["value"]) * float(resp[1]["value"]) | ||
| soc = float(resp[2]["value"]) | ||
| return power, soc |
There was a problem hiding this comment.
get_values is annotated to return Tuple[float, List[float]], but it actually returns (power, soc) where soc is a float. This mismatch makes the API misleading and will be flagged by type checkers; adjust the return type (and related imports) to match the real return value.
| def get_values(self) -> Tuple[float, List[float]]: | ||
| # Bat Current, Bat Voltage, Bat SoC | ||
| params = (('dxsEntries', ['33556225', '33556226', '33556229']),) | ||
| resp = req.get_http_session().get('http://'+self.ip_address+'/api/dxs.json', | ||
| params=params, | ||
| timeout=3).json()["dxsEntries"] | ||
| power = float(resp[0]["value"]) * float(resp[1]["value"]) | ||
| soc = float(resp[2]["value"]) | ||
| return power, soc | ||
|
|
||
| def update(self): | ||
| power, soc = self.get_values() | ||
|
|
||
| self.peak_filter.check_values(power) | ||
| imported, exported = self.sim_counter.sim_count(power) | ||
| bat_state = BatState( | ||
| imported=imported, | ||
| exported=exported, | ||
| power=power, | ||
| soc=soc | ||
| ) | ||
| self.store.set(bat_state) |
There was a problem hiding this comment.
New battery integration adds parsing of /api/dxs.json responses and calculation of power/SoC, but there are no unit tests covering the JSON parsing and sign/units behavior. Add a pytest similar to other device battery tests (mocking the HTTP response and SimCounter) to lock in expected BatState.power and BatState.soc values.
|
@ndrsnhs Bitte noch den TypeHint korrigieren. |
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 3 out of 3 changed files in this pull request and generated 1 comment.
Comments suppressed due to low confidence (1)
packages/modules/devices/kostal/kostal_piko/bat.py:16
log = logging.getLogger(__name__)is defined but not used anywhere in this module. Consider removing it (and possibly theloggingimport) unless you plan to add logging here.
log = logging.getLogger(__name__)
| @@ -0,0 +1,61 @@ | |||
| #!/usr/bin/env python3 | |||
| import logging | |||
| from typing import Any, List, Tuple, TypedDict | |||
| def get_values(self) -> Tuple[float, float]: | ||
| # Bat Current, Bat Voltage, Bat SoC | ||
| params = (('dxsEntries', ['33556225', '33556226', '33556229']),) | ||
| resp = req.get_http_session().get('http://'+self.ip_address+'/api/dxs.json', | ||
| params=params, | ||
| timeout=3).json()["dxsEntries"] | ||
| power = float(resp[0]["value"]) * float(resp[1]["value"]) | ||
| soc = float(resp[2]["value"]) | ||
| return power, soc |
* add bat module * fix type * adjust return type * remove unused imports
Ticket #66003061
Speichermodul für den Kostal Piko