Skip to content

Commit 1a75c8a

Browse files
sde1000sde1000
authored
Merge pull request #153 from sikmir/move-usb-backends
Move USB backends to separate file
2 parents 7e53656 + 9fb61d6 commit 1a75c8a

3 files changed

Lines changed: 116 additions & 116 deletions

File tree

‎dali/driver/base.py‎

Lines changed: 0 additions & 114 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,3 @@
1-
import threading
2-
import usb
3-
4-
51
###############################################################################
62
# driver contracts
73
###############################################################################
@@ -124,113 +120,3 @@ def listen(self):
124120
"""
125121
raise NotImplementedError(
126122
'Abstract ``Listener`` does not implement ``listen``')
127-
128-
129-
###############################################################################
130-
# USB backends
131-
###############################################################################
132-
133-
class USBBackend(Backend):
134-
"""Backend implementation for communicating with USB devices.
135-
"""
136-
137-
def __init__(self, vendor, product, bus=None,
138-
port_numbers=None, interface=0):
139-
self._device = None
140-
# lookup devices by vendor and product
141-
devices = [dev for dev in usb.core.find(
142-
find_all=True,
143-
idVendor=vendor,
144-
idProduct=product
145-
)]
146-
# use first device if bus or port_numers not defined
147-
if bus is None or port_numbers is None:
148-
self._device = devices[0]
149-
else:
150-
for dev in devices:
151-
if dev.bus == bus and dev.port_numbers == port_numbers:
152-
self._device = dev
153-
break
154-
# if queried device not found, raise
155-
if not self._device:
156-
raise usb.core.USBError('Device not found')
157-
# detach kernel driver if necessary
158-
if self._device.is_kernel_driver_active(interface) is True:
159-
self._device.detach_kernel_driver(interface)
160-
# set device configuration
161-
self._device.set_configuration()
162-
# claim interface
163-
usb.util.claim_interface(self._device, interface)
164-
# get active configuration
165-
cfg = self._device.get_active_configuration()
166-
intf = cfg[(0, 0)]
167-
# get write end point
168-
def match_ep_out(e):
169-
return usb.util.endpoint_direction(e.bEndpointAddress) == \
170-
usb.util.ENDPOINT_OUT
171-
self._ep_write = usb.util.find_descriptor(
172-
intf, custom_match=match_ep_out)
173-
# get read end point
174-
def match_ep_in(e):
175-
return usb.util.endpoint_direction(e.bEndpointAddress) == \
176-
usb.util.ENDPOINT_IN
177-
self._ep_read = usb.util.find_descriptor(
178-
intf, custom_match=match_ep_in)
179-
180-
def read(self, timeout=None):
181-
"""Read data from USB device.
182-
"""
183-
return self._ep_read.read(self._ep_read.wMaxPacketSize, timeout=timeout)
184-
185-
def write(self, data):
186-
"""Write data to USB device.
187-
"""
188-
return self._ep_write.write(data)
189-
190-
def close(self):
191-
"""Close connection to USB device.
192-
"""
193-
usb.util.dispose_resources(self._device)
194-
195-
196-
class USBListener(USBBackend, Listener):
197-
"""Listener implementation for communicating with USB devices.
198-
"""
199-
200-
def __init__(self, driver, vendor, product, bus=None,
201-
port_numbers=None, interface=0):
202-
super(USBListener, self).__init__(
203-
vendor,
204-
product,
205-
bus=bus,
206-
port_numbers=port_numbers,
207-
interface=interface
208-
)
209-
self.driver = driver
210-
# flag whether actually disconnecting from device
211-
self._disconnecting = False
212-
# event to stop listening
213-
self._stop_listening = threading.Event()
214-
# create and start listener thread
215-
self._listener = threading.Thread(target=self.listen)
216-
self._listener.start()
217-
218-
def listen(self):
219-
"""Poll data from USB device.
220-
"""
221-
while not self._stop_listening.is_set():
222-
try:
223-
self.driver.receive(self.read())
224-
except usb.core.USBError as e:
225-
# read timeout
226-
if e.errno == 110:
227-
continue
228-
if not self._disconnecting:
229-
self.close()
230-
231-
def close(self):
232-
"""Close connection to USB device.
233-
"""
234-
self._disconnecting = True
235-
self._stop_listening.set()
236-
super(USBListener, self).close()

‎dali/driver/tridonic.py‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2,8 +2,8 @@
22
from dali.driver.base import AsyncDALIDriver
33
from dali.driver.base import DALIDriver
44
from dali.driver.base import SyncDALIDriver
5-
from dali.driver.base import USBBackend
6-
from dali.driver.base import USBListener
5+
from dali.driver.usb import USBBackend
6+
from dali.driver.usb import USBListener
77
from dali.frame import BackwardFrame
88
from dali.frame import ForwardFrame
99
import logging

‎dali/driver/usb.py‎

Lines changed: 114 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,114 @@
1+
import threading
2+
import usb
3+
4+
from dali.driver.base import Backend, Listener
5+
6+
7+
###############################################################################
8+
# USB backends
9+
###############################################################################
10+
11+
class USBBackend(Backend):
12+
"""Backend implementation for communicating with USB devices.
13+
"""
14+
15+
def __init__(self, vendor, product, bus=None,
16+
port_numbers=None, interface=0):
17+
self._device = None
18+
# lookup devices by vendor and product
19+
devices = [dev for dev in usb.core.find(
20+
find_all=True,
21+
idVendor=vendor,
22+
idProduct=product
23+
)]
24+
# use first device if bus or port_numers not defined
25+
if bus is None or port_numbers is None:
26+
self._device = devices[0]
27+
else:
28+
for dev in devices:
29+
if dev.bus == bus and dev.port_numbers == port_numbers:
30+
self._device = dev
31+
break
32+
# if queried device not found, raise
33+
if not self._device:
34+
raise usb.core.USBError('Device not found')
35+
# detach kernel driver if necessary
36+
if self._device.is_kernel_driver_active(interface) is True:
37+
self._device.detach_kernel_driver(interface)
38+
# set device configuration
39+
self._device.set_configuration()
40+
# claim interface
41+
usb.util.claim_interface(self._device, interface)
42+
# get active configuration
43+
cfg = self._device.get_active_configuration()
44+
intf = cfg[(0, 0)]
45+
# get write end point
46+
def match_ep_out(e):
47+
return usb.util.endpoint_direction(e.bEndpointAddress) == \
48+
usb.util.ENDPOINT_OUT
49+
self._ep_write = usb.util.find_descriptor(
50+
intf, custom_match=match_ep_out)
51+
# get read end point
52+
def match_ep_in(e):
53+
return usb.util.endpoint_direction(e.bEndpointAddress) == \
54+
usb.util.ENDPOINT_IN
55+
self._ep_read = usb.util.find_descriptor(
56+
intf, custom_match=match_ep_in)
57+
58+
def read(self, timeout=None):
59+
"""Read data from USB device.
60+
"""
61+
return self._ep_read.read(self._ep_read.wMaxPacketSize, timeout=timeout)
62+
63+
def write(self, data):
64+
"""Write data to USB device.
65+
"""
66+
return self._ep_write.write(data)
67+
68+
def close(self):
69+
"""Close connection to USB device.
70+
"""
71+
usb.util.dispose_resources(self._device)
72+
73+
74+
class USBListener(USBBackend, Listener):
75+
"""Listener implementation for communicating with USB devices.
76+
"""
77+
78+
def __init__(self, driver, vendor, product, bus=None,
79+
port_numbers=None, interface=0):
80+
super(USBListener, self).__init__(
81+
vendor,
82+
product,
83+
bus=bus,
84+
port_numbers=port_numbers,
85+
interface=interface
86+
)
87+
self.driver = driver
88+
# flag whether actually disconnecting from device
89+
self._disconnecting = False
90+
# event to stop listening
91+
self._stop_listening = threading.Event()
92+
# create and start listener thread
93+
self._listener = threading.Thread(target=self.listen)
94+
self._listener.start()
95+
96+
def listen(self):
97+
"""Poll data from USB device.
98+
"""
99+
while not self._stop_listening.is_set():
100+
try:
101+
self.driver.receive(self.read())
102+
except usb.core.USBError as e:
103+
# read timeout
104+
if e.errno == 110:
105+
continue
106+
if not self._disconnecting:
107+
self.close()
108+
109+
def close(self):
110+
"""Close connection to USB device.
111+
"""
112+
self._disconnecting = True
113+
self._stop_listening.set()
114+
super(USBListener, self).close()

0 commit comments

Comments
 (0)