This connects to readsb and will convert positions to degree and distance.
import asyncio, math
from haversine import haversine
local_position = (21.443234, 17.73423423)
def calculate_initial_compass_bearing(pointA, pointB):
"""
Calculates the bearing between two points.
The formulae used is the following:
θ = atan2(sin(Δlong).cos(lat2),
cos(lat1).sin(lat2) − sin(lat1).cos(lat2).cos(Δlong))
:Parameters:
- `pointA: The tuple representing the latitude/longitude for the
first point. Latitude and longitude must be in decimal degrees
- `pointB: The tuple representing the latitude/longitude for the
second point. Latitude and longitude must be in decimal degrees
:Returns:
The bearing in degrees
:Returns Type:
float
"""
if (type(pointA) != tuple) or (type(pointB) != tuple):
raise TypeError("Only tuples are supported as arguments")
lat1 = math.radians(pointA[0])
lat2 = math.radians(pointB[0])
diffLong = math.radians(pointB[1] - pointA[1])
x = math.sin(diffLong) * math.cos(lat2)
y = math.cos(lat1) * math.sin(lat2) - (math.sin(lat1)
* math.cos(lat2) * math.cos(diffLong))
initial_bearing = math.atan2(x, y)
# Now we have the initial bearing but math.atan2 return values
# from -180° to + 180° which is not what we want for a compass bearing
# The solution is to normalize the initial bearing as shown below
initial_bearing = math.degrees(initial_bearing)
compass_bearing = (initial_bearing + 360) % 360
return compass_bearing
def handle_position(icoa, alt, lat, lon):
dist = round(haversine(local_position, (lat,lon)),5)
dir = round(calculate_initial_compass_bearing(local_position, (lat,lon)),5)
print(f'{icoa},{alt},{lat},{lon},{dist},{dir}')
def handle_message(msg):
if len(msg) != 22:
return
msg_id = int(msg[1])
icoa = msg[4]
match msg_id:
case 3:
try:
alt = int(msg[11])
lat = float(msg[14])
lon = float(msg[15])
handle_position(icoa, alt, lat, lon)
except:
return
async def tcp_echo_client():
reader, writer = await asyncio.open_connection('localhost', 30003)
while True:
line = await reader.readline()
line = line.decode('latin1').rstrip()
msg = line.split(',')
handle_message(msg)
asyncio.run(tcp_echo_client())
This will then convert above stuff to geojson that can you can plot in leaflet.
import csv,sys
from geojson import MultiPoint
from geojson import Polygon
import numpy as np
local_position = (52.3231232, 7.1345345)
black_list = ['406F7B', '48C121', '485E2F', '4CAE68', '7815FA', '78100A', '781093', '06A2E8', 'A31A76', '78158E', '406F77', '70C0F9', '485F82', '78100a', '80166C', '781003', '8966C3', '40731C', '70209E', '70C125', '4CADF4', '485785', '48C229', '4073CE', '406D70', '51408C', '78140C']
def print_points(rows):
points = []
for row in rows:
points.append((float(row[3]), float(row[2])))
print(MultiPoint(points))
def print_range(rows):
resolution = 360
max_range = np.zeros(shape=resolution)
max_range_points = np.empty(shape=(resolution,2))
for i in range(resolution):
max_range_points[i] = local_position
for row in rows:
dist = float(row[4])
dir = float(row[5])
i = int(round((dir / 360.0) * resolution)) % resolution
if ((dist < 500) and (max_range[i] < dist) and (row[0] not in black_list)):
max_range[i] = dist
max_range_points[i] = (float(row[2]), float(row[3]))
#if (dist > 250):
# print(row[0])
result = []
for i in range(resolution):
result.append((max_range_points[i][0], max_range_points[i][1]))
print(Polygon(result))
with open(sys.argv[1], newline='') as csvfile:
rows = csv.reader(csvfile, delimiter=',')
if (sys.argv[2]=='-points'):
print_points(rows)
if (sys.argv[2]=='-range'):
print_range(rows)
Something like this then:
// fetch the current 50 range polygon from the server
$.getJSON('./curr.json', function (data) {
// and add it to the map
var polygon = L.polygon(data.coordinates, {
color: '#000000',
weight: 1,
fillOpacity: 0.1,
opacity: 0.4
}).addTo(map);
});
Edit: Blacklist is there for planes that i do not want in the range plot mostly because they flew over a GPS-jammed area.
