Send the alarm time in the given timezone when it is reached.
(websocket, alarm_at, tzinfo)
| 25 | |
| 26 | |
| 27 | async def alarm(websocket, alarm_at, tzinfo): |
| 28 | """Send the alarm time in the given timezone when it is reached.""" |
| 29 | alarm_at = alarm_at.replace(tzinfo=tzinfo) |
| 30 | now = datetime.datetime.now(tz=datetime.timezone.utc) |
| 31 | |
| 32 | try: |
| 33 | async with asyncio.timeout((alarm_at - now).total_seconds()): |
| 34 | await websocket.wait_closed() |
| 35 | except asyncio.TimeoutError: |
| 36 | try: |
| 37 | await websocket.send(alarm_at.isoformat()) |
| 38 | except ConnectionClosed: |
| 39 | return |
| 40 | |
| 41 | |
| 42 | async def timer(websocket, alarm_after): |
nothing calls this directly
no test coverage detected
searching dependent graphs…