Miko/Hippolyzer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

reader well on its way

Browse Source
This commit is contained in:
locklainn.linden
2008-07-28 18:08:28 +00:00
committed by Salad Dais
parent 67bcfb4b1b
commit 795fc5fce8
7 changed files with 183 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

46
pyogp/lib/base/message/data_packer.py
View File

@@ -5,25 +5,25 @@ from pyogp.lib.base.message.message_types import MsgType
class DataPacker(object):
def __init__(self):
self.packer = {}
self.packer[MsgType.MVT_VARIABLE] = self.__pack_string
self.packer[MsgType.MVT_U8] = '>B'
self.packer[MsgType.MVT_U16] = '>H'
self.packer[MsgType.MVT_U32] = '>I'
self.packer[MsgType.MVT_U64] = '>Q'
self.packer[MsgType.MVT_S8] = '>b'
self.packer[MsgType.MVT_S16] = '>h'
self.packer[MsgType.MVT_S32] = '>i'
self.packer[MsgType.MVT_S64] = '>q'
self.packer[MsgType.MVT_F32] = '>f'
self.packer[MsgType.MVT_F64] = '>d'
self.packer[MsgType.MVT_LLVector3] = self.__pack_vector3
self.packer[MsgType.MVT_LLVector3d] = self.__pack_vector3d
self.packer[MsgType.MVT_LLVector4] = self.__pack_vector4
self.packer[MsgType.MVT_LLQuaternion] = self.__pack_quat
self.packer[MsgType.MVT_LLUUID] = self.__pack_uuid
self.packer[MsgType.MVT_BOOL] = '>B'
self.packer[MsgType.MVT_IP_ADDR] = self.__pack_string
self.packer[MsgType.MVT_IP_PORT] = self.__pack_string
self.packer[MsgType.MVT_VARIABLE] = self.__pack_string
self.packer[MsgType.MVT_U8] = '>B'
self.packer[MsgType.MVT_U16] = '>H'
self.packer[MsgType.MVT_U32] = '>I'
self.packer[MsgType.MVT_U64] = '>Q'
self.packer[MsgType.MVT_S8] = '>b'
self.packer[MsgType.MVT_S16] = '>h'
self.packer[MsgType.MVT_S32] = '>i'
self.packer[MsgType.MVT_S64] = '>q'
self.packer[MsgType.MVT_F32] = '>f'
self.packer[MsgType.MVT_F64] = '>d'
self.packer[MsgType.MVT_LLVector3] = self.__pack_vector3
self.packer[MsgType.MVT_LLVector3d] = self.__pack_vector3d
self.packer[MsgType.MVT_LLVector4] = self.__pack_vector4
self.packer[MsgType.MVT_LLQuaternion] = self.__pack_quat
self.packer[MsgType.MVT_LLUUID] = self.__pack_uuid
self.packer[MsgType.MVT_BOOL] = '>B'
self.packer[MsgType.MVT_IP_ADDR] = self.__pack_string
self.packer[MsgType.MVT_IP_PORT] = self.__pack_string
def pack_data(self, data, data_type):
if data_type in self.packer:
@@ -40,18 +40,18 @@ class DataPacker(object):
return struct.pack('>' + str(size) + tp, *tup)
def __pack_vector3(self, vec):
return pack_tuple(vec, 'f')
return __pack_tuple(vec, 'f')
def __pack_vector3d(self, vec):
return pack_tuple(vec, 'd')
return __pack_tuple(vec, 'd')
def __pack_vector4(self, vec):
return pack_tuple(vec, 'f')
return __pack_tuple(vec, 'f')
def __pack_quat(self, quat):
#first, pack to vector3
vec = quat_to_vec3(quat)
return pack_vector3(vec)
return __pack_vector3(vec)
def __pack_uuid(self, uuid):
return uuid.bytes

3
pyogp/lib/base/message/message_template.py
View File

@@ -32,6 +32,7 @@ import pprint
#the packet flags and the sequence number. After the ID, then comes the header
#NOTE: This will be moved into a messaging system eventually
PACKET_ID_LENGTH = 6
PHL_OFFSET = 5
#this probably needs to implement an interface so it can be serialized
class MsgData(object):
@@ -122,6 +123,8 @@ class MessageTemplate(object):
self.block_map = {}
#this is the function or object that will handle this type of message
self.handler = None
self.received_count = 0
self.name = name
self.frequency = None
self.msg_num = 0

