added virtual buttons
This commit is contained in:
@@ -1,7 +1,10 @@
|
||||
import json
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime
|
||||
from typing import TYPE_CHECKING, Any, Dict
|
||||
|
||||
from appdaemon.entity import Entity
|
||||
|
||||
from . import console
|
||||
|
||||
if TYPE_CHECKING:
|
||||
@@ -23,17 +26,58 @@ class Button:
|
||||
namespace='mqtt',
|
||||
button=self.button_name,
|
||||
)
|
||||
self.create_button_entity()
|
||||
self.logger.info(f'MQTT topic [topic]{topic}[/] controls [room]{self.adapi.name}[/]')
|
||||
|
||||
@property
|
||||
def eid(self) -> str:
|
||||
return f'input_button.{self.adapi.name}'
|
||||
|
||||
@property
|
||||
def virtual_entity(self) -> Entity:
|
||||
if not self.adapi.entity_exists(self.eid):
|
||||
self.adapi.set_state(self.eid, state=datetime.now())
|
||||
|
||||
return self.adapi.get_entity(self.eid)
|
||||
|
||||
def create_button_entity(self):
|
||||
self.adapi.listen_event(
|
||||
self.handle_virtual_button, domain='input_button', service='press', entity_id=self.eid
|
||||
)
|
||||
self.logger.info(f'Listening for button press service call on [purple]{self.eid}[/]')
|
||||
|
||||
def handle_virtual_button(
|
||||
self, event_name: str, data: Dict[str, Any], **kwargs: Dict[str, Any]
|
||||
):
|
||||
if event_name == 'appd_started':
|
||||
return
|
||||
elif event_name == 'state_changed':
|
||||
return
|
||||
|
||||
try:
|
||||
if data['service_data']['entity_id'] == self.eid:
|
||||
self.logger.info(f'Virtual button press: {event_name}')
|
||||
# self.virtual_entity.set_state(state=datetime.now())
|
||||
self.virtual_entity.set_state(state=self.adapi.get_now())
|
||||
self.do_action('single')
|
||||
except KeyError as e:
|
||||
self.logger.error(f'Bad data from {event_name}: {json.dumps(data, indent=4)}')
|
||||
|
||||
def handle_button(self, event_name: str, data: Dict[str, Any], **kwargs: Dict[str, Any]):
|
||||
if event_name == 'appd_started':
|
||||
return
|
||||
|
||||
self.logger.info(f'Button callback: {event_name}, {data}')
|
||||
try:
|
||||
payload = json.loads(data['payload'])
|
||||
action = payload['action']
|
||||
except json.JSONDecodeError:
|
||||
self.logger.error(f'Error decoding JSON from {data["payload"]}')
|
||||
except KeyError:
|
||||
return
|
||||
else:
|
||||
self.do_action(action)
|
||||
|
||||
def do_action(self, action: str):
|
||||
"""Action can be single, double, or others"""
|
||||
if isinstance(action, str) and action != '':
|
||||
self.logger.info(f'Action: [yellow]{action}[/]')
|
||||
if action == 'single':
|
||||
|
||||
Reference in New Issue
Block a user