DummyDriver._write() - Code Metrics - Inspection of "Add debug option to driver" - half2me/libant - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — add-logging ( b5daf2...d977ae )

unknown

created 2016-09-28 10:05 UTC

DummyDriver._write() A

↳ Parent: DummyDriver

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	1
c	1
b	0
f	0
dl	0
loc	2
rs	10

from abc import abstractmethod
from queue import Queue, Empty
from threading import Lock, Thread, Event

from serial import Serial, SerialException, SerialTimeoutException

import usb
import time

from libAnt.constants import MESSAGE_TX_SYNC, MESSAGE_CHANNEL_BROADCAST_DATA
from libAnt.message import Message, SystemResetMessage


class DriverException(Exception):
    pass


class Driver:
    """
    The driver provides an interface to read and write raw data to and from an ANT+ capable hardware device
    """

    def __init__(self, debug=False):
        self._lock = Lock()
        self._debug = debug

    def __enter__(self):
        self.open()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.close()

    def isOpen(self) -> bool:
        with self._lock:
            return self._isOpen()

    def open(self) -> None:
        with self._lock:
            if not self._isOpen():
                self._open()

    def close(self) -> None:
        with self._lock:
            if self._isOpen:
                self._close()

    def reOpen(self) -> None:
        with self._lock:
            if self._isOpen():
                self._close()
            self._open()

    def read(self, timeout=None) -> Message:
        if not self.isOpen():
            raise DriverException("Device is closed")

        with self._lock:
            while True:
                sync = self._read(1, timeout=timeout)[0]
                if sync is not MESSAGE_TX_SYNC:
                    continue
                len = self._read(1, timeout=timeout)[0]
                type = self._read(1, timeout=timeout)[0]
                data = self._read(len, timeout=timeout)
                chk = self._read(1, timeout=timeout)[0]
                msg = Message(type, data)
                if self._debug:
                    logMsg = bytes([sync, len, type, data, chk])
                    timestamp = time.time()
                    # TODO: save to file
                if msg.checksum() == chk:
                    return msg

    def write(self, msg: Message) -> None:
        if not self.isOpen():
            raise DriverException("Device is closed")

        with self._lock:
            self._write(msg.encode())

    @abstractmethod
    def _isOpen(self) -> bool:
        pass

    @abstractmethod
    def _open(self) -> None:
        pass

    @abstractmethod
    def _close(self) -> None:
        pass

    @abstractmethod
    def _read(self, count: int, timeout=None) -> bytes:
        pass

    @abstractmethod
    def _write(self, data: bytes) -> None:
        pass


class SerialDriver(Driver):
    """
    An implementation of a serial ANT+ device driver
    """

    def __init__(self, device: str, baudRate: int = 115200, debug=False):
        super().__init__(debug=debug)
        self._device = device
        self._baudRate = baudRate
        self._serial = None

    def __str__(self):
        if self.isOpen():
            return self._device + " @ " + str(self._baudRate)
        return None

    def _isOpen(self) -> bool:
        return self._serial is None

    def _open(self) -> None:
        try:
            self._serial = Serial(self._device, self._baudRate)
        except SerialException as e:
            raise DriverException(str(e))

        if not self._serial.isOpen():
            raise DriverException("Could not open specified device")

    def _close(self) -> None:
        self._serial.close()
        self._serial = None

    def _read(self, count: int, timeout=None) -> bytes:
        return self._serial.read(count, timeout=timeout)

    def _write(self, data: bytes) -> None:
        try:
            self._serial.write(data)
            self._serial.flush()
        except SerialTimeoutException as e:
            raise DriverException(str(e))


