name: homeassistant-integration-patterns description: Project-specific patterns for the Marstek integration (config flow, coordinator, scanner, entities, translations)
Home Assistant Integration Patterns (Marstek)
This skill helps you make correct, repo-consistent changes to this Home Assistant custom integration.
When to Use
- Adding/changing sensors
- Changing discovery/config flow behavior
- Updating coordinator error handling
- Working on translations or diagnostics
- Ensuring changes meet Home Assistant integration quality expectations
Quick Map
| Task | File(s) |
|---|---|
| Setup / teardown / coordinator wiring | __init__.py |
| Config flow (user, dhcp, integration discovery) | config_flow.py |
| Central polling (tiered intervals) | coordinator.py |
| IP-change scanner | scanner.py |
| Sensors | sensor.py (EntityDescription pattern) |
| Binary sensors | binary_sensor.py (EntityDescription pattern) |
| Select entities | select.py |
| Services | services.py (idempotent registration) |
| Device automation actions | device_action.py |
| Device info helper | device_info.py |
| Mode configuration | mode_config.py |
| Diagnostics | diagnostics.py |
| Text / translations | strings.json, translations/en.json |
| Icons | icons.json |
| Local API reference | docs/marstek_device_openapi.MD |
| UDP client library | pymarstek/ |
Core Rules
Coordinator-only I/O
- Never add per-entity UDP calls.
- Read everything from
MarstekDataUpdateCoordinator.data. - Use
_async_setup()for one-time initialization during first refresh. - Set
always_update=Falseif data supports__eq__comparison.
Async-only
- Only do async I/O; never block the event loop.
Coordinator Error Handling
async def _async_update_data(self):
try:
return await self.api.fetch_data()
except AuthError as err:
# Triggers reauth flow automatically
raise ConfigEntryAuthFailed from err
except RateLimitError:
# Backoff with retry_after
raise UpdateFailed(retry_after=60)
except ConnectionError as err:
raise UpdateFailed(f"Connection failed: {err}")
Avoid unavailable clutter
- Only create entities when there’s a corresponding data key in coordinator output.
- Prefer explicit per-sensor classes or a description table keyed by coordinator data.
Use translation-aware config-flow errors
- Config flow errors should use keys defined in
custom_components/marstek/strings.json. - Reauth flows should ask only for the changed credential and update the existing entry.
- Config flow errors should use keys defined in
Stable identifiers
- Use BLE-MAC-based unique IDs for entities and devices; never pivot on IPs.
- Keep
_attr_has_entity_name = Trueand setdevice_infofor grouping.
Adding a new sensor
Steps:
- Find the value on
coordinator.data(a plaindict[str, Any]coming frompymarstek). - Add a
MarstekSensorEntityDescriptionto theSENSORStuple insensor.py. - Use
exists_fnto conditionally create entities (avoids permanent unavailable state). - Keep the
unique_idstable (BLE-MAC + sensor key). - Add translation keys in
translations/en.json(and keepstrings.jsonin sync). - Use
suggested_display_precisionfor numeric sensors. - Only register entities for data keys that exist to avoid permanent
unavailablenoise.
Entity Patterns (Mandatory for New Integrations)
class MarstekSensor(CoordinatorEntity, SensorEntity):
_attr_has_entity_name = True # MANDATORY
def __init__(self, coordinator, description):
super().__init__(coordinator)
self.entity_description = description
self._attr_unique_id = f"{coordinator.ble_mac}_{description.key}"
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, coordinator.ble_mac)},
name=coordinator.device_name,
manufacturer="Marstek",
)
EntityDescription Pattern (Recommended)
@dataclass(kw_only=True)
class MarstekSensorEntityDescription(SensorEntityDescription):
value_fn: Callable[[dict], StateType]
exists_fn: Callable[[dict], bool] = lambda _: True
SENSORS: tuple[MarstekSensorEntityDescription, ...] = (
MarstekSensorEntityDescription(
key="battery_soc",
translation_key="battery_soc",
device_class=SensorDeviceClass.BATTERY,
native_unit_of_measurement=PERCENTAGE,
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda data: data.get("soc"),
),
MarstekSensorEntityDescription(
key="power",
device_class=SensorDeviceClass.POWER,
native_unit_of_measurement=UnitOfPower.WATT,
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda data: data.get("power"),
),
)
Icon Translations (Preferred over icon property)
Create icons.json:
{
"entity": {
"sensor": {
"battery_soc": {
"default": "mdi:battery",
"state": {
"100": "mdi:battery",
"50": "mdi:battery-50"
}
}
}
}
}
Entity Categories
EntityCategory.DIAGNOSTIC- RSSI, firmware version, temperatureEntityCategory.CONFIG- Settings the user can change- Set
entity_registry_enabled_default = Falsefor rarely-used sensors
State Classes for Energy Sensors
SensorStateClass.MEASUREMENT- Instantaneous values (power, temperature)SensorStateClass.TOTAL- Values that can increase/decrease (net energy)SensorStateClass.TOTAL_INCREASING- Only increases, resets to 0 (lifetime energy)- Use
SensorDeviceClass.ENERGY_STORAGEfor battery capacity (stored Wh)
Common pitfalls
- Polling/discovery storms (too many UDP requests too frequently).
- Doing IP discovery inside setup or coordinator updates (scanner already handles this).
- Sending control commands without pausing polling (causes concurrent UDP traffic and flaky results).
- Breaking unique IDs (must remain stable across upgrades and IP changes).
- Skipping options reload listeners or reauth handling in
config_flow.py. - Using aggressive polling intervals instead of event-driven triggers (prefer immediate scan on failure over shorter intervals).
Event-Driven Best Practices
When detecting connectivity issues or IP changes:
- Prefer event-driven over polling: Trigger immediate action on failure instead of shortening poll intervals
- Debounce: Add minimum intervals between event-triggered actions (e.g., 30s minimum between scans)
- Keep backup polling: Maintain a longer interval (10 min) as a backstop for edge cases
Example: The coordinator triggers MarstekScanner.async_request_scan() when it hits the failure threshold, enabling fast IP change detection without aggressive periodic scanning.