import struct
class PecosMessage:
"""
A class to generate bytearrays for Pecos Message Structure (PMS) messages.
"""
def __init__(self, imei: int, msg_id: int, datetime: int, payload: bytes, encrypted: bool = False):
"""
Initialize the PecosMessage.
:param imei: Device IMEI (signed 64-bit integer, 8 bytes)
:param msg_id: Message ID (unsigned 16-bit integer, 2 bytes)
:param datetime: Epoch time in milliseconds (unsigned 64-bit integer, 8 bytes)
:param payload: The message payload (bytes)
:param encrypted: Whether the payload is encrypted (boolean)
"""
self.imei = imei
self.msg_id = msg_id
self.datetime = datetime
self.payload = payload
self.encrypted = encrypted
self.checksum = self.calculate_payload_checksum()
self.pecos_checksum = self.calculate_pecos_checksum()
def calculate_payload_checksum(self) -> int:
"""
Calculate a simple checksum over the payload.
:return: Checksum as a single byte
"""
return sum(self.payload) % 256
def calculate_pecos_checksum(self) -> int:
"""
Calculate a checksum over the entire PECOS message, including the header and payload.
:return: Checksum as a single byte
"""
encrypted_flag = 1 if self.encrypted else 0
header = struct.pack("!qHQB", self.imei, self.msg_id, self.datetime, encrypted_flag)
full_message = header + self.payload
return sum(full_message) % 256
def to_bytearray(self) -> bytearray:
"""
Convert the message to a bytearray.
:return: Bytearray representing the encoded message
"""
encrypted_flag = 1 if self.encrypted else 0
header = struct.pack("!qHQB", self.imei, self.msg_id, self.datetime, encrypted_flag)
checksum = struct.pack("!B", self.checksum)
pecos_checksum = struct.pack("!B", self.pecos_checksum)
return bytearray(header + self.payload + checksum + pecos_checksum)
class PSNMessage(PecosMessage):
"""
A subclass for PSN (Position Short) messages.
"""
def __init__(self, imei: int, msg_id: int, brevity: int, pdop: float, lat: float, lon: float, alt: float, crs: float, spd: float):
"""
Initialize a PSN message with position data.
:param imei: Device IMEI (signed 64-bit integer, 8 bytes)
:param msg_id: Message ID (unsigned 16-bit integer, 2 bytes)
:param brevity: Brevity code (int, 4 bytes)
:param pdop: Position Dilution of Precision (float, 4 bytes)
:param lat: Latitude (double, 8 bytes)
:param lon: Longitude (double, 8 bytes)
:param alt: Altitude in meters (float, 4 bytes)
:param crs: Course over ground (float, 4 bytes)
:param spd: Speed over ground (float, 4 bytes)
"""
brevity = int(brevity)
pdop = float(pdop)
lat = float(lat)
lon = float(lon)
alt = float(alt)
crs = float(crs)
spd = float(spd)
try:
payload = struct.pack("!Iddfff", brevity, lat, lon, pdop, alt, crs, spd)
except struct.error as e:
raise ValueError(f"Struct packing error: {e}")
super().__init__(imei=imei, msg_id=msg_id, datetime=0, payload=payload, encrypted=False)
# Debugging Helper
if __name__ == "__main__":
test_imei = -1 # Example IMEI
test_msg_id = 0x5033 # Example Message ID
test_brevity = 1
test_pdop = 2.5
test_lat = 37.7749
test_lon = -122.4194
test_alt = 30.0
test_crs = 180.0
test_spd = 15.5
# Create message and print bytearray
psn = PSNMessage(test_imei, test_msg_id, test_brevity, test_pdop, test_lat, test_lon, test_alt, test_crs, test_spd)
print(psn.to_bytearray().hex())