2
pyogp/lib/base/message/message_template_parser.py
View File

@@ -88,7 +88,7 @@ class MessageTemplateParser(object):
current_template.msg_num = msg_num
current_template.msg_num_hex = msg_num_hex
msg_trust = None
if parts[3] == 'Trusted':
msg_trust = MsgTrust.LL_TRUSTED

137
pyogp/lib/base/message/message_template_reader.py
View File

@@ -7,13 +7,16 @@ from pyogp.lib.base.message.message_template import MsgData, MsgBlockData, \
MsgVariableData
#import pyogp.lib.base.message_types
from pyogp.lib.base.message.message_types import MsgType, MsgBlockType, MsgFrequency, sizeof
from pyogp.lib.base.message.data_packer import DataPacker
from pyogp.lib.base.message.data_unpacker import DataUnpacker
class MessageTemplateReader(object):
def __init__(self, template_dict):
self.template_dict = template_dict
self.current_template = None
self.unpacker = DataUnpacker()
self.receive_size = -1
self.current_msg = None
self.current_block = None
@@ -22,29 +25,130 @@ class MessageTemplateReader(object):
def validate_message(self, message_buffer, buffer_size):
""" Determines if the message follows a given template. """
header = message_buffer[message_template.PACKET_ID_LENGTH:]
self.current_template = self.decode_header(header)
if self.current_template == None:
return False
self.receive_size = buffer_size
if self.__decode_template(message_buffer, buffer_size) == True:
self.current_template.received_count += 1
return True
return True
return False
def read_message(self, message_buffer):
""" Goes through the message and decodes all the data in it. """
pass
return decode_data(message_buffer)
def decode_template(self, message_buffer, buffer_size):
def get_data(self, block_name, var_name, data_type, block_number = 0):
if self.receive_size == -1:
#error
return None
if self.current_msg == None:
#error
return None
block_list = self.current_msg.get_block(block_name)
if block_number not in block_list:
#error: block not in message
return None
block_data = block_list[block_number]
var_data = block_data.get_variable(var_name)
if var_data.type != data_type:
#error: variable types don't match
return None
return var_data.data
def __decode_template(self, message_buffer, buffer_size):
""" Determines the template that the message in the buffer
appears to be using. """
pass
header = message_buffer[message_template.PACKET_ID_LENGTH:]
self.current_template = __decode_header(header)
if self.current_template != None:
return True
def decode_header(self, header):
frequency = decode_frequency(header)
num = decode_num(header)
return False
def __decode_data(self, data):
if self.current_template == None:
raise Exception('Attempting to decode data without validating it')
if self.current_msg != None:
print 'WARNING: Attempting to decode data without clearing the last message'
self.current_msg = None
#at the offset position, the messages stores the offset to where the
#payload begins (may be extra header information)
offset = data[message_template.PHL_OFFSET]
decode_pos = message_template.PACKET_ID_LENGTH + \
self.current_template.frequency + \
offset
self.current_msg = MsgData(self.current_template.name)
for block in self.current_template.blocks:
repeat_count = 0
if blocks.type == MsgBlockType.MBT_SINGLE:
repeat_count = 1
elif blocks.type == MsgBlockType.MBT_MULTIPLE:
repeat_count = block.number
elif blocks.type == MsgBlockType.MBT_VARIABLE:
#if the block type is VARIABLE, then the current position
#will be the repeat count written in
count = data[decode_pos]
decode_pos += 1
else:
#error
return False
for i in range(repeat_count):
block_data = MsgBlockData(block.name)
block_data.number = repeat_count
self.current_block = block_data
self.current_msg.add_block(self.current_block)
for variable in block.variables:
var_data = MsgVariableData(variable.name, variable.type)
self.current_block.add_variable(var_data)
var_size = variable.size
if variable.type == MsgType.MVT_VARIABLE:
#this isn't the size of the data, but the max bytes
#the data can be
#need to copy because we have to advance our decode_pos
#afterwards
data_size = var_size
if data_size == 1:
var_size = struct.unpack('>B', data[decode_pos:decode_pos+1])
elif data_size == 2:
var_size = struct.unpack('>H', data[decode_pos:decode_pos+2])
elif data_size == 4:
var_size = struct.unpack('>I', data[decode_pos:decode_pos+4])
else:
raise Exception('Attempting to read variable with unknown size \
of ' + str(data_size))
decode_pos += data_size
unpacked_data = self.unpacker.unpack(data[decode_pos:decod_pos+var_size],variable.type)
self.current_block.add_data(variable.name, unpacked_data, var_size)
decode_pos += var_size
if len(self.current_msg.block_map) <= 0 and len(self.current_template.blocks) > 0:
#error, blank message
return False
return True
def __decode_header(self, header):
frequency = __decode_frequency(header)
num = __decode_num(header)
return self.template_dict.get_template_by_pair(frequency, num)
def decode_frequency(self, header):
def __decode_frequency(self, header):
#if it is not a high
if header[0] == '\xFF':
#if it is not a medium frequency message
@@ -64,8 +168,8 @@ class MessageTemplateReader(object):
return None
def decode_num(self, header):
frequency = decode_frequency(header)
def __decode_num(self, header):
frequency = __decode_frequency(header)
if frequency == 'Low':
return struct.unpack('>B', header[2:4])[0] #int("0x"+ByteToHex(header[2:4]).replace(' ', ''),16)
@@ -81,3 +185,6 @@ class MessageTemplateReader(object):
else:
return None

