33 lines
905 B
Python
33 lines
905 B
Python
from msg.container import MsgContainer
|
|
from ubx.ubx import frame_parse, frame_create
|
|
from ubx.messages.msg import UbxMsg
|
|
from transceiver import Transceiver
|
|
from ubx.listener import UbxListener
|
|
from threading import BoundedSemaphore
|
|
|
|
class UbxQuery(UbxListener):
|
|
def __init__(self, transceiver: Transceiver, msg: UbxMsg):
|
|
UbxListener.__init__(self, msg)
|
|
self.transceiver = transceiver
|
|
self.sema = BoundedSemaphore()
|
|
|
|
def on_recv(self, msg: MsgContainer):
|
|
if self.is_msg(msg):
|
|
data, hdr = frame_parse(msg.data)
|
|
msg_parsed = self.msg.from_bytes(data)
|
|
self.msg = msg_parsed
|
|
self.sema.release()
|
|
|
|
def poll(self):
|
|
frame = frame_create(self.msg.mclass, self.msg.mid, b'')
|
|
self.transceiver.send(frame)
|
|
self.sema.acquire()
|
|
print(self.msg)
|
|
|
|
def query(self, data: bytes):
|
|
frame = frame_create(0x01, 0x35, b'')
|
|
self.transceiver.send(frame)
|
|
self.sema.acquire()
|
|
print(self.msg)
|
|
|