update CRC code

This commit is contained in:
Junzi Sun 2019-05-27 21:26:18 +02:00
parent 7edbf3fd30
commit 6159691c3d
3 changed files with 95 additions and 44 deletions

View File

@ -5,48 +5,61 @@ from pyModeS.decoder import fastcrc
def hex2bin(hexstr):
"""Convert a hexdecimal string to binary string, with zero fillings. """
"""Convert a hexdecimal string to binary string, with zero fillings."""
num_of_bits = len(hexstr) * 4
binstr = bin(int(hexstr, 16))[2:].zfill(int(num_of_bits))
return binstr
def bin2int(binstr):
"""Convert a binary string to integer. """
return int(binstr, 2)
def hex2int(hexstr):
"""Convert a hexdecimal string to integer. """
"""Convert a hexdecimal string to integer."""
return int(hexstr, 16)
def int2hex(n):
"""Convert a integer to hexadecimal string."""
return hex(n)[2:].rjust(6, '0').upper()
def bin2int(binstr):
"""Convert a binary string to integer."""
return int(binstr, 2)
def bin2hex(hexstr):
"""Convert a hexdecimal string to integer."""
return int2hex(bin2int(hexstr))
def bin2np(binstr):
"""Convert a binary string to numpy array. """
"""Convert a binary string to numpy array."""
return np.array([int(i) for i in binstr])
def np2bin(npbin):
"""Convert a binary numpy array to string. """
"""Convert a binary numpy array to string."""
return np.array2string(npbin, separator='')[1:-1]
def df(msg):
"""Decode Downlink Format vaule, bits 1 to 5."""
msgbin = hex2bin(msg)
return min( bin2int(msgbin[0:5]) , 24 )
return min(bin2int(msgbin[0:5]), 24)
def crc(msg, encode=False):
"""Mode-S Cyclic Redundancy Check
Detect if bit error occurs in the Mode-S message
"""Mode-S Cyclic Redundancy Check.
Detect if bit error occurs in the Mode-S message. When encode option is on,
the checksum is generated.
Args:
msg (string): 28 bytes hexadecimal message string
encode (bool): True to encode the date only and return the checksum
Returns:
int: message checksum, or partity bits (encoder)
"""
"""
if encode:
msg = msg[:-6] + "000000"
@ -55,27 +68,53 @@ def crc(msg, encode=False):
return reminder_int
def crc_legacy(msg, encode=False):
"""Mode-S Cyclic Redundancy Check. (Legacy code, slow)."""
# the polynominal generattor code for CRC [1111111111111010000001001]
generator = np.array([1,1,1,1,1,1,1,1,1,1,1,1,1,0,1,0,0,0,0,0,0,1,0,0,1])
ng = len(generator)
msgnpbin = bin2np(hex2bin(msg))
if encode:
msgnpbin[-24:] = [0] * 24
# loop all bits, except last 24 piraty bits
for i in range(len(msgnpbin)-24):
if msgnpbin[i] == 0:
continue
# perform XOR, when 1
msgnpbin[i:i+ng] = np.bitwise_xor(msgnpbin[i:i+ng], generator)
# last 24 bits
reminder = bin2int(np2bin(msgnpbin[-24:]))
return reminder
def floor(x):
""" Mode-S floor function
"""Mode-S floor function.
Defined as the greatest integer value k, such that k <= x
Defined as the greatest integer value k, such that k <= x
For example: floor(3.6) = 3 and floor(-3.6) = -4
eg.: floor(3.6) = 3, while floor(-3.6) = -4
"""
return int(np.floor(x))
def icao(msg):
"""Calculate the ICAO address from an Mode-S message
with DF4, DF5, DF20, DF21
"""Calculate the ICAO address from an Mode-S message.
Applicable only with DF4, DF5, DF20, DF21 messages.
Args:
msg (String): 28 bytes hexadecimal message string
Returns:
String: ICAO address in 6 bytes hexadecimal string
"""
"""
DF = df(msg)
if DF in (11, 17, 18):
@ -91,8 +130,7 @@ def icao(msg):
def is_icao_assigned(icao):
""" Check whether the ICAO address is assigned (Annex 10, Vol 3)"""
"""Check whether the ICAO address is assigned (Annex 10, Vol 3)."""
if (icao is None) or (not isinstance(icao, str)) or (len(icao)!=6):
return False
@ -127,7 +165,7 @@ def typecode(msg):
def cprNL(lat):
"""NL() function in CPR decoding"""
"""NL() function in CPR decoding."""
if lat == 0:
return 59
@ -145,8 +183,11 @@ def cprNL(lat):
NL = floor(nl)
return NL
def idcode(msg):
"""Computes identity (squawk code) from DF5 or DF21 message, bit 20-32.
"""Compute identity (squawk code).
Applicable only for DF5 or DF21 messages, bit 20-32.
credit: @fbyrkjeland
Args:
@ -154,8 +195,8 @@ def idcode(msg):
Returns:
string: squawk code
"""
"""
if df(msg) not in [5, 21]:
raise RuntimeError("Message must be Downlink Format 5 or 21.")
@ -184,7 +225,9 @@ def idcode(msg):
def altcode(msg):
"""Computes the altitude from DF4 or DF20 message, bit 20-32.
"""Compute the altitude.
Applicable only for DF4 or DF20 message, bit 20-32.
credit: @fbyrkjeland
Args:
@ -192,8 +235,8 @@ def altcode(msg):
Returns:
int: altitude in ft
"""
"""
if df(msg) not in [0, 4, 16, 20]:
raise RuntimeError("Message must be Downlink Format 0, 4, 16, or 20.")
@ -255,7 +298,7 @@ def gray2alt(codestr):
def gray2int(graystr):
"""Convert greycode to binary"""
"""Convert greycode to binary."""
num = bin2int(graystr)
num ^= (num >> 8)
num ^= (num >> 4)
@ -265,18 +308,19 @@ def gray2int(graystr):
def data(msg):
"""Return the data frame in the message, bytes 9 to 22"""
"""Return the data frame in the message, bytes 9 to 22."""
return msg[8:-6]
def allzeros(msg):
"""check if the data bits are all zeros
"""Check if the data bits are all zeros.
Args:
msg (String): 28 bytes hexadecimal message string
Returns:
bool: True or False
"""
d = hex2bin(data(msg))
@ -287,10 +331,11 @@ def allzeros(msg):
def wrongstatus(data, sb, msb, lsb):
"""Check if the status bit and field bits are consistency. This Function
is used for checking BDS code versions.
"""
"""Check if the status bit and field bits are consistency.
This Function is used for checking BDS code versions.
"""
# status bit, most significant bit, least significant bit
status = int(data[sb-1])
value = bin2int(data[msb-1:lsb])

