merge annotations

pull/83/head
Junzi Sun 4 years ago
commit 2085a2c432

@ -19,6 +19,7 @@ cpdef int typecode(str msg)
cpdef int cprNL(double lat)
cpdef str idcode(str msg)
cpdef int altcode(str msg)
cpdef int altitude(str binstr)
cpdef str data(str msg)
cpdef bint allzeros(str msg)

@ -256,7 +256,6 @@ cpdef str idcode(str msg):
"""Compute identity (squawk code).
Applicable only for DF5 or DF21 messages, bit 20-32.
credit: @fbyrkjeland
Args:
msg (String): 28 bytes hexadecimal message string
@ -280,7 +279,7 @@ cpdef str idcode(str msg):
cdef unsigned char A2 = mbin[22]
cdef unsigned char C4 = mbin[23]
cdef unsigned char A4 = mbin[24]
# _ = mbin[25]
# X = mbin[25]
cdef unsigned char B1 = mbin[26]
cdef unsigned char D1 = mbin[27]
cdef unsigned char B2 = mbin[28]
@ -288,11 +287,6 @@ cpdef str idcode(str msg):
cdef unsigned char B4 = mbin[30]
cdef unsigned char D4 = mbin[31]
# byte1 = int(A4 + A2 + A1, 2)
# byte2 = int(B4 + B2 + B1, 2)
# byte3 = int(C4 + C2 + C1, 2)
# byte4 = int(D4 + D2 + D1, 2)
idcode[0] = int_to_char((char_to_int(A4)*2 + char_to_int(A2))*2 + char_to_int(A1))
idcode[1] = int_to_char((char_to_int(B4)*2 + char_to_int(B2))*2 + char_to_int(B1))
idcode[2] = int_to_char((char_to_int(C4)*2 + char_to_int(C2))*2 + char_to_int(C1))
@ -321,47 +315,62 @@ cpdef int altcode(str msg):
if df(msg) not in [0, 4, 16, 20]:
raise RuntimeError("Message must be Downlink Format 0, 4, 16, or 20.")
# Altitude code, bit 20-32
cdef bytearray _mbin = bytearray(hex2bin(msg).encode())
alt = altitude(hex2bin(msg)[19:32])
return alt
@cython.boundscheck(False)
@cython.wraparound(False)
cpdef int altitude(str binstr):
if len(binstr) != 13 or set(binstr) != set('01'):
raise RuntimeError("Input must be 13 bits binary string")
cdef bytearray _mbin = bytearray(binstr.encode())
cdef unsigned char[:] mbin = _mbin
cdef char mbit = mbin[25] # M bit: 26
cdef char qbit = mbin[27] # Q bit: 28
cdef char Mbit = binstr[6]
cdef char Qbit = binstr[8]
cdef int alt = 0
cdef bytearray vbin
cdef bytearray _graybytes = bytearray(11)
cdef unsigned char[:] graybytes = _graybytes
if mbit == 48: # unit in ft, "0" -> 48
if qbit == 49: # 25ft interval, "1" -> 49
vbin = _mbin[19:25] + _mbin[26:27] + _mbin[28:32]
if bin2int(binstr) == 0:
# altitude unknown or invalid
alt = -9999
elif Mbit == 48: # unit in ft, "0" -> 48
if Qbit == 49: # 25ft interval, "1" -> 49
vbin = _mbin[:6] + _mbin[7:8] + _mbin[9:]
alt = bin2int(vbin.decode()) * 25 - 1000
if qbit == 48: # 100ft interval, above 50175ft, "0" -> 48
graybytes[8] = mbin[19]
graybytes[2] = mbin[20]
graybytes[9] = mbin[21]
graybytes[3] = mbin[22]
graybytes[10] = mbin[23]
graybytes[4] = mbin[24]
# _ = mbin[25]
graybytes[5] = mbin[26]
# cdef char D1 = mbin[27] # always zero
graybytes[6] = mbin[28]
graybytes[0] = mbin[29]
graybytes[7] = mbin[30]
graybytes[1] = mbin[31]
if Qbit == 48: # 100ft interval, above 50175ft, "0" -> 48
graybytes[8] = mbin[0]
graybytes[2] = mbin[1]
graybytes[9] = mbin[2]
graybytes[3] = mbin[3]
graybytes[10] = mbin[4]
graybytes[4] = mbin[5]
# M = mbin[6]
graybytes[5] = mbin[7]
# Q = mbin[8]
graybytes[6] = mbin[9]
graybytes[0] = mbin[10]
graybytes[7] = mbin[11]
graybytes[1] = mbin[12]
# graybytes = D2 + D4 + A1 + A2 + A4 + B1 + B2 + B4 + C1 + C2 + C4
alt = gray2alt(_graybytes.decode())
if mbit == 49: # unit in meter, "1" -> 49
vbin = _mbin[19:25] + _mbin[26:31]
elif Mbit == 49: # unit in meter, "1" -> 49
vbin = _mbin[:6] + _mbin[7:]
alt = int(bin2int(vbin.decode()) * 3.28084) # convert to ft
return alt
cpdef int gray2alt(str codestr):
cdef str gc500 = codestr[:8]
cdef int n500 = gray2int(gc500)

@ -139,17 +139,13 @@ def altitude(msg):
raise RuntimeError("%s: Not a airborn position message" % msg)
mb = common.hex2bin(msg)[32:]
altbin = mb[8:20]
if tc < 19:
# barometric altitude
q = mb[15]
if q:
n = common.bin2int(mb[8:15] + mb[16:20])
alt = n * 25 - 1000
else:
alt = None
altcode = altbin[0:6] + "0" + altbin[6:]
else:
# GNSS altitude, meters -> feet
alt = common.bin2int(mb[8:20]) * 3.28084
altcode = altbin[0:6] + "0" + altbin[6:]
alt = common.altitude(altcode)
return alt

@ -282,6 +282,8 @@ class TcpClient(object):
# raise RuntimeError("test exception")
except zmq.error.Again:
continue
except Exception as e:
tb = traceback.format_exc()
exception_queue.put(tb)

@ -96,7 +96,7 @@ def crc_legacy(msg: str, encode: bool = False) -> int:
continue
# perform XOR, when 1
msgnpbin[i: i + ng] = np.bitwise_xor(msgnpbin[i: i + ng], generator)
msgnpbin[i : i + ng] = np.bitwise_xor(msgnpbin[i : i + ng], generator)
# last 24 bits
msgbin = np.array2string(msgnpbin[-24:], separator="")[1:-1]
@ -209,7 +209,6 @@ def idcode(msg: str) -> str:
"""Compute identity (squawk code).
Applicable only for DF5 or DF21 messages, bit 20-32.
credit: @fbyrkjeland
Args:
msg (String): 28 bytes hexadecimal message string
@ -229,7 +228,7 @@ def idcode(msg: str) -> str:
A2 = mbin[22]
C4 = mbin[23]
A4 = mbin[24]
# _ = mbin[25]
# X = mbin[25]
B1 = mbin[26]
D1 = mbin[27]
B2 = mbin[28]
@ -246,10 +245,9 @@ def idcode(msg: str) -> str:
def altcode(msg: str) -> Optional[int]:
"""Compute the altitude.
"""Compute the altitude of DF4 or DF20.
Applicable only for DF4 or DF20 message, bit 20-32.
credit: @fbyrkjeland
Args:
msg (String): 28 bytes hexadecimal message string
@ -267,44 +265,69 @@ def altcode(msg: str) -> Optional[int]:
# Altitude code, bit 20-32
mbin = hex2bin(msg)
mbit = mbin[25] # M bit: 26
qbit = mbin[27] # Q bit: 28
altitude_code = mbin[19:32]
if mbit == "0": # unit in ft
if qbit == "1": # 25ft interval
vbin = mbin[19:25] + mbin[26] + mbin[28:32]
alt = altitude(altitude_code)
return alt
def altitude(binstr):
"""Decode 13 bits altitude code.
Args:
binstr (String): 13 bits binary string
Returns:
int: altitude in ft
"""
if len(binstr) != 13 or set(binstr) != set("01"):
raise RuntimeError("Input must be 13 bits binary string")
Mbit = binstr[6]
Qbit = binstr[8]
if bin2int(binstr) == 0:
# altitude unknown or invalid
alt = None
elif Mbit == "0": # unit in ft
if Qbit == "1": # 25ft interval
vbin = binstr[:6] + binstr[7] + binstr[9:]
alt = bin2int(vbin) * 25 - 1000
if qbit == "0": # 100ft interval, above 50175ft
C1 = mbin[19]
A1 = mbin[20]
C2 = mbin[21]
A2 = mbin[22]
C4 = mbin[23]
A4 = mbin[24]
# _ = mbin[25]
B1 = mbin[26]
# D1 = mbin[27] # always zero
B2 = mbin[28]
D2 = mbin[29]
B4 = mbin[30]
D4 = mbin[31]
if Qbit == "0": # 100ft interval, above 50187.5ft
C1 = binstr[0]
A1 = binstr[1]
C2 = binstr[2]
A2 = binstr[3]
C4 = binstr[4]
A4 = binstr[5]
# M = binstr[6]
B1 = binstr[7]
# Q = binstr[8]
B2 = binstr[9]
D2 = binstr[10]
B4 = binstr[11]
D4 = binstr[12]
graystr = D2 + D4 + A1 + A2 + A4 + B1 + B2 + B4 + C1 + C2 + C4
alt = gray2alt(graystr)
if mbit == "1": # unit in meter
vbin = mbin[19:25] + mbin[26:31]
if Mbit == "1": # unit in meter
vbin = binstr[:6] + binstr[7:]
alt = int(bin2int(vbin) * 3.28084) # convert to ft
return alt
def gray2alt(codestr: str) -> Optional[int]:
gc500 = codestr[:8]
def gray2alt(binstr: str) -> Optional[int]:
gc500 = binstr[:8]
n500 = gray2int(gc500)
# in 100-ft step must be converted first
gc100 = codestr[8:]
gc100 = binstr[8:]
n100 = gray2int(gc100)
if n100 in [0, 5, 6]:
@ -320,9 +343,9 @@ def gray2alt(codestr: str) -> Optional[int]:
return alt
def gray2int(graystr: str) -> int:
def gray2int(binstr: str) -> int:
"""Convert greycode to binary."""
num = bin2int(graystr)
num = bin2int(binstr)
num ^= num >> 8
num ^= num >> 4
num ^= num >> 2
@ -361,7 +384,7 @@ def wrongstatus(data: str, sb: int, msb: int, lsb: int) -> bool:
"""
# status bit, most significant bit, least significant bit
status = int(data[sb - 1])
value = bin2int(data[msb - 1: lsb])
value = bin2int(data[msb - 1 : lsb])
if not status:
if value != 0:

Loading…
Cancel
Save