Installation is quite simple.
1) mkdir ~/domoticz/plugins/AirPlaneTracker
2) Copy the code to ~/domoticz/plugins/AirPlaneTracker/plugin.py
3) do a 'sudo systemctl restart domoticz'
4) Goto hardware and select the plugin 'Airplane Tracker', give it a name (adapt settings) and click on 'Add'
5) Goto the 'Utility'tab, find the Airplanes - Counter, click edit and change the counter Type from Energy to Custom
5) Off you go
The names of the sensors are made up of 2 parts.
- The name you give to the plugin @install
- the name of the sensor
Example: I gave the plugin the name Vliegtuigen so the name of the Tracker sensor is Vliegtuigen - Tracker
The sensors are:
Tracker - displays the planes flying over your head
Counter - counts the planes flying over your head
Types - Counts the types of planes flying over your head
When you click on the CallSign of an plane like KLM1844 or RYR9HN or BAW979 in de Tracker sensor, a new Airplanes.live tab opens in your browser and shows the plane on the map. This is a one day score within 9 miles from my house.
Code: Select all
# Airplane Tracker Plugin for Domoticz
# Author: Hein
"""
<plugin key="AirplaneTracker" name="Airplane Tracker" author="Hein" version="2.28.1">
<description>
<h2>Airplane Tracker</h2><br/>
<p>Monitor live air traffic around your location using the airplanes.live API.</p>
<br/>
<h3>Features</h3>
<ul>
<li><b>Tracker</b>: Shows the last 3 aircraft detected with details (altitude, speed, direction)</li>
<li><b>Counter</b>: Cumulative count of all aircraft seen (resets daily by Domoticz)</li>
<li><b>Types</b>: Daily summary of aircraft types detected</li>
</ul>
<br/>
<h3>Configuration</h3>
<ul>
<li><b>Radius</b>: Detection radius in miles around your Domoticz location</li>
<li><b>API Interval</b>: Time between API calls in seconds (default: 60).</li>
<li><b>Log Unknown Types</b>: Logs unclassified types to ~/unknown_aircraft_types.log.</li>
<li><b>Log Level</b>: Controls logging verbosity.</li>
<li><b>Important - Counter Device</b>: After creation, go to <b>Devices</b>, click <b>Edit</b> on the Counter, and change <b>Type</b> to <b>Custom</b>.</li>
</ul>
</description>
<params>
<param field="Mode1" label="Radius (Miles)" width="75px" required="true" default="9"/>
<param field="Mode2" label="API Call Interval (Seconds)" width="75px" required="true" default="60"/>
<param field="Mode3" label="Log Unknown Types" width="75px" required="true" default="False">
<options>
<option label="No" value="False" default="true"/>
<option label="Yes" value="True"/>
</options>
</param>
<param field="Mode6" label="Log Level" width="150px" required="true" default="Error">
<options>
<option label="Error" value="Error" default="true"/>
<option label="Status" value="Status"/>
<option label="Info" value="Info"/>
</options>
</param>
</params>
</plugin>
"""
import Domoticz
import urllib.request, json, socket, time, math, os
from datetime import datetime
class BasePlugin:
def __init__(self):
self.last_seen_ts = {}
self.tracker_buffer = {}
self.type_counts = {}
self.logged_unknowns = set()
self.today = datetime.now().strftime('%Y-%m-%d')
self.total_count = 0
self.home_dir = os.path.expanduser("~")
self.last_api_call = 0
def should_log(self, level):
log_level = Parameters.get("Mode6", "Error")
if log_level == "Info": return True
elif log_level == "Status": return level in ["Status", "Error"]
else: return level == "Error"
def classify_aircraft(self, t, desc):
t = t.upper() if t else ""
desc = desc.upper() if desc else ""
# 1. Helicopters (Priority)
if "HELICOPTER" in desc or "ROTORCRAFT" in desc or t in ["EC35", "EC45", "H135", "H145", "AS32", "EH10", "NH90", "CH47"]:
return ("Helicopters", "helicopter")
# 2. Special / Military / Large Transport
if "RIVET" in desc or t == "R135":
return ("Boeing RC-135 (Radar)", "known")
if "AWACS" in desc or t == "E3TF":
return ("Boeing E-3 AWACS", "known")
if "GLOBEMASTER" in desc or t == "C17":
return ("Boeing C-17 Globemaster", "known")
if "ATLAS" in desc or t == "A400":
return ("Airbus A400M Atlas", "known")
if "HERCULES" in desc or t == "C130":
return ("Lockheed C-130 Hercules", "known")
if "STRATOTANKER" in desc or t == "K35R":
return ("Boeing KC-135 Tanker", "known")
# Specifiek voor A330 tankers (Voyager/MRTT) op basis van omschrijving
if "VOYAGER" in desc or "MRTT" in desc:
return ("Airbus A330 Tanker/Transport", "known")
if "BELUGA" in desc or t in {"A3ST", "A337"}:
return ("Airbus Beluga", "known")
# 3. Airbus Families
if t.startswith("A38") or "380" in desc:
return ("Airbus A380", "known")
if t.startswith("A35") or "350" in desc:
return ("Airbus A350", "known")
if t.startswith("A34") or "340" in desc:
return ("Airbus A340", "known")
# Algemene A330 check (vangt nu ook A332 op die niet Voyager is)
if t.startswith("A33") or "330" in desc:
return ("Airbus A330", "known")
if (t.startswith("A31") or t.startswith("A32") or t.startswith("A2") or
"A32" in desc or "A-32" in desc):
return ("Airbus A320-family", "known")
if t.startswith("BCS") or "A220" in desc:
return ("Airbus A220", "known")
# 4. Boeing Families
if t.startswith("B73") or t.startswith("B3") or "737" in desc:
return ("Boeing 737", "known")
if t.startswith("B74") or "747" in desc:
return ("Boeing 747", "known")
if t.startswith("B77") or "777" in desc:
return ("Boeing 777", "known")
if t.startswith("B78") or "787" in desc:
return ("Boeing 787", "known")
if t.startswith("B75") or "757" in desc:
return ("Boeing 757", "known")
if t.startswith("B76") or "767" in desc:
return ("Boeing 767", "known")
# 5. Commercial Regional
if (t.startswith("E17") or t.startswith("E19") or t.startswith("E2") or
"E-JET" in desc or "E170" in desc or "E190" in desc):
return ("Embraer E-Jet Family", "known")
if t.startswith("DH8") or "DASH 8" in desc or "Q400" in desc:
return ("Dash 8 / Q400", "known")
if t.startswith("AT") or "ATR" in desc:
return ("ATR 42/72", "known")
if t.startswith("RJ") or t.startswith("B46") or "AVRO" in desc or "BAE 146" in desc:
return ("Avro RJ / BAe 146", "known")
if "FOKKER" in desc or t in ["F70", "F100"]:
return ("Fokker 70/100", "known")
# 6. Business & Recreational
if (t in ["E135", "E140", "E145", "ER3", "ER4", "ERJ"] or "ERJ-1" in desc or "ERJ 1" in desc):
return ("Business & Recreational", "small")
small_codes = {
"C25A", "C25B", "C25C", "C525", "C550", "C560", "C680",
"C510", "C500", "C501", "C551", "GLF4", "GLF5", "GLF6", "G280", "GALX",
"FA50", "FA7X", "FA20", "FA2K", "FA10", "LJ35", "LJ45", "LJ60",
"BE20", "BE30", "BE40", "BE9L", "BE10", "PC12", "PC6", "PC24",
"CL30", "CL35", "CL60", "GL5T", "H25B", "H25C", "HA4T", "SW4",
"P28A", "P28R", "P28T", "PA46", "C172", "C182", "C208", "SR20", "SR22"
}
small_keywords = ["CITATION", "GULFSTREAM", "FALCON", "LEARJET", "HAWKER",
"CHALLENGER", "PHENOM", "LEGACY", "BEECHCRAFT", "PILATUS",
"CESSNA", "PIPER", "CIRRUS", "METRO", "KINGAIR"]
if t in small_codes or any(k in desc for k in small_keywords):
return ("Business & Recreational", "small")
# 7. Other Commercial / Classics
if "MD11" in desc or t == "MD11":
return ("McDonnell Douglas MD-11", "known")
if "MD8" in desc or t.startswith("MD8"):
return ("McDonnell Douglas MD-80", "known")
return ("Other", "other")
def log_unknown_type(self, hex_id, t, desc, reg):
if Parameters.get("Mode3", "False") != "True": return
log_key = f"{t}|{desc}"
if log_key in self.logged_unknowns: return
self.logged_unknowns.add(log_key)
log_file = os.path.join(self.home_dir, "unknown_aircraft_types.log")
timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
log_entry = f"{timestamp} | hex={hex_id} | t={t} | desc={desc} | r={reg}\n"
try:
with open(log_file, 'a') as f: f.write(log_entry)
except: pass
def get_cardinal_dir(self, angle):
directions = ["N", "NNE", "NE", "ENE", "E", "ESE", "SE", "SSE",
"S", "SSW", "SW", "WSW", "W", "WNW", "NW", "NNW"]
return directions[int((angle + 11.25) / 22.5) % 16]
def calculate_distance(self, lat1, lon1, lat2, lon2):
R = 6371
dLat, dLon = math.radians(lat2 - lat1), math.radians(lon2 - lon1)
a = math.sin(dLat/2)**2 + math.cos(math.radians(lat1)) * math.cos(math.radians(lat2)) * math.sin(dLon/2)**2
return R * 2 * math.atan2(math.sqrt(a), math.sqrt(1-a))
def onStart(self):
if 1 not in Devices: Domoticz.Device(Name="Tracker", Unit=1, TypeName="Text", Used=1).Create()
if 2 not in Devices: Domoticz.Device(Name="Counter", Unit=2, Type=113, Subtype=0, Used=1).Create()
else:
try:
if Devices[2].sValue: self.total_count = int(Devices[2].sValue.split(';')[0])
except: self.total_count = 0
if 3 not in Devices: Domoticz.Device(Name="Types", Unit=3, TypeName="Text", Used=1).Create()
Domoticz.Log("Airplane Tracker started.")
def onHeartbeat(self):
try:
if "Location" not in Settings: return
loc = Settings["Location"].split(";")
my_lat, my_lon = float(loc[0]), float(loc[1])
now_ts, now_dt = time.time(), datetime.now()
if now_dt.strftime('%Y-%m-%d') != self.today:
self.today = now_dt.strftime('%Y-%m-%d')
self.last_seen_ts.clear(); self.tracker_buffer.clear(); self.type_counts.clear()
api_interval = int(Parameters.get("Mode2", "60"))
if (now_ts - self.last_api_call) < api_interval:
self._update_displays(now_ts, now_dt, False); return
self.last_api_call = now_ts
url = f"https://api.airplanes.live/v2/point/{my_lat}/{my_lon}/{Parameters['Mode1']}"
req = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0'})
new_reg_found = False
try:
with urllib.request.urlopen(req, timeout=10) as r:
ac_data = json.loads(r.read().decode()).get('ac', [])
for p in ac_data:
hex_id = p.get('hex', '0').upper()
if hex_id == '0': continue
flight = (p.get('flight') or hex_id).strip()
t, desc, reg = p.get('t', ''), p.get('desc', 'Unknown'), p.get('r', '')
if hex_id not in self.last_seen_ts or (now_ts - self.last_seen_ts[hex_id]) > 300:
self.total_count += 1; new_reg_found = True
t_disp, cat = self.classify_aircraft(t, desc)
self.type_counts[t_disp] = self.type_counts.get(t_disp, 0) + 1
if cat == "other": self.log_unknown_type(hex_id, t, desc, reg)
self.last_seen_ts[hex_id] = now_ts
t_disp, _ = self.classify_aircraft(t, desc)
alt = round((p.get('alt_geom') or p.get('alt_baro', 0)) * 0.3048)
gs = round(p.get('gs', 0) * 1.852)
v_rate = p.get('baro_rate', 0)
v_str = f" (↑ {round(v_rate * 0.3048)} m/min)" if v_rate > 128 else (f" (↓ {round(abs(v_rate) * 0.3048)} m/min)" if v_rate < -128 else "")
dist = self.calculate_distance(my_lat, my_lon, p.get('lat', 0), p.get('lon', 0))
link = f"<a href='https://globe.airplanes.live/?icao={hex_id}' target='_blank' style='color:inherit;text-decoration:none;'>{flight}</a>"
line1 = f"{now_dt.strftime('%H:%M')} {link} - {t_disp} ({reg})"
line2 = f"{alt}m{v_str}, {gs}km/h, {dist:.1f}km {self.get_cardinal_dir(p.get('track', 0))}"
self.tracker_buffer[hex_id] = (line1 + "<br>" + line2 + "<br>", now_ts)
except Exception as e:
if self.should_log("Error"): Domoticz.Error(f"API Error: {e}")
self._update_displays(now_ts, now_dt, new_reg_found)
except Exception as e:
if self.should_log("Error"): Domoticz.Error(f"Heartbeat Error: {e}")
def _update_displays(self, now_ts, now_dt, new_reg_found):
expired = [k for k, v in self.tracker_buffer.items() if (now_ts - v[1]) > 1800]
for k in expired: del self.tracker_buffer[k]
if 1 in Devices:
sorted_p = sorted(self.tracker_buffer.values(), key=lambda x: x[1], reverse=True)
Devices[1].Update(nValue=0, sValue="<br>".join([p[0] for p in sorted_p[:3]]) if sorted_p else now_dt.strftime('%H:%M') + " No traffic")
if 2 in Devices and new_reg_found: Devices[2].Update(nValue=0, sValue=str(self.total_count))
if 3 in Devices and self.type_counts:
sorted_t = sorted(self.type_counts.items(), key=lambda x: (x[0] == "Other", -x[1], x[0]))
Devices[3].Update(nValue=0, sValue="<br>".join([f"{n}: {c}" for n, c in sorted_t]))
_plugin = BasePlugin()
def onStart(): _plugin.onStart()
def onHeartbeat(): _plugin.onHeartbeat()