mirrors
/
tuhi
mirror of https://github.com/tuhiproject/tuhi.git
1
0
Fork 0
Code Issues Releases Wiki Activity

protocol: store the interaction name in the NordicData where appropriate 

This means the data logger can then access that name without needing a
separate lookup table.

Signed-off-by: Peter Hutterer <peter.hutterer@who-t.net>
pull/193/head
Peter Hutterer 2019-08-23 13:38:18 +10:00
parent e30787234c
commit 79480b867d
2 changed files with 12 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
tuhi/protocol.py
View File

@ -324,17 +324,21 @@ class NordicData(list):
The data length of this message. This field is guaranteed to be
equivalent to len(data) or an exception is raised.
.. attribute:: name
The name of this message, may be None
'''
def __init__(self, bs):
def __init__(self, bs, name=None):
data = bs[2:]
super().__init__(data)
self.opcode = bs[0]
self.length = bs[1]
self.name = name
if self.length != len(data):
raise UnexpectedDataError(bs, f'Invalid data: length field {self.length}, data length is {len(data)}')
def __repr__(self):
return f'{self.opcode:02x} / {self.length:02x} / {as_hex_string(self)}'
return f'{self.name if self.Name else "UNKNOWN"}{self.opcode:02x} / {self.length:02x} / {as_hex_string(self)}'
class ProtocolError(Exception):
@ -543,7 +547,8 @@ class Msg(object):
if self.opcode == Msg.OPCODE_NOOP:
return self # allow chaining
self.request = NordicData([self.opcode, len(self.args or []), *(self.args or [])])
self.request = NordicData([self.opcode, len(self.args or []), *(self.args or [])],
name=self.interaction.name)
self.reply = self._callback(request=self.request if self.requires_request else None,
requires_reply=self.requires_reply,
timeout=self.timeout or None,

28
tuhi/wacom.py
View File

@ -163,24 +163,6 @@ class DataLogger(object):
def recv(self, data):
return self.parent._recv(self.source, data)
commands = {
0xb1: 'set mode',
0xb6: 'set time',
0xb7: 'get firmware',
0xb9: 'read battery info',
0xbb: 'get/set name',
0xc1: 'check for data',
0xc3: 'start reading',
0xc5: 'fetch data',
0xc8: 'end of data',
0xca: 'ack transaction',
0xcc: 'fetch data',
0xea: 'get dimensions',
0xe5: 'finish registering',
0xe6: 'check connection',
0xdb: 'get name',
}
def __init__(self, bluez_device):
self.logger = logging.getLogger('tuhi.fw')
self.device = bluez_device
@ -240,20 +222,18 @@ class DataLogger(object):
self.logger.debug(f'{self.btaddr}: RX {source} <-- {convert(data)}')
self._init_file()
if data[0] in self.commands:
self.logfile.write(f'# {self.commands[data[0]]}\n')
self.logfile.write(f' - recv: {list2hexlist(data)}\n')
if source != 'NORDIC':
self.logfile.write(f' source: {source}\n')
def _request(self, source, request):
if request.opcode in self.commands:
self.logger.debug(f'command: {self.commands[request.opcode]}')
if request.name:
self.logger.debug(f'command: {request.name}')
self.logger.debug(f'{self.btaddr}: TX {source} --> {request.opcode:02x} / {len(request):02x} / {list2hex(request)}')
self._init_file()
if request.opcode in self.commands:
self.logfile.write(f'# {self.commands[request.opcode]}\n')
if request.name:
self.logfile.write(f'# {request.name}\n')
data = [request.opcode, len(request), *request]
self.logfile.write(f' - send: {list2hexlist(data)}\n')