class USBDriver(Driver):
    """
    An implementation of a USB ANT+ device driver
    """

    def __init__(self, vid, pid, debug=False):
        super().__init__(debug=debug)
        self._idVendor = vid
        self._idProduct = pid
        self._dev = None
        self._epOut = None
        self._epIn = None
        self._interfaceNumber = None
        self._packetSize = 0x20
        self._queue = None
        self._loop = None
        self._driver_open = False

    def __str__(self):
        if self.isOpen():
            return str(self._dev)
        return "Closed"

    def _isOpen(self) -> bool:
        return self._driver_open

    def _open(self) -> None:
        print('USB OPEN START')
        try:
            # find the first USB device that matches the filter
            self._dev = usb.core.find(idVendor=self._idVendor, idProduct=self._idProduct)

            if self._dev is None:
                raise DriverException("Could not open specified device")

            # Detach kernel driver
            try:
                if self._dev.is_kernel_driver_active(0):
                    try:
                        self._dev.detach_kernel_driver(0)
                    except usb.USBError as e:
                        raise DriverException("Could not detach kernel driver")
            except NotImplementedError:
                pass  # for non unix systems

            # set the active configuration. With no arguments, the first
            # configuration will be the active one
            self._dev.set_configuration()

            # get an endpoint instance
            cfg = self._dev.get_active_configuration()
            self._interfaceNumber = cfg[(0, 0)].bInterfaceNumber
            interface = usb.util.find_descriptor(cfg, bInterfaceNumber=self._interfaceNumber,
                                                 bAlternateSetting=usb.control.get_interface(self._dev,
                                                                                             self._interfaceNumber))
            usb.util.claim_interface(self._dev, self._interfaceNumber)

            self._epOut = usb.util.find_descriptor(interface, custom_match=lambda e: usb.util.endpoint_direction(
                e.bEndpointAddress) == usb.ENDPOINT_OUT)

            self._epIn = usb.util.find_descriptor(interface, custom_match=lambda e: usb.util.endpoint_direction(
                e.bEndpointAddress) == usb.ENDPOINT_IN)

            if self._epOut is None or self._epIn is None:
                raise DriverException("Could not initialize USB endpoint")

            self._queue = Queue()
            self._loop = USBLoop(self._epIn, self._packetSize, self._queue)
            self._loop.start()
            self._driver_open = True
            print('USB OPEN SUCCESS')
        except IOError as e:
            self._close()
            raise DriverException(str(e))

    def _close(self) -> None:
        print('USB CLOSE START')
        if self._loop is not None:
            if self._loop.is_alive():
                self._loop.stop()
                self._loop.join()
        self._loop = None
        try:
            self._dev.reset()
            usb.util.dispose_resources(self._dev)
        except:
            pass
        self._dev = self._epOut = self._epIn = None
        self._driver_open = False
        print('USB CLOSE END')

    def _read(self, count: int, timeout=None) -> bytes:
        data = bytearray()
        for i in range(0, count):
            b = self._queue.get(timeout=timeout)
            if b is None:
                self._close()
                raise DriverException("Device is closed!")
            data.append(b)
        return bytes(data)

    def _write(self, data: bytes) -> None:
        return self._epOut.write(data)


class USBLoop(Thread):
    def __init__(self, ep, packetSize: int, queue: Queue):
        super().__init__()
        self._stopper = Event()
        self._ep = ep
        self._packetSize = packetSize
        self._queue = queue

    def stop(self) -> None:
        self._stopper.set()

    def run(self) -> None:
        while not self._stopper.is_set():
            try:
                data = self._ep.read(self._packetSize, timeout=1000)
                for d in data:
                    self._queue.put(d)
            except usb.core.USBError as e:
                if e.errno not in (60, 110) and e.backend_error_code != -116: # Timout errors
                    self._stopper.set()
        #  We Put in an invalid byte so threads will realize the device is stopped
        self._queue.put(None)

class DummyDriver(Driver):
    def __init__(self):
        self._isopen = False
        self._data = Queue()
        msg1 = Message(MESSAGE_CHANNEL_BROADCAST_DATA, b'\x00\x00\x00\x00\x00\x00\x00\x00').encode()
        for b in msg1:
            self._data.put(b)
        msg2 = Message(MESSAGE_CHANNEL_BROADCAST_DATA, b'\x00\x00\x00\x00\x00\x00\x00\x00').encode()
        for b in msg2:
            self._data.put(b)
        super().__init__(debug=True)

    def _isOpen(self) -> bool:
        return self._isopen

    def _close(self) -> None:
        self._isopen = False

    def _read(self, count: int, timeout=None) -> bytes:
        return self._data.get(timeout=timeout)

    def _open(self) -> None:
        self._isopen = True

    def _write(self, data: bytes) -> None:
        pass


