oddstr13/srne-mqtt
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Add timeout to read function

Browse source 
Fix __exit__
This commit is contained in:
Odd Stråbø 2021-11-05 03:01:43 +01:00
parent d60a07dfcc
commit cf294d18d6
2 changed files with 39 additions and 15 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

35
solar_ble.py
View file

@ -24,6 +24,11 @@ INTERVAL = 15
WRITE_DEVICE = "0000ffd1-0000-1000-8000-00805f9b34fb" WRITE_DEVICE = "0000ffd1-0000-1000-8000-00805f9b34fb"
READ_DEVICE = "0000fff1-0000-1000-8000-00805f9b34fb" READ_DEVICE = "0000fff1-0000-1000-8000-00805f9b34fb"
ACTION_READ = 0x03
ACTION_WRITE = 0x03
POSSIBLE_MARKER = (0x01, 0xFD, 0xFE, 0xFF)
# get(255, 12, 2) # get(255, 12, 2)
# "ff 03 00 0c 00 02" # "ff 03 00 0c 00 02"
CMD_GET_1 = b"\xff\x03\x00\x0c\x00\x02" CMD_GET_1 = b"\xff\x03\x00\x0c\x00\x02"
@ -316,14 +321,14 @@ class BTLEUart(io.RawIOBase):
left = timeout - (time.time() - start) left = timeout - (time.time() - start)
if left < 0: if left < 0:
break break
self.device.waitForNotifications(0.1) self._poll()
try: try:
self._read_buffer.extend(self.delegate.queue.get_nowait()) self._read_buffer.extend(self.delegate.queue.get_nowait())
except queue.Empty: except queue.Empty:
pass pass
try: try:
while True: while True:
self.device.waitForNotifications(0.01) self._poll()
self._read_buffer.extend(self.delegate.queue.get_nowait()) self._read_buffer.extend(self.delegate.queue.get_nowait())
except queue.Empty: except queue.Empty:
@ -334,7 +339,7 @@ class BTLEUart(io.RawIOBase):
left = timeout - (time.time() - start) left = timeout - (time.time() - start)
if left < 0: if left < 0:
break break
self.device.waitForNotifications(0.1) self._poll()
try: try:
self._read_buffer.extend(self.delegate.queue.get_nowait()) self._read_buffer.extend(self.delegate.queue.get_nowait())
except queue.Empty: except queue.Empty:
@ -360,6 +365,22 @@ class BTLEUart(io.RawIOBase):
def readall(self) -> bytes: def readall(self) -> bytes:
return self._read() return self._read()
def read(
self, size: Optional[int] = None, timeout: Optional[float] = None
) -> Optional[bytes]:
if timeout:
_timeout = self.timeout
self.timeout = timeout
if size is None:
res = super().read()
else:
res = super().read(size)
if timeout:
self.timeout = _timeout
return res
def write(self, b: "ReadableBuffer") -> Optional[int]: def write(self, b: "ReadableBuffer") -> Optional[int]:
self._ensure_connected() self._ensure_connected()
if TYPE_CHECKING: if TYPE_CHECKING:
@ -373,8 +394,9 @@ class BTLEUart(io.RawIOBase):
def __enter__(self): def __enter__(self):
return self return self
def __exit__(self): def __exit__(self, type, value, traceback):
self.device.disconnect() self.device.disconnect()
del self.device
def seekable(self) -> bool: def seekable(self) -> bool:
return False return False
@ -403,8 +425,9 @@ def write(fh, data):
fh.write(data + bcrc) fh.write(data + bcrc)
def construct_request(address, words=1): def construct_request(address, words=1, action=ACTION_READ, marker=0xFF):
return struct.pack("!BBHH", 0xFF, 0x03, address, words) assert marker in {POSSIBLE_MARKER}, f"marker should be one of {POSSIBLE_MARKER}"
return struct.pack("!BBHH", marker, action, address, words)
def poll(dev: btle.Peripheral, timeout: float = 1) -> bool: def poll(dev: btle.Peripheral, timeout: float = 1) -> bool:

19
test_bleuart.py
View file

@ -1,16 +1,17 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
from solar_ble import MAC, BTLEUart, construct_request, write from solar_ble import MAC, BTLEUart, construct_request, write
x = BTLEUart(MAC) with BTLEUart(MAC, timeout=1) as x:
print(x) print(x)
write(x, construct_request(0x0C, 8))
print(x.read(3)) write(x, construct_request(0x0E, words=3))
print(x.read(8 * 2)) x.read(3, timeout=1)
print(x.read(2)) print(x.read(6, timeout=0.01))
x.read(2, timeout=0.01)
print(x.timeout)
x.timeout = 10 # x.timeout = 2
print(x.read()) # print(x.read())
print(x.read(1)) # print(x.read(1))