101 lines
3.5 KiB
Python
101 lines
3.5 KiB
Python
|
|
|
|
|
|
from datetime import datetime
|
|
from typing import Any
|
|
|
|
from appdaemon import adbase as ad
|
|
from appdaemon import utils
|
|
from appdaemon.plugins.hass import Hass
|
|
|
|
# from globals import GLOBAL_MODE, GLOBAL_VAR
|
|
|
|
|
|
class SimpleApp(Hass):
|
|
def initialize(self):
|
|
match self.ping():
|
|
case float() as ping:
|
|
ping = utils.format_timedelta(ping)
|
|
self.log(f'{self.__class__.__name__} Initialized: {ping}')
|
|
case _:
|
|
pass
|
|
|
|
# self.get_entity('input_button.test_button').listen_state(self.handle_button)
|
|
|
|
# self.set_app_pin(True)
|
|
|
|
# Listen for a button press event with a specific entity_id
|
|
self.listen_event(
|
|
self.handle_button,
|
|
'call_service',
|
|
service='press',
|
|
entity_id='input_button.test_button'
|
|
)
|
|
|
|
def handle_button(self, event_type: str, data: dict[str, Any], **kwargs: Any) -> None:
|
|
match data:
|
|
case {"service_data": {"entity_id": eid}}:
|
|
friendly_name = self.get_state(eid, attribute='friendly_name')
|
|
self.log(f'pressed {friendly_name}')
|
|
case _:
|
|
self.log(f'Unhandled button press: {data}', level='WARNING')
|
|
|
|
def delayed(self, **kwargs):
|
|
self.log(f'This is a delayed log message: {kwargs}')
|
|
# self.reload_apps()
|
|
# self.stop_app('child_app')
|
|
|
|
def my_callback(self, event_type: str, data: dict[str, Any], **kwargs: Any) -> None:
|
|
match data:
|
|
case {
|
|
"service_data": {"entity_id": eid},
|
|
"metadata": {"time_fired": time_fired}
|
|
}:
|
|
friendly_name = self.get_state(eid, attribute='friendly_name')
|
|
time_fired = datetime.fromisoformat(time_fired).astimezone(self.AD.tz)
|
|
fmt = "%I:%M:%S %p"
|
|
self.log(f'{friendly_name} was pressed at {time_fired.strftime(fmt)}')
|
|
self.log(f'Kwargs: {kwargs}')
|
|
case _:
|
|
self.log(f'Unhandled button press: {data}', level='WARNING')
|
|
|
|
@ad.global_lock
|
|
def write(self):
|
|
return
|
|
|
|
# def test_event_handler(self, event_type: str, data: dict[str, Any], **kwargs: Any) -> None:
|
|
def test_event_handler(self, *_, value_at_listen, **kwargs: Any) -> None:
|
|
# self.log(f' {event_type} '.center(30, '#'))
|
|
self.log(f'Data from event: {value_at_listen}')
|
|
# self.log(f'Data from registration: {kwargs}')
|
|
return
|
|
|
|
|
|
class BaseApp(ad.ADBase):
|
|
def initialize(self):
|
|
self.adapi = self.get_ad_api()
|
|
self.log = self.adapi.log
|
|
self.hassapi = self.get_plugin_api("HASS")
|
|
assert isinstance(self.hassapi, Hass)
|
|
self.log(f'{self.__class__.__name__} Initialized')
|
|
self.config_model = {'name': self.__class__.__name__, 'module': 'simple_app', 'class': 'SimpleApp'}
|
|
|
|
self.global_vars['abc'] = 123
|
|
self.config
|
|
self.app_config
|
|
# self.log(f'Global mode is set to: {GLOBAL_MODE} ({GLOBAL_MODE.value})')
|
|
self.ad = self.AD
|
|
|
|
# self.adapi.run_in(self.delayed, 0.75)
|
|
# self.alexa_notify("Home assistant has been restarted", target="fake")
|
|
|
|
self.adapi.listen_event(
|
|
self.handle_button,
|
|
'call_service',
|
|
service='press',
|
|
entity_id='input_button.test_button'
|
|
)
|
|
|
|
def handle_button(self, event_type: str, data: dict[str, Any], **kwargs: Any) -> None:
|
|
self.adapi.fire_event("test_event", values_at_fire=123)
|