-
-
Notifications
You must be signed in to change notification settings - Fork 0
/
code.py
123 lines (97 loc) · 4.16 KB
/
code.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
from __future__ import annotations
import gc
import math
import os
from adafruit_datetime import datetime, timedelta
from adafruit_pyportal import PyPortal
from skyportal.displaylib import SkyPortalUI
from skyportal.maplib import build_bounding_box
from skyportal.networklib import APIExceptionError, APIHandlerBase, APITimeoutError
try:
from secrets import secrets
except ImportError as e:
raise Exception("Could not locate secrets file.") from e
try:
import skyportal_config
except ImportError as e:
raise Exception("Could not locate configuration file.") from e
def _utc_to_local(utc_timestamp: float, utc_offset: str = "-0000") -> datetime:
"""
Convert the given timestamp into local time with the provided UTC offset.
UTC offset is assumed to be provided as `"±HHMM"`.
"""
hours = int(utc_offset[:3])
minutes = math.copysign(int(utc_offset[-2:]), hours)
delta = timedelta(hours=hours, minutes=minutes)
utc_time = datetime.fromtimestamp(utc_timestamp)
return utc_time + delta
# Starting with CircuitPython v9.0 an "sd" folder has to be created to be used as a mount point
# For newer installs this should already be present but may not if migrating from v8.x
# See: https://github.com/adafruit/circuitpython/issues/8872
# See: https://github.com/adafruit/circuitpython/pull/8860
if "sd" not in os.listdir("/"):
# Our boot.py already makes the filesystem writeable so we shouldn't need to guard
os.mkdir("/sd")
# Device Initialization
PYPORTAL = PyPortal() # This also takes care of mounting the SD to /sd
skyportal_ui = SkyPortalUI()
PYPORTAL.network.connect()
print("Wifi connected")
SESSION = PYPORTAL.network._wifi.requests
# The internal PyPortal query to AIO returns as "%Y-%m-%d %H:%M:%S.%L %j %u %z %Z"
# This method sets the internal clock, but we also retain it to transform the API time to local
init_timestamp = PYPORTAL.get_local_time(location=secrets["timezone"])
utc_offset = init_timestamp.split()[4]
grid_bounds = build_bounding_box()
skyportal_ui.post_connect_init(grid_bounds, SESSION)
api_handler: APIHandlerBase
if skyportal_config.AIRCRAFT_DATA_SOURCE == "adsblol":
from skyportal.networklib import ADSBLol
api_handler = ADSBLol(
request_session=SESSION,
lat=skyportal_config.MAP_CENTER_LAT,
lon=skyportal_config.MAP_CENTER_LON,
radius=skyportal_config.GRID_WIDTH_MI * 2,
)
print("Using ADSB.lol as aircraft data source")
elif skyportal_config.AIRCRAFT_DATA_SOURCE == "opensky":
from skyportal.networklib import OpenSky
api_handler = OpenSky(request_session=SESSION, grid_bounds=grid_bounds)
print("Using OpenSky as aircraft data source")
elif skyportal_config.AIRCRAFT_DATA_SOURCE == "proxy":
from skyportal.networklib import ProxyAPI
api_handler = ProxyAPI(
request_session=SESSION,
lat=skyportal_config.MAP_CENTER_LAT,
lon=skyportal_config.MAP_CENTER_LON,
radius=skyportal_config.GRID_WIDTH_MI * 2,
)
print("Using proxy API as aircraft data source")
else:
raise ValueError(f"Unknown API specified: '{skyportal_config.AIRCRAFT_DATA_SOURCE}'")
gc.collect()
print(f"\n{'='*40}\nInitialization complete\n{'='*40}\n")
# Main loop
skyportal_ui.touch_on()
loop_start_time = datetime.now() - api_handler.refresh_interval # Force first API call
while True:
if (datetime.now() - loop_start_time) >= api_handler.refresh_interval:
skyportal_ui.touch_off()
try:
api_handler.update()
except (APITimeoutError, APIExceptionError) as e:
print(e)
gc.collect()
if api_handler.can_draw:
print("Updating aircraft locations")
skyportal_ui.draw_aircraft(api_handler.aircraft)
skyportal_ui.time_label.text = f"{_utc_to_local(api_handler.api_time, utc_offset)}"
else:
print("No aircraft to draw, skipping redraw")
loop_start_time = datetime.now()
next_request_at = loop_start_time + api_handler.refresh_interval
print(f"Sleeping... next refresh at {next_request_at} local")
skyportal_ui.touch_on()
p = skyportal_ui.touchscreen_handler.touch_point
if p:
skyportal_ui.process_touch(p)