AgentSkillsCN

marstek-integration-development

针对 Marstek 集成的仓库特定指导——以符合 Home Assistant 的方式,实现 UDP Open API、轮询、发现/扫描等功能。

SKILL.md
--- frontmatter
name: marstek-integration-development
description: Repo-specific guidance for maintaining the Marstek integration (UDP Open API, polling, discovery/scanner) in a Home Assistant-friendly way

Marstek Integration Development

This skill focuses on changes that must preserve the intent of this repository: a robust local UDP Open API integration for Marstek energy storage devices, while remaining safe and compatible with Home Assistant.

When to Use

  • You want to tweak polling behavior, discovery, or performance
  • You’re modifying unique IDs / device registry behavior
  • You’re changing scanner/config flow IP-update behavior
  • You need to align changes with Home Assistant integration quality practices

Project Intent

  • The integration is local (no cloud dependency) and uses Marstek’s OPEN API over UDP.
  • Discovery is broadcast-based; IP changes are handled automatically.
  • The domain is marstek.

Polling Interval Policy

Polling is tiered to reduce device load (configurable via options flow):

TierDefaultCommands
Fast30sES.GetMode, ES.GetStatus, EM.GetStatus (real-time power)
Medium60sPV.GetStatus (solar data, Venus A/D only)
Slow300sWifi.GetStatus, Bat.GetStatus (diagnostics)

Request delay: 5 seconds between API calls during a polling cycle (device stability).

The IP-change scanner runs separately (scanner.py, 10 min interval as backup).

Avoid shortening intervals without validating device/network stability. The device can be sensitive to request bursts.

Event-Driven Scanner Pattern

The scanner uses an event-driven approach for IP change detection:

  1. Periodic backup (10 min): Scanner runs discovery every 10 minutes
  2. Event-driven trigger: When coordinator hits failure threshold, it calls MarstekScanner.async_request_scan() to trigger immediate discovery
  3. Debounce: Minimum 30 seconds between event-triggered scans

This pattern avoids aggressive polling while still detecting IP changes quickly (~30s after connection failure).

python
# In coordinator.py - on failure threshold:
if self.consecutive_failures >= failure_threshold:
    scanner = MarstekScanner.async_get(self.hass)
    scanner.async_request_scan()  # Immediate scan, debounced
    raise UpdateFailed(...)

Best practice: Prefer event-driven triggers over shortening scan intervals.

Config / Options / Reauth

  • Keep setup, options, and reauth flows in config_flow.py; prefer selectors over raw text where it improves UX.
  • Abort duplicates: ensure only one entry per device (BLE-MAC/host) and update entries when IP changes via scanner signals.
  • Options updates should register async_on_unload(entry.add_update_listener(...)) to reload on change.
  • Reauth flows should request only the credential that changed and update the existing entry.

Device Identification & Stability

  • Unique IDs and device identifiers should stay stable across IP changes.
  • Prefer BLE-MAC (then WiFi MAC) as the device identifier; IP addresses can change.
  • Keep identifiers stable; changing them causes device/entity duplication for users.

Adding New User-Facing Text

  • Add text via translations/en.json and keep keys stable.
  • Keep custom_components/marstek/strings.json and custom_components/marstek/translations/en.json in sync.

Local Dev / Smoke Testing

  • Use the VS Code task “Start Home Assistant” to run a dev instance.
  • Watch logs for the marstek logger.
  • Run pytest with pytest-homeassistant-custom-component; keep manifests hassfest/HACS-clean and requirements pinned.

Verification After Changes (MANDATORY)

After every code modification, run both checks:

bash
# 1. Type checking (strict mode)
python3 -m mypy --strict custom_components/marstek/

# 2. All tests
pytest tests/ -q

Both must pass before considering any change complete. The repository enforces strict typing with mypy --strict:

  • All functions need return type annotations
  • All parameters need type annotations
  • Generic types need explicit parameters (dict[str, Any], Future[None])
  • Use cast() for type narrowing with mocked objects

Common “gotchas”

  • Sending extra UDP requests per update loop can overload the device.
  • Doing discovery inside setup or coordinator updates can race with the scanner and cause flapping.
  • Returning unstable unique IDs (e.g., IP-based) causes duplication after DHCP changes.