24
pyogp/lib/base/message/message_types.py
View File

@@ -4,18 +4,12 @@ class MsgBlockType(object):
MBT_VARIABLE = range(3)
#pack flags
#= '\x80'
#= '\x80'
#= '\x40'
#= '\x20'
#= '\x10'
#= '\x00'
class PackFlags(object):
LL_ZERO_CODE_FLAG, \
LL_RELIABLE_FLAG, \
LL_RESENT_FLAG, \
LL_ACK_FLAG, \
LL_NONE = range(5)
LL_ZERO_CODE_FLAG = '\x80'
LL_RELIABLE_FLAG = '\x40'
LL_RESENT_FLAG = '\x20'
LL_ACK_FLAG = '\x10'
LL_NONE = '\x00'
#frequency for messages
#= '\xFF\xFF\xFF'
@@ -23,10 +17,10 @@ class PackFlags(object):
#= '\xFF'
#= ''
class MsgFrequency(object):
FIXED_FREQUENCY_MESSAGE, \
LOW_FREQUENCY_MESSAGE, \
MEDIUM_FREQUENCY_MESSAGE, \
HIGH_FREQUENCY_MESSAGE = range(4)
FIXED_FREQUENCY_MESSAGE = -1 #marking it
LOW_FREQUENCY_MESSAGE = 4
MEDIUM_FREQUENCY_MESSAGE = 2
HIGH_FREQUENCY_MESSAGE = 1
class MsgTrust(object):
LL_TRUSTED, \

2
pyogp/lib/base/tests/test_template_parser.py
View File

@@ -82,7 +82,7 @@ class TestTemplates(unittest.TestCase):
def test_template_low(self):
template = self.msg_dict['AddCircuitCode']
hx = template.msg_num_hex
assert hx == '\xff\xff\x00\x02', "Expected: '\xff\xff\x00\x02' Returned: " + repr(hx)
assert hx == '\xff\xff\x00\x02', "Expected: " + r'\xff\xff\x00\x02' + " Returned: " + repr(hx)
def test_deprecated(self):
template = self.msg_dict['ObjectPosition']

26
pyogp/lib/base/tests/test_template_reader.py
View File

@@ -20,9 +20,31 @@ class TestTemplateReader(unittest.TestCase):
self.template_list = parser.message_templates
self.template_dict = TemplateDictionary(self.template_list)
self.builder = MessageTemplateBuilder(self.template_dict)
self.reader = MessageTemplateReader(self.template_dict)
def test_validation(self):
self.builder.new_message('CompletePingCheck')
self.builder.next_block('PingID')
self.builder.add_data('PingID', 0x01, MsgType.MVT_U8)
message, size = self.builder.build_message()
assert self.reader.validate_message(message, size) == True, \
"Validation failed"
def test_validation_fail(self):
assert self.reader.validate_message('\x00\x00\x00\x00', 4) == False, \
"Validation failed"
def test_validation(self):
self.builder.new_message('CompletePingCheck')
self.builder.next_block('PingID')
self.builder.add_data('PingID', 0x01, MsgType.MVT_U8)
message, size = self.builder.build_message()
self.reader.validate_message(message, size)
msg = self.reader.read_message(message)
def test_suite():
"""def test_suite():
from unittest import TestSuite, makeSuite
suite = TestSuite()
suite.addTest(makeSuite(TestTemplateReader))
return suite
return suite"""