27 lines
864 B
Python
27 lines
864 B
Python
from msg_container import MsgContainer
|
|
from ubx.ubx import frame_parse, frame_parse_hdr, frame_create
|
|
from ubx.msg import UbxMsg
|
|
from transceiver import Transceiver
|
|
from receiver.ubx_listener import UbxListener
|
|
from threading import BoundedSemaphore
|
|
|
|
class UbxQuery(UbxListener):
|
|
def __init__(self, transceiver: Transceiver, parser: UbxMsg, mclass: int|None = None, mid: int|None = None):
|
|
UbxListener.__init__(self, parser, mclass, mid)
|
|
self.transceiver = transceiver
|
|
self.sema = BoundedSemaphore()
|
|
self.msg = UbxMsg()
|
|
|
|
def on_recv(self, msg: MsgContainer):
|
|
if self.is_msg(msg):
|
|
data, hdr = frame_parse(msg.data)
|
|
msg_parsed = self.parser.from_bytes(data)
|
|
self.msg = msg_parsed
|
|
self.sema.release()
|
|
|
|
def query(self, data: bytes):
|
|
frame = frame_create(0x01, 0x35, b'')
|
|
self.transceiver.send(frame)
|
|
self.sema.acquire()
|
|
print(self.msg)
|