-
-
-
-
-
-
+
Fetching data from Radiacode, please wait (it might take up to 1 minute over bluetooth)...
+
Error reading data, try restarting the application...
+
+
+
+
+
+
+
+
-
-
-
-
-
-
-
-
diff --git a/radiacode-examples/webserver.py b/radiacode-examples/webserver.py
index 687e8ac..2930128 100644
--- a/radiacode-examples/webserver.py
+++ b/radiacode-examples/webserver.py
@@ -4,7 +4,6 @@
import pathlib
from aiohttp import web
-
from radiacode import RadiaCode, RealTimeData
@@ -15,47 +14,61 @@ async def handle_index(request):
async def handle_ws(request):
ws = web.WebSocketResponse()
await ws.prepare(request)
- request.app.ws_clients.append(ws)
- async for _ in ws:
- pass
- request.app.ws_clients.remove(ws)
+ request.app['ws_clients'].append(ws)
+
+ try:
+ async for _ in ws:
+ pass
+ except Exception as e:
+ print(f'Unexpected error in websocket: {str(e)}')
+ finally:
+ request.app['ws_clients'].remove(ws)
+
return ws
async def handle_spectrum(request):
- cn = request.app.rc_conn
+ cn = request.app['rc_conn']
accum = request.query.get('accum') == 'true'
- spectrum = cn.spectrum_accum() if accum else cn.spectrum()
+
+ try:
+ spectrum = await (cn.async_spectrum_accum() if accum else cn.async_spectrum())
+ except Exception as e:
+ print(f'Unexpected error while fetching data: {str(e)}')
+ return web.json_response({'error': str(e)}, status=500)
+
# apexcharts can't handle 0 in logarithmic view
spectrum_data = [(channel, cnt if cnt > 0 else 0.5) for channel, cnt in enumerate(spectrum.counts)]
- print('Spectrum updated')
+
return web.json_response(
{
'coef': [spectrum.a0, spectrum.a1, spectrum.a2],
'duration': spectrum.duration.total_seconds(),
'series': [{'name': 'spectrum', 'data': spectrum_data}],
- },
+ }
)
async def handle_spectrum_reset(request):
- cn = request.app.rc_conn
- cn.spectrum_reset()
- print('Spectrum reset')
- return web.json_response({})
+ cn = request.app['rc_conn']
+ await cn.async_spectrum_reset()
+ return web.json_response({'message': 'Spectrum reset'})
async def process(app):
max_history_size = 128
history = []
+
while True:
- databuf = app.rc_conn.data_buf()
+ databuf = await app['rc_conn'].async_data_buf()
+
for v in databuf:
if isinstance(v, RealTimeData):
history.append(v)
history.sort(key=lambda x: x.dt)
history = history[-max_history_size:]
+
jdata = json.dumps(
{
'series': [
@@ -70,37 +83,53 @@ async def process(app):
],
},
)
- print(f'Rates updated, sending to {len(app.ws_clients)} connected clients')
- await asyncio.gather(*[ws.send_str(jdata) for ws in app.ws_clients], asyncio.sleep(1.0))
+
+ print(f'Rates updated, sending to {len(app["ws_clients"])} connected clients')
+ await asyncio.gather(*(ws.send_str(jdata) for ws in app['ws_clients']))
+ await asyncio.sleep(1.0)
async def on_startup(app):
- asyncio.create_task(process(app))
+ app['process_task'] = asyncio.create_task(process(app))
+ app['ws_clients'] = []
-if __name__ == '__main__':
+async def init_connection(app):
+ if app['args'].bluetooth_mac:
+ print('will use Bluetooth connection via MAC address')
+ app['rc_conn'] = await RadiaCode.async_init(bluetooth_mac=app['args'].bluetooth_mac)
+ elif app['args'].bluetooth_uuid:
+ print('will use Bluetooth connection via UUID')
+ app['rc_conn'] = await RadiaCode.async_init(bluetooth_uuid=app['args'].bluetooth_uuid)
+ else:
+ print('will use USB connection')
+ app['rc_conn'] = await RadiaCode.async_init()
+
+
+def main():
parser = argparse.ArgumentParser()
- parser.add_argument('--bluetooth-mac', type=str, required=False, help='bluetooth MAC address of radiascan device')
- parser.add_argument('--listen-host', type=str, required=False, default='0.0.0.0', help='listen host for webserver')
- parser.add_argument('--listen-port', type=int, required=False, default=8080, help='listen port for webserver')
+ parser.add_argument('--bluetooth-mac', type=str, required=False, help='Bluetooth MAC address of radiascan device')
+ parser.add_argument('--bluetooth-uuid', type=str, required=False, help='Bluetooth UUID of radiascan device')
+ parser.add_argument('--listen-host', type=str, required=False, default='127.0.0.1', help='Listen host for webserver')
+ parser.add_argument('--listen-port', type=int, required=False, default=8080, help='Listen port for webserver')
args = parser.parse_args()
app = web.Application()
- app.ws_clients = []
- if args.bluetooth_mac:
- print('will use Bluetooth connection')
- app.rc_conn = RadiaCode(bluetooth_mac=args.bluetooth_mac)
- else:
- print('will use USB connection')
- app.rc_conn = RadiaCode()
-
+ app['args'] = args
+ app.on_startup.append(init_connection)
app.on_startup.append(on_startup)
+
app.add_routes(
[
web.get('/', handle_index),
web.get('/spectrum', handle_spectrum),
web.post('/spectrum/reset', handle_spectrum_reset),
web.get('/ws', handle_ws),
- ],
+ ]
)
+
web.run_app(app, host=args.listen_host, port=args.listen_port)
+
+
+if __name__ == '__main__':
+ main()
diff --git a/radiacode/logger.py b/radiacode/logger.py
new file mode 100644
index 0000000..7e02e1f
--- /dev/null
+++ b/radiacode/logger.py
@@ -0,0 +1,35 @@
+from enum import Enum
+
+
+class LogLevel(Enum):
+ Notification = ('notification', '[\033[1;32m\N{check mark}\033[0m]')
+ Error = ('error', '[\033[1;31m\N{aegean check mark}\033[0m]')
+ Info = ('info', '[\033[1;34m\N{information source}\033[0m]')
+ Warning = ('warning', '[\033[1;35m\N{warning sign}\033[0m]')
+
+
+class Logger:
+ @staticmethod
+ def log(message):
+ print(message)
+
+ @staticmethod
+ def notify(message):
+ Logger._log_level(message, LogLevel.Notification)
+
+ @staticmethod
+ def error(message):
+ Logger._log_level(message, LogLevel.Error)
+
+ @staticmethod
+ def info(message):
+ Logger._log_level(message, LogLevel.Info)
+
+ @staticmethod
+ def warning(message):
+ Logger._log_level(message, LogLevel.Warning)
+
+ @staticmethod
+ def _log_level(message, level):
+ prefix = level.value[1]
+ print(f'{prefix} {message}')
diff --git a/radiacode/radiacode.py b/radiacode/radiacode.py
index e3217c2..e1746c9 100644
--- a/radiacode/radiacode.py
+++ b/radiacode/radiacode.py
@@ -1,8 +1,10 @@
import datetime
import struct
import platform
+import asyncio
from typing import List, Optional, Union
+from radiacode.logger import Logger
from radiacode.bytes_buffer import BytesBuffer
from radiacode.decoders.databuf import decode_VS_DATA_BUF
from radiacode.decoders.spectrum import decode_RC_VS_SPECTRUM
@@ -19,44 +21,208 @@ def spectrum_channel_to_energy(channel_number: int, a0: float, a1: float, a2: fl
class RadiaCode:
_connection: Union[Bluetooth, Usb]
+ """
+ To use this class synchronously:
+ rc = Radiacode()
+ serial = rc.serial_number()
+ """
+
def __init__(
self,
bluetooth_mac: Optional[str] = None,
+ bluetooth_serial: Optional[str] = None,
+ bluetooth_uuid: Optional[str] = None,
serial_number: Optional[str] = None,
- ignore_firmware_compatibility_check: bool = False,
+ ignore_firmware_compatibility_check: Optional[bool] = False,
+ async_init: Optional[bool] = False,
):
self._seq = 0
+ self._usb = False
+ self._async_init = async_init
+ self._ignore_firmware_compatibility_check = ignore_firmware_compatibility_check
+ self.loop = asyncio.new_event_loop()
+ asyncio.set_event_loop(self.loop)
+
+ if platform.system() == 'Darwin' and bluetooth_mac:
+ Logger.warning(
+ 'You would like to establish a bluetooth connection using a MAC address, but you appear to be on a Mac.'
+ )
+ Logger.warning('Apple does not expose Bluetooth MAC addresses anymore, so this method will not work.')
+ Logger.warning('Try connecting to the device using another option (UUID or Serial).')
- # Bluepy doesn't support MacOS: https://github.com/IanHarvey/bluepy/issues/44
- self._bt_supported = platform.system() != 'Darwin'
+ raise Exception('Exception: The chosen connection method does not work on this platform.')
- if bluetooth_mac is not None and self._bt_supported is True:
- self._connection = Bluetooth(bluetooth_mac)
+ if bluetooth_mac is not None:
+ self._connection = Bluetooth(bluetooth_mac=bluetooth_mac)
+ elif bluetooth_serial is not None:
+ self._connection = Bluetooth(bluetooth_serial=bluetooth_serial)
+ elif bluetooth_uuid is not None:
+ self._connection = Bluetooth(bluetooth_uuid=bluetooth_uuid)
else:
+ self._usb = True
+ # Connect over USB
self._connection = Usb(serial_number=serial_number)
- # init
- self.execute(b'\x07\x00', b'\x01\xff\x12\xff')
+ if self._async_init is False:
+ self.loop.run_until_complete(self._init_device())
+
+ """
+ Use this method to create an asynchronous Radiacode() object:
+ rc = Radiacode.async_init()
+ serial = await rc.async_serial_number()
+ """
+
+ @classmethod
+ async def async_init(
+ cls,
+ bluetooth_mac: Optional[str] = None,
+ bluetooth_serial: Optional[str] = None,
+ bluetooth_uuid: Optional[str] = None,
+ serial_number: Optional[str] = None,
+ ignore_firmware_compatibility_check: Optional[bool] = False,
+ ):
+ if platform.system() == 'Darwin' and bluetooth_mac:
+ Logger.warning('You want to connect over bluetooth using a MAC address, but you appear to be on a Mac.')
+ Logger.warning('Apple does not expose Bluetooth MAC addresses anymore, so this method will not work.')
+ Logger.warning('Try connecting to the device using another option (UUID or Serial).')
+
+ raise Exception('Exception: The chosen connection method does not work on this platform.')
+
+ self = cls(
+ bluetooth_mac=bluetooth_mac,
+ bluetooth_serial=bluetooth_serial,
+ bluetooth_uuid=bluetooth_uuid,
+ serial_number=serial_number,
+ ignore_firmware_compatibility_check=ignore_firmware_compatibility_check,
+ async_init=True,
+ )
+
+ await self._init_device()
+ return self
+
+ async def _init_device(self):
+ if self._usb is False:
+ # Connect over Bluetooth
+ await self._connection.connect()
+
+ # Init
+ Logger.info('Initializing Radiacode...')
+ await self._async_execute(b'\x07\x00', b'\x01\xff\x12\xff')
+
self._base_time = datetime.datetime.now()
- self.set_local_time(self._base_time)
- self.device_time(0)
+ await self.async_set_local_time(self._base_time)
+ await self.async_device_time(0)
- (_, (vmaj, vmin, _)) = self.fw_version()
- if ignore_firmware_compatibility_check is False and vmaj < 4 or (vmaj == 4 and vmin < 8):
+ (_, (vmaj, vmin, _)) = await self.async_fw_version()
+ if self._ignore_firmware_compatibility_check is False and vmaj < 4 or (vmaj == 4 and vmin < 8):
raise Exception(
f'Incompatible firmware version {vmaj}.{vmin}, >=4.8 required. Upgrade device firmware or use radiacode==0.2.2'
)
self._spectrum_format_version = 0
- for line in self.configuration().split('\n'):
+ lines = await self.async_configuration()
+
+ for line in lines.split('\n'):
if line.startswith('SpecFormatVersion'):
self._spectrum_format_version = int(line.split('=')[1])
break
+ Logger.notify('Initialization completed')
+
def base_time(self) -> datetime.datetime:
return self._base_time
- def execute(self, reqtype: bytes, args: Optional[bytes] = None) -> BytesBuffer:
+ # def _execute(self, reqtype: bytes, args: Optional[bytes] = None) -> BytesBuffer:
+ # return self.loop.run_until_complete(self._async_execute(reqtype, args))
+
+ def read_request(self, command_id: Union[int, VS, VSFR]) -> BytesBuffer:
+ return self.loop.run_until_complete(self.async_read_request(command_id))
+
+ def write_request(self, command_id: Union[int, VSFR], data: Optional[bytes] = None) -> None:
+ return self.loop.run_until_complete(self.async_write_request(command_id, data))
+
+ def batch_read_vsfrs(self, vsfr_ids: List[VSFR]) -> List[int]:
+ return self.loop.run_until_complete(self.async_batch_read_vsfrs(vsfr_ids))
+
+ def status(self) -> str:
+ return self.loop.run_until_complete(self.async_status())
+
+ def set_local_time(self, dt: datetime.datetime) -> None:
+ return self.loop.run_until_complete(self.async_set_local_time(dt))
+
+ def fw_signature(self) -> str:
+ return self.loop.run_until_complete(self.async_fw_signature())
+
+ def fw_version(self) -> tuple[tuple[int, int, str], tuple[int, int, str]]:
+ return self.loop.run_until_complete(self.async_fw_version())
+
+ def hw_serial_number(self) -> str:
+ return self.loop.run_until_complete(self.async_hw_serial_number())
+
+ def configuration(self) -> str:
+ return self.loop.run_until_complete(self.async_configuration())
+
+ def text_message(self) -> str:
+ return self.loop.run_until_complete(self.async_text_message())
+
+ def serial_number(self) -> str:
+ return self.loop.run_until_complete(self.async_serial_number())
+
+ def commands(self) -> str:
+ return self.loop.run_until_complete(self.async_commands())
+
+ def device_time(self, v: int) -> None:
+ return self.loop.run_until_complete(self.async_device_time(v))
+
+ def data_buf(self) -> List[Union[DoseRateDB, RareData, RealTimeData, RawData, Event]]:
+ return self.loop.run_until_complete(self.async_data_buf())
+
+ def spectrum(self) -> Spectrum:
+ return self.loop.run_until_complete(self.async_spectrum())
+
+ def spectrum_accum(self) -> Spectrum:
+ return self.loop.run_until_complete(self.async_spectrum_accum())
+
+ def dose_reset(self) -> None:
+ return self.loop.run_until_complete(self.async_dose_reset())
+
+ def spectrum_reset(self) -> None:
+ return self.loop.run_until_complete(self.async_spectrum_reset())
+
+ def energy_calib(self) -> List[float]:
+ return self.loop.run_until_complete(self.async_energy_calib())
+
+ def set_energy_calib(self, coef: List[float]) -> None:
+ return self.loop.run_until_complete(self.async_set_energy_calib(coef))
+
+ def set_language(self, lang='en') -> None:
+ return self.loop.run_until_complete(self.async_set_language(lang))
+
+ def set_device_on(self, on: bool):
+ return self.loop.run_until_complete(self.async_set_device_on(on))
+
+ def set_sound_on(self, on: bool) -> None:
+ return self.loop.run_until_complete(self.async_set_sound_on(on))
+
+ def set_vibro_on(self, on: bool) -> None:
+ return self.loop.run_until_complete(self.async_set_vibro_on(on))
+
+ def set_sound_ctrl(self, ctrls: List[CTRL]) -> None:
+ return self.loop.run_until_complete(self.async_set_sound_ctrl(ctrls))
+
+ def set_display_off_time(self, seconds: int) -> None:
+ return self.loop.run_until_complete(self.async_set_display_off_time(seconds))
+
+ def set_display_brightness(self, brightness: int) -> None:
+ return self.loop.run_until_complete(self.async_set_display_brightness(brightness))
+
+ def set_display_direction(self, direction: DisplayDirection) -> None:
+ return self.loop.run_until_complete(self.async_set_display_direction(direction))
+
+ def set_vibro_ctrl(self, ctrls: List[CTRL]) -> None:
+ return self.loop.run_until_complete(self.async_set_vibro_ctrl(ctrls))
+
+ async def _async_execute(self, reqtype: bytes, args: Optional[bytes] = None) -> BytesBuffer:
assert len(reqtype) == 2
req_seq_no = 0x80 + self._seq
self._seq = (self._seq + 1) % 32
@@ -65,54 +231,56 @@ def execute(self, reqtype: bytes, args: Optional[bytes] = None) -> BytesBuffer:
request = req_header + (args or b'')
full_request = struct.pack('
BytesBuffer:
- r = self.execute(b'\x26\x08', struct.pack(' BytesBuffer:
+ r = await self._async_execute(b'\x26\x08', struct.pack(' None:
- r = self.execute(b'\x25\x08', struct.pack(' None:
+ r = await self._async_execute(b'\x25\x08', struct.pack(' List[int]:
+ async def async_batch_read_vsfrs(self, vsfr_ids: List[VSFR]) -> List[int]:
assert len(vsfr_ids)
- r = self.execute(b'\x2a\x08', b''.join(struct.pack(' str:
- r = self.execute(b'\x05\x00')
+ async def async_status(self) -> str:
+ r = await self._async_execute(b'\x05\x00')
flags = r.unpack(' None:
+ async def async_set_local_time(self, dt: datetime.datetime) -> None:
d = struct.pack(' str:
- r = self.execute(b'\x01\x01')
+ async def async_fw_signature(self) -> str:
+ r = await self._async_execute(b'\x01\x01')
signature = r.unpack(' tuple[tuple[int, int, str], tuple[int, int, str]]:
- r = self.execute(b'\x0a\x00')
+ async def async_fw_version(self) -> tuple[tuple[int, int, str], tuple[int, int, str]]:
+ r = await self._async_execute(b'\x0a\x00')
boot_minor, boot_major = r.unpack(' tuple[tuple[int, int, str], tuple[int, int, str]]:
assert r.size() == 0
return ((boot_major, boot_minor, boot_date), (target_major, target_minor, target_date.strip('\x00')))
- def hw_serial_number(self) -> str:
- r = self.execute(b'\x0b\x00')
+ async def async_hw_serial_number(self) -> str:
+ r = await self._async_execute(b'\x0b\x00')
serial_len = r.unpack(' str:
- r = self.read_request(VS.CONFIGURATION)
+ async def async_configuration(self) -> str:
+ r = await self.async_read_request(VS.CONFIGURATION)
return r.data().decode('cp1251')
- def text_message(self) -> str:
- r = self.read_request(VS.TEXT_MESSAGE)
+ async def async_text_message(self) -> str:
+ r = await self.async_read_request(VS.TEXT_MESSAGE)
return r.data().decode('ascii')
- def serial_number(self) -> str:
- r = self.read_request(8)
+ async def async_serial_number(self) -> str:
+ r = await self.async_read_request(8)
return r.data().decode('ascii')
- def commands(self) -> str:
- br = self.read_request(257)
+ async def async_commands(self) -> str:
+ br = await self.async_read_request(257)
return br.data().decode('ascii')
# called with 0 after init!
- def device_time(self, v: int) -> None:
- self.write_request(VSFR.DEVICE_TIME, struct.pack(' None:
+ await self.async_write_request(VSFR.DEVICE_TIME, struct.pack(' List[Union[DoseRateDB, RareData, RealTimeData, RawData, Event]]:
- r = self.read_request(VS.DATA_BUF)
+ async def async_data_buf(self) -> List[Union[DoseRateDB, RareData, RealTimeData, RawData, Event]]:
+ r = await self.async_read_request(VS.DATA_BUF)
return decode_VS_DATA_BUF(r, self._base_time)
- def spectrum(self) -> Spectrum:
- r = self.read_request(VS.SPECTRUM)
+ async def async_spectrum(self) -> Spectrum:
+ r = await self.async_read_request(VS.SPECTRUM)
return decode_RC_VS_SPECTRUM(r, self._spectrum_format_version)
- def spectrum_accum(self) -> Spectrum:
- r = self.read_request(VS.SPEC_ACCUM)
+ async def async_spectrum_accum(self) -> Spectrum:
+ r = await self.async_read_request(VS.SPEC_ACCUM)
return decode_RC_VS_SPECTRUM(r, self._spectrum_format_version)
- def dose_reset(self) -> None:
- self.write_request(VSFR.DOSE_RESET)
+ async def async_dose_reset(self) -> None:
+ await self.async_write_request(VSFR.DOSE_RESET)
- def spectrum_reset(self) -> None:
- r = self.execute(b'\x27\x08', struct.pack(' None:
+ r = await self._async_execute(b'\x27\x08', struct.pack(' List[float]:
- r = self.read_request(VS.ENERGY_CALIB)
+ async def async_energy_calib(self) -> List[float]:
+ r = await self.async_read_request(VS.ENERGY_CALIB)
return list(r.unpack(' None:
+ async def async_set_energy_calib(self, coef: List[float]) -> None:
assert len(coef) == 3
pc = struct.pack(' None:
+ async def async_set_language(self, lang='en') -> None:
assert lang in {'ru', 'en'}, 'unsupported lang value - use "ru" or "en"'
- self.write_request(VSFR.DEVICE_LANG, struct.pack(' None:
- self.write_request(VSFR.SOUND_ON, struct.pack(' None:
+ await self.async_write_request(VSFR.SOUND_ON, struct.pack(' None:
- self.write_request(VSFR.SOUND_ON, struct.pack(' None:
+ await self.async_write_request(VSFR.SOUND_ON, struct.pack(' None:
+ async def async_set_sound_ctrl(self, ctrls: List[CTRL]) -> None:
flags = 0
+
for c in ctrls:
flags |= int(c)
- self.write_request(VSFR.SOUND_CTRL, struct.pack(' None:
+ async def async_set_display_off_time(self, seconds: int) -> None:
assert seconds in {5, 10, 15, 30}
v = 3 if seconds == 30 else (seconds // 5) - 1
- self.write_request(VSFR.DISP_OFF_TIME, struct.pack(' None:
+ async def async_set_display_brightness(self, brightness: int) -> None:
assert 0 <= brightness and brightness <= 9
- self.write_request(VSFR.DISP_BRT, struct.pack(' None:
+ async def async_set_display_direction(self, direction: DisplayDirection) -> None:
assert isinstance(direction, DisplayDirection)
- self.write_request(VSFR.DISP_DIR, struct.pack(' None:
+ async def async_set_vibro_ctrl(self, ctrls: List[CTRL]) -> None:
flags = 0
for c in ctrls:
assert c != CTRL.CLICKS, 'CTRL.CLICKS not supported for vibro'
flags |= int(c)
- self.write_request(VSFR.VIBRO_CTRL, struct.pack(' None:
+ if self.bluetooth_mac:
+ if len(self.bluetooth_mac) != 17:
+ Logger.warning('The provided MAC address does not seem to be valid, but we will try anyway')
+
+ self.client = BleakClient(self.bluetooth_mac)
+ elif self.bluetooth_serial:
+ if len(self.bluetooth_serial) < 6:
+ Logger.warning('To improve reliability, try to provide at least 6 digits of your Radiacode serial number')
+
+ bt_devices = await self._async_scan()
+
+ if len(bt_devices) == 0:
+ raise DeviceNotFound(
+ 'No Radiacodes found. Check that bluetooth is enabled and that the Radiacode is not connected to another device.'
+ )
+
+ ble, _ = (None, None)
+
+ # Find the requested device in the list
+ for b, a in bt_devices:
+ if self.bluetooth_serial in str(a.local_name):
+ ble = b
+ break
+
+ if ble is None:
+ raise DeviceNotFound(f'No matching serial found while scanning. We were looking for: {self.bluetooth_serial}')
+
+ self.client = BleakClient(ble)
+ elif self.bluetooth_uuid:
+ if len(self.bluetooth_uuid) != 36:
+ Logger.warning('The provided UUID does not seem to be valid, but we will try anyway')
+
+ self.client = BleakClient(self.bluetooth_uuid)
+ else:
+ raise DeviceNotFound('No connection parameters specified.')
+
+ # Attempt connection
+ await self.client.connect()
+
+ if self.client.is_connected:
+ Logger.notify(f'Connected to Radiacode via bluetooth to: {self.client.address}')
+ else:
+ raise DeviceNotFound('Cannot connect to the requested Radiacode')
+
+ # Start notifications
+ self.notiffd = self.client.services.get_characteristic(RADIACODE_NOTIFYFD_UUID)
+ await self.client.start_notify(self.notiffd, self.handleNotification)
+
+ self.writefd = self.client.services.get_characteristic(RADIACODE_WRITEFD_UUID)
+
+ Logger.notify('Notifications started')
- try:
- self.p = Peripheral(mac)
- except BTLEDisconnectError as ex:
- raise DeviceNotFound('Device not found or bluetooth adapter is not powered on') from ex
+ def _scan(self) -> List[Tuple[BLEDevice, AdvertisementData]]:
+ return self.loop.run_until_complete(self._async_scan())
- self.p.withDelegate(self)
+ async def _async_scan(self) -> List[Tuple[BLEDevice, AdvertisementData]]:
+ """Returns a list of Tuples of valid Radiacodes"""
+ radiacodes = []
- service = self.p.getServiceByUUID('e63215e5-7003-49d8-96b0-b024798fb901')
- self.write_fd = service.getCharacteristics('e63215e6-7003-49d8-96b0-b024798fb901')[0].getHandle()
- notify_fd = service.getCharacteristics('e63215e7-7003-49d8-96b0-b024798fb901')[0].getHandle()
- self.p.writeCharacteristic(notify_fd + 1, b'\x01\x00')
+ Logger.info('Scan started...')
+ devices = await BleakScanner.discover(return_adv=True, cb=dict(use_bdaddr=False))
- def handleNotification(self, chandle, data):
- if self._resp_size == 0:
- self._resp_size = 4 + struct.unpack('= 0
- if self._resp_size == 0:
- self._response = self._resp_buffer
- self._resp_buffer = b''
-
- def execute(self, req) -> BytesBuffer:
- for pos in range(0, len(req), 18):
- rp = req[pos : min(pos + 18, len(req))]
- self.p.writeCharacteristic(self.write_fd, rp)
-
- while self._response is None:
- self.p.waitForNotifications(2.0)
-
- br = BytesBuffer(self._response)
- self._response = None
- return br
+ Logger.notify(f'Active service found for device ID: {adv.local_name} - UUID: {ble.address}')
+
+ bt_devices.append(r)
+
+ return bt_devices
+
+ def handleNotification(self, characteristic, data) -> None:
+ if self._resp_size == 0:
+ self._resp_size = 4 + struct.unpack('= 0
+
+ if self._resp_size == 0:
+ self._response = self._resp_buffer
+ self._resp_buffer = b''
+
+ # self._response_event.set()
+ # Logger.notify(f'Notification: {characteristic.description}: {self._resp_buffer}')
+
+ async def execute(self, req) -> BytesBuffer:
+ if self.client is None or self.client.is_connected is False:
+ await self.client.connect()
+
+ if self.client.is_connected is False:
+ raise DeviceNotFound('Connection to the device not active')
+
+ # Request data
+ self._response = []
+ await self.client.write_gatt_char(self.writefd, req, response=False)
+ await asyncio.sleep(1.5)
+
+ # If we knew how much data to expect from every request, we could monitor
+ # for that in handleNotification(), rather than sleeping
+ # await self._response_event.wait()
+ # await self.client.stop_notify(notif)
+
+ br = BytesBuffer(self._response)
+ self._response = None
+ return br
diff --git a/radiacode/transports/usb.py b/radiacode/transports/usb.py
index dfac09e..c05d727 100644
--- a/radiacode/transports/usb.py
+++ b/radiacode/transports/usb.py
@@ -27,26 +27,31 @@ def __init__(self, serial_number=None, timeout_ms=3000):
# usb.core.find(..., serial_number=None) will attempt to match against a value of None,
# rather than ignoring it as a match condition.
self._device = usb.core.find(idVendor=_vid, idProduct=_pid)
+
self._timeout_ms = timeout_ms
+
if self._device is None:
raise DeviceNotFound
+
while True:
try:
self._device.read(0x81, 256, timeout=100)
except usb.core.USBTimeoutError:
break
- def execute(self, request: bytes) -> BytesBuffer:
+ async def execute(self, request: bytes) -> BytesBuffer:
self._device.write(0x1, request)
trials = 0
max_trials = 3
+
while trials < max_trials: # repeat until non-zero lenght data received
data = self._device.read(0x81, 256, timeout=self._timeout_ms).tobytes()
if len(data) != 0:
break
else:
trials += 1
+
if trials >= max_trials:
raise MultipleUSBReadFailure(str(trials) + ' USB Read Failures in sequence')
diff --git a/tests/test_radiacode.py b/tests/test_radiacode.py
new file mode 100644
index 0000000..b698edc
--- /dev/null
+++ b/tests/test_radiacode.py
@@ -0,0 +1,93 @@
+import unittest
+import datetime
+from radiacode.radiacode import RadiaCode
+from radiacode.types import DoseRateDB, Event, RareData, RawData, RealTimeData, Spectrum
+
+class TestRadiaCodeIntegration(unittest.TestCase):
+ """
+ This test class requires a real and available Radiacode
+ """
+ @classmethod
+ def setUpClass(cls):
+ # Connect to the device only once
+ cls.rc = RadiaCode(bluetooth_serial='RadiaCode-10')
+
+ def test_serial_number_integration(self):
+ serial_number = self.rc.serial_number()
+
+ self.assertIsInstance(serial_number, str)
+ self.assertIn('RC-10', serial_number)
+
+ def test_fw_version_integration(self):
+ fw_version = self.rc.fw_version()
+
+ self.assertIsInstance(fw_version, tuple)
+ self.assertEqual(len(fw_version), 2)
+
+ # tuple[tuple[int, int, str], tuple[int, int, str]]:
+ for ver in fw_version:
+ self.assertIsInstance(ver, tuple)
+ self.assertEqual(len(ver), 3)
+ self.assertIsInstance(ver[0], int)
+ self.assertIsInstance(ver[1], int)
+ self.assertIsInstance(ver[2], str)
+
+ def test_spectrum_integration(self):
+ spectrum = self.rc.spectrum()
+
+ self.assertIsInstance(spectrum, Spectrum)
+ self.assertIsInstance(spectrum.duration, datetime.timedelta)
+ self.assertIsInstance(spectrum.a0, float)
+ self.assertIsInstance(spectrum.a1, float)
+ self.assertIsInstance(spectrum.a2, float)
+ self.assertIsInstance(spectrum.counts, list)
+ self.assertTrue(all(isinstance(count, int) for count in spectrum.counts))
+
+ def test_data_buf_integration(self):
+ data_buf = self.rc.data_buf()
+
+ self.assertIsInstance(data_buf, list)
+ for item in data_buf:
+ self.assertTrue(isinstance(item, (DoseRateDB, RareData, RealTimeData, RawData, Event)))
+ self.assertIsInstance(item.dt, datetime.datetime)
+
+ def test_spectrum_accum_integration(self):
+ spectrum_accum = self.rc.spectrum_accum()
+
+ self.assertIsInstance(spectrum_accum, Spectrum)
+ self.assertIsInstance(spectrum_accum.duration, datetime.timedelta)
+ self.assertIsInstance(spectrum_accum.a0, float)
+ self.assertIsInstance(spectrum_accum.a1, float)
+ self.assertIsInstance(spectrum_accum.a2, float)
+ self.assertIsInstance(spectrum_accum.counts, list)
+ self.assertTrue(all(isinstance(count, int) for count in spectrum_accum.counts))
+
+ def test_configuration_integration(self):
+ configuration = self.rc.configuration()
+
+ self.assertIsInstance(configuration, str)
+ self.assertIn('DeviceParams', configuration)
+ self.assertIn('CHN_ChargeLevel', configuration)
+
+ def test_hw_serial_number_integration(self):
+ hw_serial_number = self.rc.hw_serial_number()
+
+ # Format is: XXXXXXXX-XXXXXXXX-XXXXXXXX (hex)
+ self.assertIsInstance(hw_serial_number, str)
+ self.assertIn('-', hw_serial_number)
+
+ def test_status_integration(self):
+ status = self.rc.status()
+
+ self.assertIsInstance(status, str)
+ self.assertIn('flags: ', status)
+
+ def test_commands_integration(self):
+ commands = self.rc.commands()
+
+ self.assertIsInstance(commands, str)
+ self.assertIn('VSFR_DEVICE_CTRL', commands)
+ self.assertIn('VSFR_SYS_MCU_TEMP', commands)
+
+if __name__ == '__main__':
+ unittest.main()