update ACAS and BDS61 structures and functions
This commit is contained in:
Junzi Sun
parent
c25c9d6b96
commit
2e9833148b
@ -10,6 +10,7 @@ except:
|
||||
|
||||
from .decoder import tell
|
||||
from .decoder import adsb
|
||||
from .decoder import acas
|
||||
from .decoder import commb
|
||||
from .decoder import bds
|
||||
from .extra import aero
|
||||
|
||||
@ -1,155 +1,123 @@
|
||||
"""
|
||||
Decoding Air-Air Surveillance (ACAS) DF=0/16
|
||||
|
||||
[To be implemented]
|
||||
"""
|
||||
from __future__ import absolute_import, print_function, division
|
||||
|
||||
from pyModeS import common
|
||||
|
||||
|
||||
def threat_type(msg):
|
||||
def rac(msg: str) -> str:
|
||||
"""Resolution Advisory Complement.
|
||||
|
||||
:param msg: 28 hexdigits string
|
||||
:return: RACs
|
||||
"""
|
||||
Determine the threat type indicator
|
||||
mv = common.hex2bin(common.data(msg))
|
||||
|
||||
===== =======
|
||||
Value Meaning
|
||||
===== =======
|
||||
0 No identity data in TID
|
||||
1 TID has Mode S address (ICAO)
|
||||
2 TID has altitude, range, and bearing
|
||||
3 Not assigned
|
||||
:param msg:
|
||||
:return: indicator of threat type
|
||||
RAC = []
|
||||
|
||||
if mv[22] == "1":
|
||||
RAC.append("do not pass below")
|
||||
|
||||
if mv[23] == "1":
|
||||
RAC.append("do not pass above")
|
||||
|
||||
if mv[24] == "1":
|
||||
RAC.append("do not pass left")
|
||||
|
||||
if mv[25] == "1":
|
||||
RAC.append("do not pass right")
|
||||
|
||||
return "; ".join(RAC)
|
||||
|
||||
|
||||
def rat(msg: str) -> bool:
|
||||
"""RA terminated indicator
|
||||
|
||||
Mode S transponder is still required to report RA 18 seconds after
|
||||
it is terminated by ACAS. Hence, the RAT filed is used.
|
||||
|
||||
:param msg: 28 hexdigits string
|
||||
:return: if RA has been terminated
|
||||
"""
|
||||
mb = common.hex2bin(msg)[32:]
|
||||
tti = common.bin2int(mb[28:30])
|
||||
return tti
|
||||
mv = common.hex2bin(common.data(msg))
|
||||
mte = int(mv[26])
|
||||
return mte
|
||||
|
||||
|
||||
def threat_identity(msg):
|
||||
mb = common.hex2bin(msg)[32:]
|
||||
tti = threat_type(msg)
|
||||
def mte(msg: str) -> bool:
|
||||
"""Multiple threat encounter.
|
||||
|
||||
# The ICAO of the threat is announced
|
||||
if tti == 1:
|
||||
return common.icao(mb[30:55])
|
||||
else:
|
||||
raise RuntimeError("%s: Missing threat identity (ICAO)")
|
||||
|
||||
|
||||
def threat_location(msg):
|
||||
:param msg: 28 hexdigits string
|
||||
:return: if there are multiple threats
|
||||
"""
|
||||
Get the altitude, range, and bearing of the threat
|
||||
Altitude is the Mode C altitude
|
||||
:param msg:
|
||||
:return: tuple of the Mode C altitude, range, and bearing
|
||||
mv = common.hex2bin(common.data(msg))
|
||||
mte = int(mv[27])
|
||||
return mte
|
||||
|
||||
|
||||
def ara(msg: str) -> str:
|
||||
"""Decode active resolution advisory.
|
||||
|
||||
:param msg: 28 bytes hexadecimal message string
|
||||
:return: RA charactristics
|
||||
"""
|
||||
mb = common.hex2bin(msg)[32:]
|
||||
tti = threat_type(msg)
|
||||
mv = common.hex2bin(common.data(msg))
|
||||
|
||||
# Altitude, range, and bearing of threat
|
||||
if tti == 2:
|
||||
grey = mb[31] + mb[32] + mb[33] + mb[34] + mb[35] + mb[36] \
|
||||
+ mb[38] + mb[38] + mb[39] + mb[40] + mb[41] + mb[42] \
|
||||
+ mb[43]
|
||||
mode_c_alt = common.gray2alt(grey)
|
||||
_range = common.bin2int(mb[44:51])
|
||||
bearing = common.bin2int(mb[51:57])
|
||||
return mode_c_alt, _range, bearing
|
||||
mte = int(mv[27])
|
||||
|
||||
|
||||
def has_multiple_threats(msg):
|
||||
"""
|
||||
Indicate if the ACAS is processing zero, one, or more than one threat
|
||||
simultaneously
|
||||
:param msg:
|
||||
:return: boolean
|
||||
"""
|
||||
mb = common.bin2int(msg)[32:]
|
||||
|
||||
mte = mb[27]
|
||||
ara_b1 = mb[8]
|
||||
|
||||
if ara_b1 == 0 and mte == 0:
|
||||
# There are no active threats
|
||||
return False
|
||||
elif ara_b1 == 1 and mte == 0:
|
||||
# There is a single threat
|
||||
return False
|
||||
elif mte == 1:
|
||||
# There are multiple threats
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
|
||||
|
||||
def active_resolution_advisories(msg):
|
||||
mb = common.bin2int(msg)[32:]
|
||||
|
||||
mte = mb[27]
|
||||
|
||||
ara_b1 = mb[8]
|
||||
ara_b2 = mb[9]
|
||||
ara_b3 = mb[10]
|
||||
ara_b4 = mb[11]
|
||||
ara_b5 = mb[12]
|
||||
ara_b6 = mb[14]
|
||||
ara_b1 = int(mv[8])
|
||||
ara_b2 = int(mv[9])
|
||||
ara_b3 = int(mv[10])
|
||||
ara_b4 = int(mv[11])
|
||||
ara_b5 = int(mv[12])
|
||||
ara_b6 = int(mv[14])
|
||||
ara_b7 = int(mv[15])
|
||||
# ACAS III are bits 15-22
|
||||
|
||||
# blah, now what? just return the byte string and leave it up to the user?
|
||||
# There are several indicators depending on if the ara_b1 and mte are set
|
||||
RA = []
|
||||
|
||||
return mb[8:15]
|
||||
if ara_b1 == 1:
|
||||
if ara_b2:
|
||||
RA.append("corrective")
|
||||
else:
|
||||
RA.append("preventive")
|
||||
|
||||
if ara_b3:
|
||||
RA.append("downward sense")
|
||||
else:
|
||||
RA.append("upward sense")
|
||||
|
||||
def is_ra_terminated(msg):
|
||||
"""
|
||||
Indicate if the threat is still being generated. If not, this can be due to
|
||||
the threat no longer existing or the aircraft generating the indicator is
|
||||
no longer broadcasting an altitude while the threat is still a threat
|
||||
:param msg:
|
||||
:return: if the threat is terminated
|
||||
"""
|
||||
mb = common.bin2int(msg)[32:]
|
||||
return mb[26] == 1
|
||||
if ara_b4:
|
||||
RA.append("increased rate")
|
||||
|
||||
if ara_b5:
|
||||
RA.append("sense reversal")
|
||||
|
||||
def no_pass_below(msg):
|
||||
"""
|
||||
Indication to aircraft to pass below or not
|
||||
:param msg:
|
||||
:return: boolean
|
||||
"""
|
||||
mb = common.bin2int(msg)[32:]
|
||||
return mb[22] == 1
|
||||
if ara_b6:
|
||||
RA.append("altitude crossing")
|
||||
|
||||
if ara_b7:
|
||||
RA.append("positive")
|
||||
else:
|
||||
RA.append("vertical speed limit")
|
||||
|
||||
def no_pass_above(msg):
|
||||
"""
|
||||
Indication to aircraft to pass above or not
|
||||
:param msg:
|
||||
:return: boolean
|
||||
"""
|
||||
mb = common.bin2int(msg)[32:]
|
||||
return mb[23] == 1
|
||||
if ara_b1 == 0 and mte == 1:
|
||||
if ara_b2:
|
||||
RA.append("requires a correction in the upward sense")
|
||||
|
||||
if ara_b3:
|
||||
RA.append("requires a positive climb")
|
||||
|
||||
def no_pass_left(msg):
|
||||
"""
|
||||
Indication to aircraft to pass on the left or not
|
||||
:param msg:
|
||||
:return: boolean
|
||||
"""
|
||||
mb = common.bin2int(msg)[32:]
|
||||
return mb[24] == 1
|
||||
if ara_b4:
|
||||
RA.append("requires a correction in downward sense")
|
||||
|
||||
if ara_b5:
|
||||
RA.append("requires a positive descent")
|
||||
|
||||
def no_pass_right(msg):
|
||||
"""
|
||||
Indication to aircraft to pass on the right or not
|
||||
:param msg:
|
||||
:return: boolean
|
||||
"""
|
||||
mb = common.bin2int(msg)[32:]
|
||||
return mb[25] == 1
|
||||
if ara_b6:
|
||||
RA.append("requires a crossing")
|
||||
|
||||
if ara_b7:
|
||||
RA.append("requires a sense reversal")
|
||||
|
||||
return "; ".join(RA)
|
||||
|
||||
@ -28,7 +28,11 @@ from pyModeS.decoder.bds.bds06 import (
|
||||
)
|
||||
from pyModeS.decoder.bds.bds08 import category, callsign
|
||||
from pyModeS.decoder.bds.bds09 import airborne_velocity, altitude_diff
|
||||
from pyModeS.decoder.bds.bds61 import is_emergency, emergency_state, emergency_squawk
|
||||
from pyModeS.decoder.bds.bds61_st1 import (
|
||||
is_emergency,
|
||||
emergency_state,
|
||||
emergency_squawk,
|
||||
)
|
||||
|
||||
|
||||
def df(msg):
|
||||
|
||||
@ -2,6 +2,7 @@
|
||||
# BDS 6,1
|
||||
# ADS-B TC=28
|
||||
# Aircraft Airborne status
|
||||
# (Subtype 1)
|
||||
# ------------------------------------------
|
||||
|
||||
from pyModeS import common
|
||||
100
pyModeS/decoder/bds/bds61_st2.py
Normal file
100
pyModeS/decoder/bds/bds61_st2.py
Normal file
@ -0,0 +1,100 @@
|
||||
# ------------------------------------------
|
||||
# BDS 6,1
|
||||
# ADS-B TC=28
|
||||
# Aircraft Airborne status
|
||||
# (Subtype 1)
|
||||
# ------------------------------------------
|
||||
|
||||
from typing import Tuple
|
||||
from pyModeS import common
|
||||
|
||||
from pyModeS.decoder import acas
|
||||
|
||||
|
||||
def threat_type(msg: str) -> int:
|
||||
"""Determine the threat type indicator.
|
||||
|
||||
Value Meaning
|
||||
----- ---------------------------------------
|
||||
0 No identity data in TID
|
||||
1 TID has Mode S address (ICAO)
|
||||
2 TID has altitude, range, and bearing
|
||||
3 Not assigned
|
||||
|
||||
:param msg: 28 hexdigits string
|
||||
:return: indicator of threat type
|
||||
"""
|
||||
mb = common.hex2bin(common.data(msg))
|
||||
tti = common.bin2int(mb[28:30])
|
||||
return tti
|
||||
|
||||
|
||||
def threat_identity(msg: str) -> str:
|
||||
mb = common.hex2bin(common.data(msg))
|
||||
tti = threat_type(msg)
|
||||
|
||||
# The ICAO of the threat is announced
|
||||
if tti == 1:
|
||||
return common.icao(mb[30:55])
|
||||
else:
|
||||
raise RuntimeError("%s: Missing threat identity (ICAO)")
|
||||
|
||||
|
||||
def threat_location(msg: str) -> Tuple:
|
||||
"""Get the altitude, range, and bearing of the threat.
|
||||
|
||||
Altitude is the Mode C altitude
|
||||
|
||||
:param msg: 28 hexdigits string
|
||||
:return: tuple of the Mode C altitude, range, and bearing
|
||||
"""
|
||||
mb = common.hex2bin(common.data(msg))
|
||||
tti = threat_type(msg)
|
||||
|
||||
# Altitude, range, and bearing of threat
|
||||
if tti == 2:
|
||||
altitude = common.altitude(mb[31:44])
|
||||
distance = common.bin2int(mb[44:51])
|
||||
bearing = common.bin2int(mb[51:57])
|
||||
return altitude, distance, bearing
|
||||
|
||||
|
||||
def has_multiple_threats(msg: str) -> bool:
|
||||
""" Indicate if the ACAS is processing multiple threats simultaneously.
|
||||
|
||||
:param msg: 28 hexdigits string
|
||||
:return: if there are multiple threats
|
||||
"""
|
||||
return acas.mte(msg) == 1
|
||||
|
||||
|
||||
def active_resolution_advisories(msg: str) -> str:
|
||||
"""Decode active resolution advisory.
|
||||
|
||||
Uses ARA decoding function from ACAS module.
|
||||
|
||||
:param msg: 28 bytes hexadecimal message string
|
||||
:return: RA charactristics
|
||||
"""
|
||||
return acars.ara(msg)
|
||||
|
||||
|
||||
def is_ra_terminated(msg: str) -> bool:
|
||||
"""Indicate if the threat is still being generated.
|
||||
|
||||
Mode S transponder is still required to report RA 18 seconds after
|
||||
it is terminated by ACAS. Hence, the RAT filed is used.
|
||||
|
||||
:param msg: 28 hexdigits string
|
||||
:return: if the threat is terminated
|
||||
"""
|
||||
return acas.rat(msg) == 1
|
||||
|
||||
|
||||
def ra_complement(msg: str) -> str:
|
||||
"""Resolution Advisory Complement.
|
||||
|
||||
:param msg: 28 hexdigits string
|
||||
:return: RACs
|
||||
"""
|
||||
return acas.rac(msg)
|
||||
40
tests/test_acas.py
Normal file
40
tests/test_acas.py
Normal file
@ -0,0 +1,40 @@
|
||||
import pyModeS as pms
|
||||
|
||||
msgs = [
|
||||
"80E1983958C392E8710C642D0BAD",
|
||||
"8609F8E19E90653083FDE096BE26",
|
||||
"80011B41F40017D0000012E06B45",
|
||||
"808182365813667446EB715E253D",
|
||||
"80E1953058AB029BE8F4E60D989E",
|
||||
"8667ECFA8FE06B30E59A124D5AEF",
|
||||
"80030AA000042C0AD05D6205DB2C",
|
||||
"8631A80000373C463B6E7A00008B",
|
||||
"80E19131588B146108DE703F3C47",
|
||||
"825CC4A0001595C4600030A40000",
|
||||
"808182365813667438EB710EB6D8",
|
||||
"80A18393581D3655E90354A664B2",
|
||||
"8081823658136309B4F22BCB5F8D",
|
||||
"80E1953058AB06516E8602756DD8",
|
||||
"80E1991058C9063AB6A3E4744DC3",
|
||||
"8073EC91840AFCED8F300BE765C5",
|
||||
"8571F54AA814BF8130066A19A31D",
|
||||
"864070E1990D3B1E78048BAE6987",
|
||||
"80E1991058C9063AB6A3E4744DC3",
|
||||
"80818298581581F7CAF8C3C1EEA4",
|
||||
"80004E98BC90000FF01010951298",
|
||||
"8526D57E4D963C92CDEE1B6C7C49",
|
||||
"802A613C93F65A7FF803A51B5ADB",
|
||||
"85B6C54279B67BE2A0001998FFDA",
|
||||
"851944F15881648338D1AF4B7A27",
|
||||
"8321014858208000787905B0E800",
|
||||
"866DD078EDEBD330404FFFAE9BA5",
|
||||
"80E196905AB503260E835D849E35",
|
||||
]
|
||||
|
||||
for msg in msgs:
|
||||
print("-" * 80)
|
||||
print(msg)
|
||||
print("ARA", pms.acas.ara(msg))
|
||||
print("RAC", pms.acas.rac(msg))
|
||||
print("MTE", pms.acas.mte(msg))
|
||||
print("RAT", pms.acas.rat(msg))
|
||||
Loading…
Reference in New Issue
Block a user