Skip to content

Commit

Permalink
Separate OnlineDeviceManager manager from DeviceManager
Browse files Browse the repository at this point in the history
  • Loading branch information
Tomer Shalev authored and tomers committed Apr 21, 2024
1 parent 561e142 commit ea82ee8
Show file tree
Hide file tree
Showing 3 changed files with 58 additions and 49 deletions.
2 changes: 1 addition & 1 deletion src/labelle/gui/q_device_selector.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@
QToolBar,
)

from labelle.lib.devices.device_manager import OnlineDeviceManager
from labelle.lib.devices.online_device_manager import OnlineDeviceManager
from labelle.lib.devices.usb_device import UsbDevice

LOG = logging.getLogger(__name__)
Expand Down
48 changes: 0 additions & 48 deletions src/labelle/lib/devices/device_manager.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,5 @@
import logging
from typing import Optional

from PyQt6 import QtCore
from PyQt6.QtCore import QTimer
from PyQt6.QtWidgets import QWidget
from usb.core import NoBackendError, USBError

from labelle.lib.constants import (
Expand Down Expand Up @@ -72,47 +68,3 @@ def find_and_select_device(self) -> UsbDevice:
msg = f"Unrecognized device: {hex(dev.id_product)}. {UNCONFIRMED_MESSAGE}"
LOG.debug(msg)
return dev


class OnlineDeviceManager(QWidget):
_last_scan_error: Optional[DeviceManagerError]
_status_time: QTimer
_device_manager: DeviceManager
last_scan_error_changed_signal = QtCore.pyqtSignal(
name="lastScanErrorChangedSignal"
)
devices_changed_signal = QtCore.pyqtSignal(name="devicesChangedSignal")

def __init__(self):
super().__init__()
self._device_manager = DeviceManager()
self._last_scan_error = None
self._init_timers()

def _refresh_devices(self):
prev = self._last_scan_error
try:
changed = self._device_manager.scan()
self._last_scan_error = None
if changed:
self.devices_changed_signal.emit()
except DeviceManagerError as e:
self._last_scan_error = e

if str(prev) != str(self._last_scan_error):
self.devices_changed_signal.emit()
self.last_scan_error_changed_signal.emit()

def _init_timers(self):
self._status_time = QTimer()
self._status_time.timeout.connect(self._refresh_devices)
self._status_time.start(2000)
self._refresh_devices()

@property
def last_scan_error(self) -> Optional[DeviceManagerError]:
return self._last_scan_error

@property
def devices(self) -> list[UsbDevice]:
return self._device_manager.devices
57 changes: 57 additions & 0 deletions src/labelle/lib/devices/online_device_manager.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,57 @@
import logging
from typing import Optional

from PyQt6 import QtCore
from PyQt6.QtCore import QTimer
from PyQt6.QtWidgets import QWidget
from usb.core import NoBackendError, USBError

from labelle.lib.devices.device_manager import DeviceManager, DeviceManagerError
from labelle.lib.devices.usb_device import UsbDevice

LOG = logging.getLogger(__name__)
POSSIBLE_USB_ERRORS = (NoBackendError, USBError)


class OnlineDeviceManager(QWidget):
_last_scan_error: Optional[DeviceManagerError]
_status_time: QTimer
_device_manager: DeviceManager
last_scan_error_changed_signal = QtCore.pyqtSignal(
name="lastScanErrorChangedSignal"
)
devices_changed_signal = QtCore.pyqtSignal(name="devicesChangedSignal")

def __init__(self):
super().__init__()
self._device_manager = DeviceManager()
self._last_scan_error = None
self._init_timers()

def _refresh_devices(self):
prev = self._last_scan_error
try:
changed = self._device_manager.scan()
self._last_scan_error = None
if changed:
self.devices_changed_signal.emit()
except DeviceManagerError as e:
self._last_scan_error = e

if str(prev) != str(self._last_scan_error):
self.devices_changed_signal.emit()
self.last_scan_error_changed_signal.emit()

def _init_timers(self):
self._status_time = QTimer()
self._status_time.timeout.connect(self._refresh_devices)
self._status_time.start(2000)
self._refresh_devices()

@property
def last_scan_error(self) -> Optional[DeviceManagerError]:
return self._last_scan_error

@property
def devices(self) -> list[UsbDevice]:
return self._device_manager.devices

0 comments on commit ea82ee8

Please sign in to comment.