65 lines
2.3 KiB
Python
65 lines
2.3 KiB
Python
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
|