1			from abc import abstractmethod
2			from queue import Queue, Empty
3			from threading import Lock, Thread, Event
4
5			from serial import Serial, SerialException, SerialTimeoutException
6
7			import usb
8			import time
9
10			from libAnt.constants import MESSAGE_TX_SYNC, MESSAGE_CHANNEL_BROADCAST_DATA
11			from libAnt.message import Message, SystemResetMessage
12
13
14			class DriverException(Exception):
15			pass
16
17
18			class Driver:
19			"""
20			The driver provides an interface to read and write raw data to and from an ANT+ capable hardware device
21			"""
22
23			def __init__(self, debug=False):
24			self._lock = Lock()
25			self._debug = debug
26
27			def __enter__(self):
28			self.open()
29			return self
30
31			def __exit__(self, exc_type, exc_val, exc_tb):
32			self.close()
33
34			def isOpen(self) -> bool:
35			with self._lock:
36			return self._isOpen()
37
38			def open(self) -> None:
39			with self._lock:
40			if not self._isOpen():
41			self._open()
42
43			def close(self) -> None:
44			with self._lock:
45			if self._isOpen:
46			self._close()
47
48			def reOpen(self) -> None:
49			with self._lock:
50			if self._isOpen():
51			self._close()
52			self._open()
53
54			def read(self, timeout=None) -> Message:
55			if not self.isOpen():
56			raise DriverException("Device is closed")
57
58			with self._lock:
59			while True:
60			sync = self._read(1, timeout=timeout)[0]
61			if sync is not MESSAGE_TX_SYNC:
62			continue
63			len = self._read(1, timeout=timeout)[0]
64			type = self._read(1, timeout=timeout)[0]
65			data = self._read(len, timeout=timeout)
66			chk = self._read(1, timeout=timeout)[0]
67			msg = Message(type, data)
68			if self._debug:
69			logMsg = bytes([sync, len, type, data, chk])
70			timestamp = time.time()
71			# TODO: save to file
72			if msg.checksum() == chk:
73			return msg
74
75			def write(self, msg: Message) -> None:
76			if not self.isOpen():
77			raise DriverException("Device is closed")
78
79			with self._lock:
80			self._write(msg.encode())
81
82			@abstractmethod
83			def _isOpen(self) -> bool:
84			pass
85
86			@abstractmethod
87			def _open(self) -> None:
88			pass
89
90			@abstractmethod
91			def _close(self) -> None:
92			pass
93
94			@abstractmethod
95			def _read(self, count: int, timeout=None) -> bytes:
96			pass
97
98			@abstractmethod
99			def _write(self, data: bytes) -> None:
100			pass
101
102
103			class SerialDriver(Driver):
104			"""
105			An implementation of a serial ANT+ device driver
106			"""
107
108			def __init__(self, device: str, baudRate: int = 115200, debug=False):
109			super().__init__(debug=debug)
110			self._device = device
111			self._baudRate = baudRate
112			self._serial = None
113
114			def __str__(self):
115			if self.isOpen():
116			return self._device + " @ " + str(self._baudRate)
117			return None
118
119			def _isOpen(self) -> bool:
120			return self._serial is None
121
122			def _open(self) -> None:
123			try:
124			self._serial = Serial(self._device, self._baudRate)
125			except SerialException as e:
126			raise DriverException(str(e))
127
128			if not self._serial.isOpen():
129			raise DriverException("Could not open specified device")
130
131			def _close(self) -> None:
132			self._serial.close()
133			self._serial = None
134
135			def _read(self, count: int, timeout=None) -> bytes:
136			return self._serial.read(count, timeout=timeout)
137
138			def _write(self, data: bytes) -> None:
139			try:
140			self._serial.write(data)
141			self._serial.flush()
142			except SerialTimeoutException as e:
143			raise DriverException(str(e))
144
145
146			class USBDriver(Driver):
147			"""
148			An implementation of a USB ANT+ device driver
149			"""
150
151			def __init__(self, vid, pid, debug=False):
152			super().__init__(debug=debug)
153			self._idVendor = vid
154			self._idProduct = pid
155			self._dev = None
156			self._epOut = None
157			self._epIn = None
158			self._interfaceNumber = None
159			self._packetSize = 0x20
160			self._queue = None
161			self._loop = None
162			self._driver_open = False
163
164			def __str__(self):
165			if self.isOpen():
166			return str(self._dev)
167			return "Closed"
168
169			def _isOpen(self) -> bool:
170			return self._driver_open
171
172			def _open(self) -> None:
173			print('USB OPEN START')
174			try:
175			# find the first USB device that matches the filter
176			self._dev = usb.core.find(idVendor=self._idVendor, idProduct=self._idProduct)
177
178			if self._dev is None:
179			raise DriverException("Could not open specified device")
180
181			# Detach kernel driver
182			try:
183			if self._dev.is_kernel_driver_active(0):
184			try:
185			self._dev.detach_kernel_driver(0)
186			except usb.USBError as e:
187			raise DriverException("Could not detach kernel driver")
188			except NotImplementedError:
189			pass # for non unix systems
190
191			# set the active configuration. With no arguments, the first
192			# configuration will be the active one
193			self._dev.set_configuration()
194
195			# get an endpoint instance
196			cfg = self._dev.get_active_configuration()
197			self._interfaceNumber = cfg[(0, 0)].bInterfaceNumber
198			interface = usb.util.find_descriptor(cfg, bInterfaceNumber=self._interfaceNumber,
199			bAlternateSetting=usb.control.get_interface(self._dev,
200			self._interfaceNumber))
201			usb.util.claim_interface(self._dev, self._interfaceNumber)
202
203			self._epOut = usb.util.find_descriptor(interface, custom_match=lambda e: usb.util.endpoint_direction(
204			e.bEndpointAddress) == usb.ENDPOINT_OUT)
205
206			self._epIn = usb.util.find_descriptor(interface, custom_match=lambda e: usb.util.endpoint_direction(
207			e.bEndpointAddress) == usb.ENDPOINT_IN)
208
209			if self._epOut is None or self._epIn is None:
210			raise DriverException("Could not initialize USB endpoint")
211
212			self._queue = Queue()
213			self._loop = USBLoop(self._epIn, self._packetSize, self._queue)
214			self._loop.start()
215			self._driver_open = True
216			print('USB OPEN SUCCESS')
217			except IOError as e:
218			self._close()
219			raise DriverException(str(e))
220
221			def _close(self) -> None:
222			print('USB CLOSE START')
223			if self._loop is not None:
224			if self._loop.is_alive():
225			self._loop.stop()
226			self._loop.join()
227			self._loop = None
228			try:
229			self._dev.reset()
230			usb.util.dispose_resources(self._dev)
231			except:
232			pass
233			self._dev = self._epOut = self._epIn = None
234			self._driver_open = False
235			print('USB CLOSE END')
236
237			def _read(self, count: int, timeout=None) -> bytes:
238			data = bytearray()
239			for i in range(0, count):
240			b = self._queue.get(timeout=timeout)
241			if b is None:
242			self._close()
243			raise DriverException("Device is closed!")
244			data.append(b)
245			return bytes(data)
246
247			def _write(self, data: bytes) -> None:
248			return self._epOut.write(data)
249
250
251			class USBLoop(Thread):
252			def __init__(self, ep, packetSize: int, queue: Queue):
253			super().__init__()
254			self._stopper = Event()
255			self._ep = ep
256			self._packetSize = packetSize
257			self._queue = queue
258
259			def stop(self) -> None:
260			self._stopper.set()
261
262			def run(self) -> None:
263			while not self._stopper.is_set():
264			try:
265			data = self._ep.read(self._packetSize, timeout=1000)
266			for d in data:
267			self._queue.put(d)
268			except usb.core.USBError as e:
269			if e.errno not in (60, 110) and e.backend_error_code != -116: # Timout errors
270			self._stopper.set()
271			# We Put in an invalid byte so threads will realize the device is stopped
272			self._queue.put(None)
273
274			class DummyDriver(Driver):
275			def __init__(self):
276			self._isopen = False
277			self._data = Queue()
278			msg1 = Message(MESSAGE_CHANNEL_BROADCAST_DATA, b'\x00\x00\x00\x00\x00\x00\x00\x00').encode()
279			for b in msg1:
280			self._data.put(b)
281			msg2 = Message(MESSAGE_CHANNEL_BROADCAST_DATA, b'\x00\x00\x00\x00\x00\x00\x00\x00').encode()
282			for b in msg2:
283			self._data.put(b)
284			super().__init__(debug=True)
285
286			def _isOpen(self) -> bool:
287			return self._isopen
288
289			def _close(self) -> None:
290			self._isopen = False
291
292			def _read(self, count: int, timeout=None) -> bytes:
293			return self._data.get(timeout=timeout)
294
295			def _open(self) -> None:
296			self._isopen = True
297
298			def _write(self, data: bytes) -> None:
299			pass
300

half2me / libant

GitHub Access Token became invalid

Push — add-logging ( b5daf2...d977ae )

DummyDriver._write() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like