All checks were successful
ci/woodpecker/tag/build/5 Pipeline was successful
ci/woodpecker/tag/build/6 Pipeline was successful
ci/woodpecker/tag/build/4 Pipeline was successful
ci/woodpecker/tag/namespace Pipeline was successful
ci/woodpecker/tag/build/1 Pipeline was successful
ci/woodpecker/tag/build/7 Pipeline was successful
ci/woodpecker/tag/config Pipeline was successful
ci/woodpecker/tag/build/3 Pipeline was successful
ci/woodpecker/tag/build/2 Pipeline was successful
ci/woodpecker/tag/deploy/2 Pipeline was successful
ci/woodpecker/tag/deploy/1 Pipeline was successful
ci/woodpecker/tag/deploy/4 Pipeline was successful
ci/woodpecker/tag/deploy/3 Pipeline was successful
ci/woodpecker/tag/deploy/5 Pipeline was successful
ci/woodpecker/tag/deploy/6 Pipeline was successful
ci/woodpecker/tag/ingress Pipeline was successful
59 lines
1.6 KiB
Python
59 lines
1.6 KiB
Python
"""Hottis Wago Modbus vendor transformations."""
|
|
|
|
import logging
|
|
from typing import Any
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
def transform_relay_to_vendor(payload: dict[str, Any]) -> str:
|
|
"""Transform abstract relay payload to Hottis Wago Modbus format.
|
|
|
|
Hottis Wago Modbus expects plain text 'true' or 'false' (not JSON).
|
|
|
|
Example:
|
|
- Abstract: {'power': 'on'}
|
|
- Hottis Wago Modbus: 'true' or 'false'
|
|
"""
|
|
power = payload.get("power", "off")
|
|
|
|
# Map abstract "on"/"off" to vendor "true"/"false"
|
|
if isinstance(power, str):
|
|
power_lower = power.lower()
|
|
if power_lower in {"on", "true", "1"}:
|
|
return "true"
|
|
if power_lower in {"off", "false", "0"}:
|
|
return "false"
|
|
|
|
# Fallback: any truthy value -> "true", else "false"
|
|
return "true" if power else "false"
|
|
|
|
|
|
def transform_relay_to_abstract(payload: str) -> dict[str, Any]:
|
|
"""Transform Hottis Wago Modbus relay payload to abstract format.
|
|
|
|
Hottis Wago Modbus sends plain text 'true' or 'false'.
|
|
|
|
Example:
|
|
- Hottis Wago Modbus: 'true'
|
|
- Abstract: {'power': 'on'}
|
|
"""
|
|
value = payload.strip().lower()
|
|
|
|
if value == "true":
|
|
power = "on"
|
|
elif value == "false":
|
|
power = "off"
|
|
else:
|
|
# Fallback for unexpected values: keep as-is
|
|
logger.warning("Unexpected relay payload from Hottis Wago Modbus: %r", payload)
|
|
power = value
|
|
|
|
return {"power": power}
|
|
|
|
|
|
# Registry of handlers for this vendor
|
|
HANDLERS = {
|
|
("relay", "to_vendor"): transform_relay_to_vendor,
|
|
("relay", "to_abstract"): transform_relay_to_abstract,
|
|
}
|