from datetime import timedelta from appdaemon import ADAPI from appdaemon.adbase import ADBase from appdaemon.entity import Entity from appdaemon.models.notification import AndroidData class TrafficAlert(ADAPI): notified: bool = False def initialize(self): self.log(f'Traffic time: {self.traffic_time}') self.traffic_entity.listen_state( self.handle_state_change, attribute='duration', # constrain_state=lambda d: d > 30.0 ) self.handle_state_change(new=self.traffic_time.total_seconds() / 60) @property def traffic_entity(self) -> Entity: return self.get_entity('sensor.work_to_home') @property def traffic_time(self) -> timedelta: return timedelta(minutes=float(self.traffic_entity.get_state('duration'))) def notify_android(self, device: str, **data): model = AndroidData.model_validate(data) res = self.call_service( f'notify/mobile_app_{device}', **model.model_dump()) return res def handle_state_change(self, entity: str = None, attribute: str = None, old: str = None, new: str = None, **kwargs: dict): self.log(f'Travel time changed: {new}') if not self.notified: actions = [ { 'action': 'URI', 'title': 'See travel times', 'uri': "https://grafana.john-stream.com/d/f4ff212e-c786-40eb-b615-205121f482e3/travel-time-details?orgId=1&from=1727927004390&to=1728013404390" } ] if new > 40.0: self.notify_android('pixel_5', **{ 'title': 'Heavy Traffic', 'message': 'Get moving!', 'data': {'tag': 'traffic', 'color': 'red', 'actions': actions}, }) elif new > 30.0: self.notify_android('pixel_5', **{ 'title': 'Increaing Traffic', 'message': 'Something needs to happen', 'data': {'tag': 'traffic', 'color': 'yellow', 'actions': actions}, }) self.notified = True