Remove factory wrapper classes
A module level function is sufficient, no wrapper needed.
This commit is contained in:
parent
ef6b7dec2c
commit
cba3533869
|
@ -60,32 +60,30 @@ class LowLevelInterface(Protocol):
|
|||
...
|
||||
|
||||
|
||||
class DeviceFactory:
|
||||
@staticmethod
|
||||
def create_device(low_level: LowLevelInterface, device_info, setting_callback=None):
|
||||
"""Opens a Logitech Device found attached to the machine, by Linux device path.
|
||||
:returns: An open file handle for the found receiver, or None.
|
||||
"""
|
||||
try:
|
||||
handle = low_level.open_path(device_info.path)
|
||||
if handle:
|
||||
# a direct connected device might not be online (as reported by user)
|
||||
return Device(
|
||||
low_level,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
handle=handle,
|
||||
device_info=device_info,
|
||||
setting_callback=setting_callback,
|
||||
)
|
||||
except OSError as e:
|
||||
logger.exception("open %s", device_info)
|
||||
if e.errno == errno.EACCES:
|
||||
raise
|
||||
except Exception:
|
||||
logger.exception("open %s", device_info)
|
||||
def create_device(low_level: LowLevelInterface, device_info, setting_callback=None):
|
||||
"""Opens a Logitech Device found attached to the machine, by Linux device path.
|
||||
:returns: An open file handle for the found receiver, or None.
|
||||
"""
|
||||
try:
|
||||
handle = low_level.open_path(device_info.path)
|
||||
if handle:
|
||||
# a direct connected device might not be online (as reported by user)
|
||||
return Device(
|
||||
low_level,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
handle=handle,
|
||||
device_info=device_info,
|
||||
setting_callback=setting_callback,
|
||||
)
|
||||
except OSError as e:
|
||||
logger.exception("open %s", device_info)
|
||||
if e.errno == errno.EACCES:
|
||||
raise
|
||||
except Exception:
|
||||
logger.exception("open %s", device_info)
|
||||
raise
|
||||
|
||||
|
||||
class Device:
|
||||
|
|
|
@ -512,35 +512,33 @@ receiver_class_mapping = {
|
|||
}
|
||||
|
||||
|
||||
class ReceiverFactory:
|
||||
@staticmethod
|
||||
def create_receiver(low_level: LowLevelInterface, device_info, setting_callback=None) -> Optional[Receiver]:
|
||||
"""Opens a Logitech Receiver found attached to the machine, by Linux device path."""
|
||||
def create_receiver(low_level: LowLevelInterface, device_info, setting_callback=None) -> Optional[Receiver]:
|
||||
"""Opens a Logitech Receiver found attached to the machine, by Linux device path."""
|
||||
|
||||
try:
|
||||
handle = low_level.open_path(device_info.path)
|
||||
if handle:
|
||||
usb_id = device_info.product_id
|
||||
if isinstance(usb_id, str):
|
||||
usb_id = int(usb_id, 16)
|
||||
try:
|
||||
product_info = low_level.product_information(usb_id)
|
||||
except ValueError:
|
||||
product_info = {}
|
||||
kind = product_info.get("receiver_kind", "unknown")
|
||||
rclass = receiver_class_mapping.get(kind, Receiver)
|
||||
return rclass(
|
||||
low_level,
|
||||
kind,
|
||||
product_info,
|
||||
handle,
|
||||
device_info.path,
|
||||
device_info.product_id,
|
||||
setting_callback,
|
||||
)
|
||||
except OSError as e:
|
||||
logger.exception("open %s", device_info)
|
||||
if e.errno == errno.EACCES:
|
||||
raise
|
||||
except Exception:
|
||||
logger.exception("open %s", device_info)
|
||||
try:
|
||||
handle = low_level.open_path(device_info.path)
|
||||
if handle:
|
||||
usb_id = device_info.product_id
|
||||
if isinstance(usb_id, str):
|
||||
usb_id = int(usb_id, 16)
|
||||
try:
|
||||
product_info = low_level.product_information(usb_id)
|
||||
except ValueError:
|
||||
product_info = {}
|
||||
kind = product_info.get("receiver_kind", "unknown")
|
||||
rclass = receiver_class_mapping.get(kind, Receiver)
|
||||
return rclass(
|
||||
low_level,
|
||||
kind,
|
||||
product_info,
|
||||
handle,
|
||||
device_info.path,
|
||||
device_info.product_id,
|
||||
setting_callback,
|
||||
)
|
||||
except OSError as e:
|
||||
logger.exception("open %s", device_info)
|
||||
if e.errno == errno.EACCES:
|
||||
raise
|
||||
except Exception:
|
||||
logger.exception("open %s", device_info)
|
||||
|
|
|
@ -106,7 +106,7 @@ def _receivers(dev_path=None):
|
|||
if dev_path is not None and dev_path != dev_info.path:
|
||||
continue
|
||||
try:
|
||||
r = receiver.ReceiverFactory.create_receiver(base, dev_info)
|
||||
r = receiver.create_receiver(base, dev_info)
|
||||
if logger.isEnabledFor(logging.DEBUG):
|
||||
logger.debug("[%s] => %s", dev_info.path, r)
|
||||
if r:
|
||||
|
@ -122,9 +122,9 @@ def _receivers_and_devices(dev_path=None):
|
|||
continue
|
||||
try:
|
||||
if dev_info.isDevice:
|
||||
d = device.DeviceFactory.create_device(base, dev_info)
|
||||
d = device.create_device(base, dev_info)
|
||||
else:
|
||||
d = receiver.ReceiverFactory.create_receiver(base, dev_info)
|
||||
d = receiver.create_receiver(base, dev_info)
|
||||
|
||||
if logger.isEnabledFor(logging.DEBUG):
|
||||
logger.debug("[%s] => %s", dev_info.path, d)
|
||||
|
|
|
@ -255,9 +255,9 @@ def _start(device_info):
|
|||
assert _status_callback and _setting_callback
|
||||
isDevice = device_info.isDevice
|
||||
if not isDevice:
|
||||
receiver_ = logitech_receiver.receiver.ReceiverFactory.create_receiver(base, device_info, _setting_callback)
|
||||
receiver_ = logitech_receiver.receiver.create_receiver(base, device_info, _setting_callback)
|
||||
else:
|
||||
receiver_ = logitech_receiver.device.DeviceFactory.create_device(base, device_info, _setting_callback)
|
||||
receiver_ = logitech_receiver.device.create_device(base, device_info, _setting_callback)
|
||||
if receiver_:
|
||||
configuration.attach_to(receiver_)
|
||||
if receiver_.bluetooth and receiver_.hid_serial:
|
||||
|
|
|
@ -82,12 +82,12 @@ def test_create_device(device_info, responses, expected_success):
|
|||
low_level_mock = LowLevelInterfaceFake(responses)
|
||||
if expected_success is None:
|
||||
with pytest.raises(PermissionError):
|
||||
device.DeviceFactory.create_device(low_level_mock, device_info)
|
||||
device.create_device(low_level_mock, device_info)
|
||||
elif not expected_success:
|
||||
with pytest.raises(TypeError):
|
||||
device.DeviceFactory.create_device(low_level_mock, device_info)
|
||||
device.create_device(low_level_mock, device_info)
|
||||
else:
|
||||
test_device = device.DeviceFactory.create_device(low_level_mock, device_info)
|
||||
test_device = device.create_device(low_level_mock, device_info)
|
||||
assert bool(test_device) == expected_success
|
||||
|
||||
|
||||
|
@ -98,7 +98,7 @@ def test_create_device(device_info, responses, expected_success):
|
|||
def test_device_name(device_info, responses, expected_codename, expected_name, expected_kind):
|
||||
low_level = LowLevelInterfaceFake(responses)
|
||||
|
||||
test_device = device.DeviceFactory.create_device(low_level, device_info)
|
||||
test_device = device.create_device(low_level, device_info)
|
||||
|
||||
assert test_device.codename == expected_codename
|
||||
assert test_device.name == expected_name
|
||||
|
|
|
@ -133,12 +133,12 @@ def test_receiver_factory_create_receiver(device_info, responses, handle, serial
|
|||
|
||||
if handle is False:
|
||||
with pytest.raises(Exception): # noqa: B017
|
||||
receiver.ReceiverFactory.create_receiver(mock_low_level, device_info, lambda x: x)
|
||||
receiver.create_receiver(mock_low_level, device_info, lambda x: x)
|
||||
elif handle is None:
|
||||
r = receiver.ReceiverFactory.create_receiver(mock_low_level, device_info, lambda x: x)
|
||||
r = receiver.create_receiver(mock_low_level, device_info, lambda x: x)
|
||||
assert r is None
|
||||
else:
|
||||
r = receiver.ReceiverFactory.create_receiver(mock_low_level, device_info, lambda x: x)
|
||||
r = receiver.create_receiver(mock_low_level, device_info, lambda x: x)
|
||||
assert r.handle == handle
|
||||
assert r.serial == serial
|
||||
assert r.max_devices == max_devices
|
||||
|
@ -155,7 +155,7 @@ def test_receiver_factory_create_receiver(device_info, responses, handle, serial
|
|||
def test_receiver_factory_props(device_info, responses, firmware, codename, remaining_pairings, pairing_info, count):
|
||||
mock_low_level = LowLevelInterfaceFake(responses)
|
||||
|
||||
r = receiver.ReceiverFactory.create_receiver(mock_low_level, device_info, lambda x: x)
|
||||
r = receiver.create_receiver(mock_low_level, device_info, lambda x: x)
|
||||
|
||||
assert len(r.firmware) == firmware if firmware is not None else firmware is None
|
||||
assert r.device_codename(2) == codename
|
||||
|
@ -175,7 +175,7 @@ def test_receiver_factory_props(device_info, responses, firmware, codename, rema
|
|||
def test_receiver_factory_string(device_info, responses, status_str, strng):
|
||||
mock_low_level = LowLevelInterfaceFake(responses)
|
||||
|
||||
r = receiver.ReceiverFactory.create_receiver(mock_low_level, device_info, lambda x: x)
|
||||
r = receiver.create_receiver(mock_low_level, device_info, lambda x: x)
|
||||
|
||||
assert r.status_string() == status_str
|
||||
assert str(r) == strng
|
||||
|
@ -191,7 +191,7 @@ def test_receiver_factory_string(device_info, responses, status_str, strng):
|
|||
def test_receiver_factory_no_device(device_info, responses):
|
||||
mock_low_level = LowLevelInterfaceFake(responses)
|
||||
|
||||
r = receiver.ReceiverFactory.create_receiver(mock_low_level, device_info, lambda x: x)
|
||||
r = receiver.create_receiver(mock_low_level, device_info, lambda x: x)
|
||||
|
||||
with pytest.raises(exceptions.NoSuchDevice):
|
||||
r.device_pairing_information(1)
|
||||
|
|
Loading…
Reference in New Issue