Driver.read() - Code Metrics - Inspection of "read and write message objects" - half2me/libant - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 650a9f...3802bb )

by Benjamin

created 2016-09-07 18:24 UTC

Driver.read() B

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
c	1
b	0
f	0
dl	0
loc	16
rs	8
cc	6

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

from serial import Serial, SerialException, SerialTimeoutException

import usb

from libAnt.constants import MESSAGE_TX_SYNC
from libAnt.message import Message


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):
        self._lock = Lock()

    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) -> Message:
        if not self.isOpen():
            raise DriverException("Device is closed")

        with self._lock:
            while True:
                sync = self._read(1)[0]
                if sync is not MESSAGE_TX_SYNC:
                    continue
                len = self._read(1)[0]
                type = self._read(1)[0]
                data = self._read(len)
                chk = self._read(1)[0]
                msg = Message(type, data)
                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):
        pass

    @abstractmethod
    def _open(self):
        pass

    @abstractmethod
    def _close(self):
        pass

    @abstractmethod
    def _read(self, count: int) -> bytearray:
        pass

    @abstractmethod
    def _write(self, data):
        pass


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

    def __init__(self, device, baudRate=115200):
        super().__init__()
        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):
        return self._serial is None

    def _open(self):
        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):
        self._serial.close()
        self._serial = None

    def _read(self, count):
        return self._serial.read(count)

    def _write(self, data):
        try:
            count = 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):
        Driver.__init__(self)
        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

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

    def _isOpen(self):
        return self._dev is not None

    def _open(self):
        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()
        except IOError as e:
            raise DriverException(str(e))

    def _close(self):
        if self._loop.is_alive():
            self._loop.stop()
            self._loop.join()
        self._loop = None
        usb.util.release_interface(self._dev, self._interfaceNumber)
        usb.util.dispose_resources(self._dev)
        self._dev = self._epOut = self._epIn = None

    def _read(self, count):
        buf = bytearray()
        for i in range(0, count):
            buf.append(self._queue.get())
        return buf

    def _write(self, data):
        return self._epOut.write(data)


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

    def stop(self):
        self._stopper.set()

    def run(self):
        while not self._stopper.is_set():
            try:
                data = self._ep.read(self._packetSize)
                for d in data:
                    self._queue.put(d)
            except usb.USBError as e:
                if e.errno not in (60, 110):
                    raise e


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

half2me / libant

GitHub Access Token became invalid

Push — master ( 650a9f...3802bb )

Driver.read() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like