rollo script for joel
This commit is contained in:
Bastian Schnorbus
@@ -1,98 +1,150 @@
|
||||
# Payload: {
|
||||
# 'trigger_type': 'event',
|
||||
# 'event_type': 'homematic.keypress',
|
||||
# 'context': <homeassistant.core.Context object at 0x7f8c9eebe890>,
|
||||
# 'name': '000B5D898D535E',
|
||||
# 'param': 'PRESS_SHORT',
|
||||
# 'channel': 3
|
||||
# }
|
||||
# =============================================================================
|
||||
# Homematic 6x Multischalter - Rollosteuerung (3 Rollos)
|
||||
# =============================================================================
|
||||
# Datei ablegen unter: config/pyscript/rollo_multischalter.py
|
||||
#
|
||||
# Tastenbelegung pro Rollo:
|
||||
# Linke Taste - kurzer Druck -> Hochfahren / Stopp (Wechsel)
|
||||
# Linke Taste - langer Druck -> Position 70 %
|
||||
# Rechte Taste - kurzer Druck -> Runterfahren / Stopp (Wechsel)
|
||||
# Rechte Taste - langer Druck -> Position 20 %
|
||||
#
|
||||
# no_state_feedback: True -> fuer Rollos die keinen opening/closing Status
|
||||
# zurueckgeben. Der Fahrzustand wird dann intern
|
||||
# im Script getrackt (toggle-Logik per Flag).
|
||||
# Nach TRAVEL_TIMEOUT Sekunden wird der interne
|
||||
# State automatisch auf "stopped" zurueckgesetzt.
|
||||
# =============================================================================
|
||||
|
||||
"""
|
||||
This script is designed to work with a Home Assistant setup using the Homematic IP WRC6 remote control.
|
||||
It listens for keypress events from the remote and performs actions based on the key pressed and the type of press (short or long).
|
||||
# ── Anpassen ──────────────────────────────────────────────────────────────────
|
||||
DEVICE_ADDRESS = "000B5D898D535E" # Homematic-Geräteadresse ← anpassen!
|
||||
|
||||
Documentation https://hacs-pyscript.readthedocs.io/en/latest/overview.html
|
||||
"""
|
||||
ROLLO_CONFIG = [
|
||||
{"entity_id": "cover.00115a498e0439", "subtype_left": 1, "subtype_right": 2, "no_state_feedback": True},
|
||||
]
|
||||
|
||||
TASTER_ID_JOEL = "000B5D898D535E" # Replace with your actual Taster ID
|
||||
ROLLO_ID_JOEL = "cover.00115a498e0439" # Replace with your actual Rollo ID
|
||||
POSITION_UP = 66
|
||||
POSITION_DOWN = 20
|
||||
# Maximale Fahrzeit in Sekunden - danach wird der interne State automatisch
|
||||
# auf "stopped" zurueckgesetzt (Rollo ist dann am Anschlag angekommen)
|
||||
TRAVEL_TIMEOUT = 20
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
last_action = dict()
|
||||
# Lookup-Tabelle: subtype -> (entity_id, Seite, no_state_feedback)
|
||||
_SUBTYPE_MAP = {}
|
||||
for _rollo in ROLLO_CONFIG:
|
||||
_SUBTYPE_MAP[_rollo["subtype_left"]] = (_rollo["entity_id"], "left", _rollo["no_state_feedback"])
|
||||
_SUBTYPE_MAP[_rollo["subtype_right"]] = (_rollo["entity_id"], "right", _rollo["no_state_feedback"])
|
||||
|
||||
state.persist('pyscript.last_action')
|
||||
# Interner Fahrzustand fuer Rollos ohne State-Feedback
|
||||
# Moegliche Werte: "moving_up" | "moving_down" | "stopped"
|
||||
_internal_state = {}
|
||||
|
||||
""" The state of the short-presses will be stored in the pyscript.last_action variable."""
|
||||
""" In the pattern of {key}_{0|1} - 0 = not pressed, 1 = pressed """
|
||||
def toggle_keystate(key):
|
||||
"""Toggle keypress state."""
|
||||
prefixed_key = f'key_{key}'
|
||||
|
||||
if prefixed_key in last_action:
|
||||
last_action[prefixed_key] = 1 if last_action.get(prefixed_key) == 0 else 0
|
||||
else:
|
||||
last_action[prefixed_key] = 1
|
||||
|
||||
log.info(f"Key {key} toggled to {last_action[prefixed_key]}")
|
||||
log.info(f"Key dict {last_action}")
|
||||
# Laufende Timeout-Tasks je Entity - damit ein neuer Befehl den alten Timer abbricht
|
||||
_timeout_tasks = {}
|
||||
|
||||
|
||||
def get_keystate(key):
|
||||
prefixed_key = f'key_{key}'
|
||||
|
||||
return last_action.get(prefixed_key, 0)
|
||||
def _is_moving_internal(entity_id: str) -> bool:
|
||||
"""Gibt True zurueck, wenn das Rollo laut internem Tracking gerade faehrt."""
|
||||
return _internal_state.get(entity_id, "stopped") in ("moving_up", "moving_down")
|
||||
|
||||
|
||||
@event_trigger('homematic.keypress')
|
||||
def taster(**payload):
|
||||
if payload.get("name") != TASTER_ID_JOEL:
|
||||
def _is_moving(entity_id: str) -> bool:
|
||||
"""Gibt True zurueck, wenn das Rollo laut HA gerade faehrt."""
|
||||
current_state = state.get(entity_id)
|
||||
if current_state is None:
|
||||
return False
|
||||
return current_state in ("opening", "closing")
|
||||
|
||||
|
||||
async def _schedule_timeout(entity_id: str):
|
||||
"""Setzt den internen State nach TRAVEL_TIMEOUT Sekunden auf 'stopped'."""
|
||||
await task.sleep(TRAVEL_TIMEOUT)
|
||||
if _internal_state.get(entity_id) in ("moving_up", "moving_down"):
|
||||
log.info(f"[Rollo] Timeout - Fahrt abgeschlossen -> {entity_id}")
|
||||
_internal_state[entity_id] = "stopped"
|
||||
|
||||
|
||||
def _set_moving(entity_id: str, direction: str):
|
||||
"""Setzt internen State und startet den Timeout-Timer."""
|
||||
_internal_state[entity_id] = direction
|
||||
|
||||
# Vorherigen Timer abbrechen falls noch aktiv
|
||||
if entity_id in _timeout_tasks:
|
||||
_timeout_tasks[entity_id].cancel()
|
||||
|
||||
_timeout_tasks[entity_id] = task.create(_schedule_timeout(entity_id))
|
||||
|
||||
|
||||
def _set_stopped(entity_id: str):
|
||||
"""Setzt internen State auf stopped und bricht den Timeout-Timer ab."""
|
||||
_internal_state[entity_id] = "stopped"
|
||||
if entity_id in _timeout_tasks:
|
||||
_timeout_tasks[entity_id].cancel()
|
||||
del _timeout_tasks[entity_id]
|
||||
|
||||
|
||||
@event_trigger("homematic.keypress")
|
||||
def handle_keypress(**kwargs):
|
||||
"""Wertet Homematic-Keypressereignisse aus."""
|
||||
address = kwargs.get("address", "")
|
||||
subtype = kwargs.get("subtype")
|
||||
action = kwargs.get("type", "") # "press_short" | "press_long"
|
||||
|
||||
# Nur Events fuer unseren Schalter verarbeiten
|
||||
if address != DEVICE_ADDRESS:
|
||||
return
|
||||
|
||||
param = payload.get("param")
|
||||
if param == "PRESS_SHORT":
|
||||
# Check if the event is a short press
|
||||
log.info("Short press detected")
|
||||
|
||||
channel_no = payload.get("channel")
|
||||
if subtype not in _SUBTYPE_MAP:
|
||||
return
|
||||
|
||||
""" Offen """
|
||||
if channel_no == 5:
|
||||
# Perform action for channel 5
|
||||
if get_keystate("5") == 1:
|
||||
cover.open_cover(entity_id=ROLLO_ID_JOEL)
|
||||
else:
|
||||
cover.stop_cover(entity_id=ROLLO_ID_JOEL)
|
||||
entity_id, side, no_state_feedback = _SUBTYPE_MAP[subtype]
|
||||
|
||||
toggle_keystate("5")
|
||||
elif channel_no == 6:
|
||||
# Perform action for channel 6
|
||||
if get_keystate("6") == 1:
|
||||
cover.close_cover(entity_id=ROLLO_ID_JOEL)
|
||||
else:
|
||||
cover.stop_cover(entity_id=ROLLO_ID_JOEL)
|
||||
|
||||
toggle_keystate("6")
|
||||
|
||||
elif channel_no == 3:
|
||||
# Perform action for channel 3
|
||||
log.info("Channel 3 action")
|
||||
# Add your action here, e.g., toggle a light
|
||||
|
||||
if param == "PRESS_LONG":
|
||||
# Check if the event is a long press
|
||||
log.info("Long press detected")
|
||||
|
||||
channel_no = payload.get("channel")
|
||||
|
||||
if channel_no == 5:
|
||||
log.info("Channel 5 action")
|
||||
cover.set_cover_position(entity_id=ROLLO_ID_JOEL, position=POSITION_UP)
|
||||
elif channel_no == 6:
|
||||
log.info("Channel 6 action")
|
||||
cover.set_cover_position(entity_id=ROLLO_ID_JOEL, position=POSITION_DOWN)
|
||||
if side == "left":
|
||||
if action == "press_short":
|
||||
_toggle_up(entity_id, no_state_feedback)
|
||||
elif action == "press_long":
|
||||
_set_position(entity_id, 70)
|
||||
if no_state_feedback:
|
||||
_set_stopped(entity_id)
|
||||
elif side == "right":
|
||||
if action == "press_short":
|
||||
_toggle_down(entity_id, no_state_feedback)
|
||||
elif action == "press_long":
|
||||
_set_position(entity_id, 20)
|
||||
if no_state_feedback:
|
||||
_set_stopped(entity_id)
|
||||
|
||||
|
||||
log.info(f"taster: {payload}")
|
||||
|
||||
|
||||
def _toggle_up(entity_id: str, no_state_feedback: bool):
|
||||
"""Kurzer Druck links: Hochfahren - oder Stopp, wenn das Rollo faehrt."""
|
||||
moving = _is_moving_internal(entity_id) if no_state_feedback else _is_moving(entity_id)
|
||||
if moving:
|
||||
log.info(f"[Rollo] Stopp (war in Bewegung) -> {entity_id}")
|
||||
cover.stop_cover(entity_id=entity_id)
|
||||
if no_state_feedback:
|
||||
_set_stopped(entity_id)
|
||||
else:
|
||||
log.info(f"[Rollo] Hochfahren -> {entity_id}")
|
||||
cover.open_cover(entity_id=entity_id)
|
||||
if no_state_feedback:
|
||||
_set_moving(entity_id, "moving_up")
|
||||
|
||||
|
||||
def _toggle_down(entity_id: str, no_state_feedback: bool):
|
||||
"""Kurzer Druck rechts: Runterfahren - oder Stopp, wenn das Rollo faehrt."""
|
||||
moving = _is_moving_internal(entity_id) if no_state_feedback else _is_moving(entity_id)
|
||||
if moving:
|
||||
log.info(f"[Rollo] Stopp (war in Bewegung) -> {entity_id}")
|
||||
cover.stop_cover(entity_id=entity_id)
|
||||
if no_state_feedback:
|
||||
_set_stopped(entity_id)
|
||||
else:
|
||||
log.info(f"[Rollo] Runterfahren -> {entity_id}")
|
||||
cover.close_cover(entity_id=entity_id)
|
||||
if no_state_feedback:
|
||||
_set_moving(entity_id, "moving_down")
|
||||
|
||||
|
||||
def _set_position(entity_id: str, position: int):
|
||||
"""Faehrt das Rollo auf eine bestimmte Position (0 = ganz zu, 100 = ganz auf)."""
|
||||
log.info(f"[Rollo] Zielposition {position}% -> {entity_id}")
|
||||
cover.set_cover_position(entity_id=entity_id, position=position)
|
||||
Reference in New Issue
Block a user