forked from Matioupi/realtime-adsb-out
-
Notifications
You must be signed in to change notification settings - Fork 0
/
realtime-adsb-out.py
executable file
·156 lines (133 loc) · 7.06 KB
/
realtime-adsb-out.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
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
#!/usr/bin/env python3
""" This file hold the main function which read user inputs
initialize and launch the simulation
This program is free software: you can redistribute it and/or modify it under
the terms of the GNU General Public License as published by the Free Software
Foundation, either version 3 of the License, or (at your option) any later
version.
This program is distributed in the hope that it will be useful, but WITHOUT
ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.
You should have received a copy of the GNU General Public License along with
this program. If not, see <http://www.gnu.org/licenses/>.
"""
import sys, time, math
import threading
from AircraftInfos import AircraftInfos
from FixedTrajectorySimulator import FixedTrajectorySimulator
from PseudoCircleTrajectorySimulator import PseudoCircleTrajectorySimulator
from RandomTrajectorySimulator import RandomTrajectorySimulator
from WaypointsTrajectorySimulator import WaypointsTrajectorySimulator
from HackRfBroadcastThread import HackRfBroadcastThread
from getopt import getopt, GetoptError
def usage(msg=False):
if msg:print(msg)
print("Usage: %s [options]\n" % sys.argv[0])
print("-h | --help Display help message.")
print("--icao <opt> Callsign in hex, Default:0x508035")
print("--callsign <opt> Callsign (8 chars max), Default:DEADBEEF")
print("--squawk <opt> 4-digits 4096 code squawk, Default:7000")
print("--trajectorytype <opt> Type of simulated trajectory amongst :")
print(" fixed : steady aircraft")
print(" circle : pseudo circular flight")
print(" random : random positions inside circle area")
print(" waypoints : fly long flight path")
print(" Default:fixed")
print("--lat <opt> Latitude for the plane in decimal degrees, Default:50.44994")
print("--long <opt> Longitude for the place in decimal degrees. Default:30.5211")
print("--altitude <opt> Altitude in decimal feet, Default:1500.0")
print("--speed <opt> Airspeed in decimal kph, Default:300.0")
print("--vspeed <opt> Vertical speed en ft/min, positive up, Default:0")
print("--maxloadfactor Specify the max load factor for aircraft simulation. Default:1.45")
print("--trackangle <opt> Track angle in decimal degrees. Default:0")
print("--timesync <opt> 0/1, 0 indicates time not synchronous with UTC, Default:0")
print("--capability <opt> Capability, Default:5")
print("--typecode <opt> ADS-B message type, Default:11")
print("--sstatus <opt> Surveillance status, Default:0")
print("--nicsupplementb <opt> NIC supplement-B, Default:0")
print("--surface Aircraft located on ground, Default:False")
print("--waypoints <opt> Waypoints file for waypoints trajectory")
print("--posrate <opt> position frame broadcast period in µs, Default: 150000")
print("")
#print("see usage.md for additionnal information")
sys.exit(2)
def main():
# Default values
icao_aa = '0x508035'
callsign = 'DEADBEEF'
squawk = '7000'
alt_ft = 1500.0
lat_deg = 50.44994
lon_deg = 30.5211
speed_kph = 300.0
vspeed_ftpmin = 0.0
maxloadfactor = 1.45
track_angle_deg = 0.0
capability = 5
type_code = 11
surveillance_status = 0
timesync = 0
nicsup = 0
on_surface = False
trajectory_type = 'fixed'
waypoints_file = None
posrate = 150000
try:
(opts, args) = getopt(sys.argv[1:], 'h', \
['help','icao=','callsign=','squawk=','trajectorytype=','lat=','long=','altitude=','speed=','vspeed=','maxloadfactor=','trackangle=',
'timesync=','capability=','typecode=','sstatus=','nicsupplementb=','surface','posrate='
])
except GetoptError as err:
usage("%s\n" % err)
if len(opts) != 0:
for (opt, arg) in opts:
if opt in ('-h', '--help'):usage()
elif opt in ('--icao'):icao_aa = arg
elif opt in ('--callsign'):callsign = arg
elif opt in ('--squawk'):squawk = arg
elif opt in ('--trajectorytype'):trajectory_type = arg
elif opt in ('--lat'):lat_deg = float(arg)
elif opt in ('--long'):lon_deg = float(arg)
elif opt in ('--altitude'):alt_ft = float(arg)
elif opt in ('--speed'):speed_kph = float(arg)
elif opt in ('--vspeed'):vspeed_ftpmin = float(arg)
elif opt in ('--maxloadfactor'):maxloadfactor = float(arg)
elif opt in ('--trackangle'):track_angle_deg = float(arg)
elif opt in ('--timesync'):timesync = int(arg)
elif opt in ('--capability'):capability = int(arg)
elif opt in ('--typecode'):type_code = int(arg)
elif opt in ('--sstatus'):surveillance_status = int(arg)
elif opt in ('--nicsupplementb'):nicsup = int(arg)
elif opt in ('--surface'):on_surface = True
elif opt in ('--posrate'):posrate = int(arg)
else:usage("Unknown option %s\n" % opt)
aircraftinfos = AircraftInfos(icao_aa,callsign,squawk, \
lat_deg,lon_deg,alt_ft,speed_kph,vspeed_ftpmin,maxloadfactor,track_angle_deg, \
timesync,capability,type_code,surveillance_status,nicsup,on_surface)
# TODO : the mutex would better be implemented as an object attribute in broadcast thread
mutex = threading.Lock()
brodcast_thread = HackRfBroadcastThread(mutex,posrate) # posrate would usally be used with random mode to generate load of tracks
if trajectory_type == 'fixed':
trajectory_simulator = FixedTrajectorySimulator(mutex,brodcast_thread,aircraftinfos)
elif trajectory_type == 'circle':
trajectory_simulator = PseudoCircleTrajectorySimulator(mutex,brodcast_thread,aircraftinfos)
elif trajectory_type == 'random':
trajectory_simulator = RandomTrajectorySimulator(mutex,brodcast_thread,aircraftinfos)
elif trajectory_type == 'waypoints':
print("WaypointsTrajectorySimulator not implemented yet")
exit(-1)
trajectory_simulator = WaypointsTrajectorySimulator(mutex,brodcast_thread,aircraftinfos,waypoints_file)
while(val:=input("Type \'s + Enter\' to start the adsb-out simulation, and type \'s + Enter\' again to stop it:\n") != 's'):
time.sleep(0.05)
trajectory_simulator.start()
brodcast_thread.start()
# user input loop. Todo : implement other commands ? (in that case don't forget to check if mutex protection is needed)
while(val:=input("") != 's'):
time.sleep(0.05)
trajectory_simulator.stop()
brodcast_thread.stop()
trajectory_simulator.join()
brodcast_thread.join()
print("reatime-adsb-out simulation is finished")
if __name__ == "__main__":
main()