View File

@ -1,13 +1,15 @@
GENERATOR = [int("11111111", 2), int("11111010", 2), int("00000100", 2), int("10000000", 2)]
"""Compute CRC checksum of a hexadecimal string."""
GENERATOR = [
int("11111111", 2), int("11111010", 2),
int("00000100", 2), int("10000000", 2)
]
class BytesWrapper:
def __init__(self, hex: str):
if len(hex) % 2 == 1:
hex += '0'
self._bytes = [b for b in bytes.fromhex(hex)]
def __init__(self, hexstr):
self._bytes = [int(hexstr[i:i+2], 16) for i in range(0, len(hexstr), 2)]
def byte_count(self):
return len(self._bytes) - 3
@ -20,13 +22,13 @@ class BytesWrapper:
def apply_matrix(self, byte_index, bit_index):
self._bytes[byte_index] = self._bytes[byte_index] ^ (GENERATOR[0] >> bit_index)
self._bytes[byte_index + 1] = self._bytes[byte_index + 1] ^ \
(0xFF & ((GENERATOR[0] << 8 - bit_index) | (GENERATOR[1] >> bit_index)))
(0xFF & ((GENERATOR[0] << 8 - bit_index) | (GENERATOR[1] >> bit_index)))
self._bytes[byte_index + 2] = self._bytes[byte_index + 2] ^ \
(0xFF & ((GENERATOR[1] << 8 - bit_index) | (GENERATOR[2] >> bit_index)))
(0xFF & ((GENERATOR[1] << 8 - bit_index) | (GENERATOR[2] >> bit_index)))
self._bytes[byte_index + 3] = self._bytes[byte_index + 3] ^ \
(0xFF & ((GENERATOR[2] << 8 - bit_index) | (GENERATOR[3] >> bit_index)))
(0xFF & ((GENERATOR[2] << 8 - bit_index) | (GENERATOR[3] >> bit_index)))
def get_suffix(self) -> int:
def get_suffix(self):
return (self._bytes[-3] << 16) | (self._bytes[-2] << 8) | self._bytes[-1]

View File

@ -1,10 +1,14 @@
from pyModeS import common
def test_hex2bin():
def test_conversions():
assert common.hex2bin('6E406B') == "011011100100000001101011"
assert common.bin2hex('011011100100000001101011') == "6E406B"
assert common.int2hex(11160538) == "AA4BDA"
def test_crc_decode():
assert common.crc_legacy("8D406B902015A678D4D220AA4BDA") == 0
assert common.crc("8D406B902015A678D4D220AA4BDA") == 0
assert common.crc('8d8960ed58bf053cf11bc5932b7d') == 0
assert common.crc('8d45cab390c39509496ca9a32912') == 0
@ -22,7 +26,7 @@ def test_crc_decode():
def test_crc_encode():
parity = common.crc("8D406B902015A678D4D220AA4BDA", encode=True)
assert int("AA4BDA", 16) == parity
assert common.int2hex(parity) == "AA4BDA"
def test_icao():
assert common.icao("8D406B902015A678D4D220AA4BDA") == "406B90"