diff options
Diffstat (limited to '.venv/lib/python3.12/site-packages/msg_parser/data_models.py')
| -rw-r--r-- | .venv/lib/python3.12/site-packages/msg_parser/data_models.py | 213 |
1 files changed, 213 insertions, 0 deletions
diff --git a/.venv/lib/python3.12/site-packages/msg_parser/data_models.py b/.venv/lib/python3.12/site-packages/msg_parser/data_models.py new file mode 100644 index 00000000..3e2e2ef5 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/msg_parser/data_models.py @@ -0,0 +1,213 @@ +# coding=utf-8 +from datetime import datetime +from datetime import timedelta +from struct import unpack + +from .properties import DATA_TYPE_MAP + + +class DataModel(object): + def __init__(self): + self.data_type_name = None + + @staticmethod + def lookup_data_type_name(data_type): + return DATA_TYPE_MAP.get(data_type) + + def get_value(self, data_value, data_type_name=None, data_type=None): + + if data_type_name: + self.data_type_name = data_type_name + elif data_type: + self.data_type_name = self.lookup_data_type_name(data_type) + else: + raise Exception( + "required arguments not provided to the constructor of the class." + ) + + if not hasattr(self, self.data_type_name): + return None + + value = getattr(self, self.data_type_name)(data_value) + return value + + @staticmethod + def PtypUnspecified(data_value): + return data_value + + @staticmethod + def PtypNull(_): + return None + + @staticmethod + def PtypInteger16(data_value): + return int(data_value.encode("hex"), 16) + + @staticmethod + def PtypInteger32(data_value): + return int(data_value.encode("hex"), 32) + + @staticmethod + def PtypFloating32(data_value): + return unpack("f", data_value)[0] + + @staticmethod + def PtypFloating64(data_value): + return unpack("d", data_value)[0] + + @staticmethod + def PtypCurrency(data_value): + return data_value + + @staticmethod + def PtypFloatingTime(data_value): + return data_value + + @staticmethod + def PtypErrorCode(data_value): + return unpack("I", data_value)[0] + + @staticmethod + def PtypBoolean(data_value): + return unpack("B", data_value[0])[0] != 0 + + @staticmethod + def PtypObject(data_value): + if data_value and b"\x00" in data_value: + data_value = data_value.replace(b"\x00", b"") + return data_value + + @staticmethod + def PtypInteger64(data_value): + return unpack("q", data_value)[0] + + @staticmethod + def PtypString8(data_value): + if data_value and b"\x00" in data_value: + data_value = data_value.replace(b"\x00", b"") + return data_value + + @staticmethod + def PtypString(data_value): + if data_value: + data_value = data_value.decode("utf-16-le", errors="ignore").replace( + "\x00", "" + ) + return data_value + + @staticmethod + def PtypTime(data_value): + return get_time(data_value) + + @staticmethod + def PtypGuid(data_value): + return data_value + + @staticmethod + def PtypServerId(data_value): + return data_value + + @staticmethod + def PtypRestriction(data_value): + return data_value + + @staticmethod + def PtypRuleAction(data_value): + return data_value + + @staticmethod + def PtypBinary(data_value): + if data_value and b"\x00" in data_value: + data_value = data_value.replace(b"\x00", b"") + return data_value + + @staticmethod + def PtypMultipleInteger16(data_value): + entry_count = int(len(data_value) / 2) + return [unpack("h", bytes[i * 2 : (i + 1) * 2])[0] for i in range(entry_count)] + + @staticmethod + def PtypMultipleInteger32(data_value): + entry_count = int(len(data_value) / 4) + return [unpack("i", bytes[i * 4 : (i + 1) * 4])[0] for i in range(entry_count)] + + @staticmethod + def PtypMultipleFloating32(data_value): + entry_count = int(len(data_value) / 4) + return [unpack("f", bytes[i * 4 : (i + 1) * 4])[0] for i in range(entry_count)] + + @staticmethod + def PtypMultipleFloating64(data_value): + entry_count = int(len(data_value) / 8) + return [unpack("d", bytes[i * 8 : (i + 1) * 8])[0] for i in range(entry_count)] + + @staticmethod + def PtypMultipleCurrency(data_value): + return data_value + + @staticmethod + def PtypMultipleFloatingTime(data_value): + entry_count = int(len(data_value) / 8) + return [ + get_floating_time(bytes[i * 8 : (i + 1) * 8]) for i in range(entry_count) + ] + + @staticmethod + def PtypMultipleInteger64(data_value): + entry_count = int(len(data_value) / 8) + return [unpack("q", bytes[i * 8 : (i + 1) * 8])[0] for i in range(entry_count)] + + @staticmethod + def PtypMultipleString(data_value): + return data_value + # string_list = [] + # for item_bytes in data_value: + # if item_bytes and '\x00' in item_bytes: + # item_bytes = item_bytes.replace('\x00', '') + # string_list.append(item_bytes.decode('utf-16-le')) + # return string_list + + @staticmethod + def PtypMultipleString8(data_value): + return data_value + + @staticmethod + def PtypMultipleTime(data_value): + entry_count = int(len(data_value) / 8) + return [get_time(bytes[i * 8 : (i + 1) * 8]) for i in range(entry_count)] + + @staticmethod + def PtypMultipleGuid(data_value): + entry_count = int(len(data_value) / 16) + return [bytes[i * 16 : (i + 1) * 16] for i in range(entry_count)] + + @staticmethod + def PtypMultipleBinary(data_value): + return data_value + + +def get_floating_time(data_value): + return datetime(year=1899, month=12, day=30) + timedelta( + days=unpack("d", data_value)[0] + ) + + +def get_time(data_value): + return datetime(year=1601, month=1, day=1) + timedelta( + microseconds=unpack("q", data_value)[0] / 10.0 + ) + + +def get_multi_value_offsets(data_value): + ul_count = unpack("I", data_value[:4])[0] + + if ul_count == 1: + rgul_data_offsets = [8] + else: + rgul_data_offsets = [ + unpack("Q", bytes[4 + i * 8 : 4 + (i + 1) * 8])[0] for i in range(ul_count) + ] + + rgul_data_offsets.append(len(data_value)) + + return ul_count, rgul_data_offsets |