feat: Strom, Leistung und Strombudget

Zeigt strip.current_ma und strip.power_w aus system/info als Sensoren
und macht max_milliamps aus system/settings ueber ein Zahlenfeld
einstellbar. Schreiben liest wie bei sACN zurueck statt den
geschriebenen Wert anzunehmen.
This commit is contained in:
Michael Sauer
2026-09-23 14:35:50 +02:00
parent ffcbbdfe4a
commit 616515ef77
10 changed files with 179 additions and 1 deletions
+4
View File
@@ -115,6 +115,10 @@ class NinstripClient:
"""sACN-Einstellungen aendern. Weggelassene Felder behalten ihren Wert."""
await self._request("PUT", "system/settings", {"type": "sacn", **fields})
async def set_strip(self, **fields: Any) -> None:
"""Strip-Einstellungen aendern. Weggelassene Felder behalten ihren Wert."""
await self._request("PUT", "system/settings", {"type": "strip", **fields})
async def reboot(self) -> None:
await self._request("PUT", "system/command", {"command": {"type": "reboot"}})
+23
View File
@@ -183,6 +183,29 @@ class NinstripCoordinator(DataUpdateCoordinator[NinstripData]):
)
)
async def async_set_strip(self, **fields: Any) -> None:
"""Strip-Einstellungen schreiben und die Antwort des Geraets uebernehmen.
Gleiches Muster wie async_set_sacn: das Geraet antwortet nur mit
{"status": "ok"}, also wird zurueckgelesen statt angenommen.
"""
async def schreiben_und_lesen(client: NinstripClient) -> dict[str, Any]:
await client.set_strip(**fields)
return await client.get_settings()
settings = await self.async_call(schreiben_und_lesen)
self._consecutive_poll_failures = 0
self.async_set_updated_data(
NinstripData(
self.data.state,
self.data.info,
self.data.presets,
self.data.segments,
settings,
)
)
async def async_set_all_segments(self, **fields: Any) -> None:
"""Schreibt dieselben Felder in JEDES Segment und liest zurueck.
+27 -1
View File
@@ -10,7 +10,7 @@ from homeassistant.components.number import (
NumberEntityDescription,
NumberMode,
)
from homeassistant.const import EntityCategory, UnitOfTime
from homeassistant.const import EntityCategory, UnitOfElectricCurrent, UnitOfTime
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
@@ -60,6 +60,7 @@ async def async_setup_entry(
entitaeten: list[NumberEntity] = [NinstripZeit(coordinator, d) for d in ZEITEN]
entitaeten.append(NinstripSacnUniverse(coordinator))
entitaeten.append(NinstripSacnStartSlot(coordinator))
entitaeten.append(NinstripStrombudget(coordinator))
async_add_entities(entitaeten)
@@ -141,3 +142,28 @@ class NinstripSacnStartSlot(NinstripSacnEntity, NumberEntity):
async def async_set_native_value(self, value: float) -> None:
await self.coordinator.async_set_sacn(start_slot=int(value))
class NinstripStrombudget(NinstripEntity, NumberEntity):
"""Was das Netzteil hergibt. Die Firmware regelt daran."""
_attr_entity_category = EntityCategory.CONFIG
_attr_translation_key = "max_milliamps"
_attr_native_min_value = 100
# Die Firmware nimmt fuer das Budget hoechstens 65535 mA an (uint16).
_attr_native_max_value = 65535
_attr_native_step = 100
_attr_mode = NumberMode.BOX
_attr_native_unit_of_measurement = UnitOfElectricCurrent.MILLIAMPERE
def __init__(self, coordinator: NinstripCoordinator) -> None:
super().__init__(coordinator, "max_milliamps")
@property
def native_value(self) -> float | None:
strip = self.coordinator.data.settings.get("strip")
wert = strip.get("max_milliamps") if isinstance(strip, dict) else None
return float(wert) if isinstance(wert, (int, float)) else None
async def async_set_native_value(self, value: float) -> None:
await self.coordinator.async_set_strip(max_milliamps=int(value))
+23
View File
@@ -16,7 +16,9 @@ from homeassistant.components.sensor import (
from homeassistant.const import (
SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
EntityCategory,
UnitOfElectricCurrent,
UnitOfInformation,
UnitOfPower,
UnitOfTime,
)
from homeassistant.core import HomeAssistant, callback
@@ -98,6 +100,27 @@ SENSORS: tuple[NinstripSensorDescription, ...] = (
entity_category=EntityCategory.DIAGNOSTIC,
value=lambda d: _strip(d, "power_scale"),
),
# 0 mA heisst entweder "zieht gerade nichts" oder "misst gar nicht" -
# aeltere Firmware ohne Shunt meldet ebenfalls 0. Von aussen nicht zu
# unterscheiden, und keine Heuristik dagegen: Schwaeche der Geraete-API.
NinstripSensorDescription(
key="current",
translation_key="current",
device_class=SensorDeviceClass.CURRENT,
native_unit_of_measurement=UnitOfElectricCurrent.MILLIAMPERE,
state_class=SensorStateClass.MEASUREMENT,
entity_category=EntityCategory.DIAGNOSTIC,
value=lambda d: _strip(d, "current_ma"),
),
NinstripSensorDescription(
key="power",
translation_key="power",
device_class=SensorDeviceClass.POWER,
native_unit_of_measurement=UnitOfPower.WATT,
state_class=SensorStateClass.MEASUREMENT,
entity_category=EntityCategory.DIAGNOSTIC,
value=lambda d: _strip(d, "power_w"),
),
# Was bei der letzten Aktualisierung schiefging - leer, solange nichts
# schiefging. Aus der Ferne der einzige Hinweis darauf.
NinstripSensorDescription(
+9
View File
@@ -105,6 +105,12 @@
"power_scale": {
"name": "Leistungsbegrenzung"
},
"current": {
"name": "Strom"
},
"power": {
"name": "Leistung"
},
"last_ota_error": {
"name": "Letzter Update-Fehler"
},
@@ -135,6 +141,9 @@
},
"transition_default": {
"name": "Übergangszeit (Vorgabe)"
},
"max_milliamps": {
"name": "Strombudget"
}
},
"binary_sensor": {
@@ -105,6 +105,12 @@
"power_scale": {
"name": "Leistungsbegrenzung"
},
"current": {
"name": "Strom"
},
"power": {
"name": "Leistung"
},
"last_ota_error": {
"name": "Letzter Update-Fehler"
},
@@ -135,6 +141,9 @@
},
"transition_default": {
"name": "Übergangszeit (Vorgabe)"
},
"max_milliamps": {
"name": "Strombudget"
}
},
"binary_sensor": {
@@ -105,6 +105,12 @@
"power_scale": {
"name": "Power limiting"
},
"current": {
"name": "Current"
},
"power": {
"name": "Power"
},
"last_ota_error": {
"name": "Last update error"
},
@@ -135,6 +141,9 @@
},
"transition_default": {
"name": "Default transition time"
},
"max_milliamps": {
"name": "Current budget"
}
},
"binary_sensor": {
+7
View File
@@ -101,6 +101,13 @@ def mock_client(
strip_settings.setdefault("sacn", {}).update(fields)
client.set_sacn = AsyncMock(side_effect=set_sacn)
async def set_strip(**fields: Any) -> None:
# Wie set_sacn: das Geraet merkt sich, was es annimmt, und der
# Aufrufer liest den Block danach zurueck.
strip_settings.setdefault("strip", {}).update(fields)
client.set_strip = AsyncMock(side_effect=set_strip)
client.apply_preset = AsyncMock(return_value=None)
client.save_preset = AsyncMock(return_value=None)
client.reboot = AsyncMock(return_value=None)
+66
View File
@@ -0,0 +1,66 @@
"""Strom, Leistung und das Budget."""
from __future__ import annotations
from typing import Any
from unittest.mock import MagicMock
from freezegun.api import FrozenDateTimeFactory
from homeassistant.const import ATTR_ENTITY_ID
from homeassistant.core import HomeAssistant
from pytest_homeassistant_custom_component.common import MockConfigEntry
from tests.helpers import abfragen, entity_id_for
async def test_strom_und_leistung(hass: HomeAssistant, init_integration: MockConfigEntry) -> None:
assert hass.states.get(entity_id_for(hass, "sensor", "current")).state == "2500"
# power_w kommt als Kommazahl von der Firmware (2,5 A bei 12 V).
assert hass.states.get(entity_id_for(hass, "sensor", "power")).state == "30.0"
async def test_budget_zeigt_den_eingestellten_wert(
hass: HomeAssistant, init_integration: MockConfigEntry
) -> None:
entity_id = entity_id_for(hass, "number", "max_milliamps")
assert hass.states.get(entity_id).state == "1800.0"
assert hass.states.get(entity_id).attributes["max"] == 65535
async def test_budget_setzen(
hass: HomeAssistant, init_integration: MockConfigEntry, mock_client: MagicMock
) -> None:
await hass.services.async_call(
"number",
"set_value",
{ATTR_ENTITY_ID: entity_id_for(hass, "number", "max_milliamps"), "value": 9000},
blocking=True,
)
mock_client.set_strip.assert_awaited_once_with(max_milliamps=9000)
# Nicht geraten, sondern zurueckgelesen: das Geraet antwortet nur mit "ok".
assert mock_client.get_settings.await_count >= 2
assert hass.states.get(entity_id_for(hass, "number", "max_milliamps")).state == "9000.0"
async def test_ohne_strip_block_in_settings_bleibt_das_budget_leer(
hass: HomeAssistant,
init_integration: MockConfigEntry,
strip_settings: dict[str, Any],
freezer: FrozenDateTimeFactory,
) -> None:
"""Aeltere Firmware ohne Strommessung meldet unter system/settings keinen strip-Block."""
strip_settings.pop("strip")
await abfragen(hass, freezer)
assert hass.states.get(entity_id_for(hass, "number", "max_milliamps")).state == "unknown"
async def test_ohne_strip_block_in_info_bleiben_strom_und_leistung_leer(
hass: HomeAssistant,
init_integration: MockConfigEntry,
strip_info: dict[str, Any],
freezer: FrozenDateTimeFactory,
) -> None:
strip_info.pop("strip")
await abfragen(hass, freezer)
assert hass.states.get(entity_id_for(hass, "sensor", "current")).state == "unknown"
assert hass.states.get(entity_id_for(hass, "sensor", "power")).state == "unknown"
+2
View File
@@ -45,6 +45,8 @@ INFO_K1Y: dict[str, Any] = {
"inverted": True,
"power_scale": 100,
"max_milliamps": 1800,
"current_ma": 2500,
"power_w": 30.0,
},
"render": {"fps": 50, "render_us": 900, "show_us": 1800, "dropped_frames": 0},
"last_ota_error": None,