Driver.reOpen() - Code Metrics - Inspection of "Updates" - half2me/libant - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( ca815e...75484e )

by Benjamin

created 2016-08-24 13:52 UTC

Driver.reOpen() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
c	0
b	0
f	0
dl	0
loc	5
rs	9.4285
cc	3

from abc import abstractmethod
from threading import Lock

from serial import Serial, SerialException, SerialTimeoutException

import usb


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()

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

    def isOpen(self):
        with self._lock:
            return self._isOpen

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

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

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

    def read(self, count):
        if count <= 0:
            raise DriverException("Count must be > 0")
        if not self.isOpen():
            raise DriverException("Device is closed")

        with self._lock:
            return self._read(count)

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

        with self._lock:
            self.write(msg)

    @abstractmethod
    def _isOpen(self):
        pass

    @abstractmethod
    def _open(self):
        pass

    @abstractmethod
    def _close(self):
        pass

    @abstractmethod
    def _read(self, count):
        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 _isOpen(self):
        return self._serial is None

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

        if not self._dev.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, idVendor, idProduct):
        Driver.__init__(self)
        self._idVendor = idVendor
        self._idProduct = idProduct
        self._dev = None
        self._epOut = None
        self._epIn = None
        self._interfaceNumber = None

    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
            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")

            # 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._ep_in = 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._ep_in is None:
                raise DriverException("Could not initialize USB endpoint")
        except IOError as e:
            raise DriverException(str(e))

    def _close(self):
        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):
        return self._epIn.read(count)

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


1			from abc import abstractmethod
2			from threading import Lock
3
4			from serial import Serial, SerialException, SerialTimeoutException
5
6			import usb
7
8
9			class DriverException(Exception):
10			pass
11
12
13			class Driver:
14			"""
15			The driver provides an interface to read and write raw data to and from an ANT+ capable hardware device
16			"""
17
18			def __init__(self):
19			self._lock = Lock()
20
21			def __enter__(self):
22			self.open()
23
24			def __exit__(self, exc_type, exc_val, exc_tb):
25			self.close()
26
27			def isOpen(self):
28			with self._lock:
29			return self._isOpen
30
31			def open(self):
32			with self._lock:
33			if not self._isOpen:
34			self._open()
35
36			def close(self):
37			with self._lock:
38			if self._isOpen:
39			self._close()
40
41			def reOpen(self):
42			with self._lock:
43			if self._isOpen:
44			self._close()
45			self._open()
46
47			def read(self, count):
48			if count <= 0:
49			raise DriverException("Count must be > 0")
50			if not self.isOpen():
51			raise DriverException("Device is closed")
52
53			with self._lock:
54			return self._read(count)
55
56			def write(self, msg):
57			if not self.isOpen():
58			raise DriverException("Device is closed")
59
60			with self._lock:
61			self.write(msg)
62
63			@abstractmethod
64			def _isOpen(self):
65			pass
66
67			@abstractmethod
68			def _open(self):
69			pass
70
71			@abstractmethod
72			def _close(self):
73			pass
74
75			@abstractmethod
76			def _read(self, count):
77			pass
78
79			@abstractmethod
80			def _write(self, data):
81			pass
82
83
84			class SerialDriver(Driver):
85			"""
86			An implementation of a serial ANT+ device driver
87			"""
88
89			def __init__(self, device, baudRate=115200):
90			super().__init__()
91			self._device = device
92			self._baudRate = baudRate
93			self._serial = None
94
95			def _isOpen(self):
96			return self._serial is None
97
98			def _open(self):
99			try:
100			self._dev = Serial(self._device, self._baudRate)
101			except SerialException as e:
102			raise DriverException(str(e))
103
104			if not self._dev.isOpen():
105			raise DriverException("Could not open specified device")
106
107			def _close(self):
108			self._serial.close()
109			self._serial = None
110
111			def _read(self, count):
112			return self._serial.read(count)
113
114			def _write(self, data):
115			try:
116			count = self._serial.write(data)
117			self._serial.flush()
118			except SerialTimeoutException as e:
119			raise DriverException(str(e))
120
121
122			class USBDriver(Driver):
123			"""
124			An implementation of a USB ANT+ device driver
125			"""
126
127			def __init__(self, idVendor, idProduct):
128			Driver.__init__(self)
129			self._idVendor = idVendor
130			self._idProduct = idProduct
131			self._dev = None
132			self._epOut = None
133			self._epIn = None
134			self._interfaceNumber = None
135
136			def _isOpen(self):
137			return self._dev is not None
138
139			def _open(self):
140			try:
141			# find the first USB device that matches the filter
142			self._dev = usb.core.find(idVendor=self._idVendor, idProduct=self._idProduct)
143
144			if self._dev is None:
145			raise DriverException("Could not open specified device")
146
147			# Detach kernel driver
148			if self._dev.is_kernel_driver_active(0):
149			try:
150			self._dev.detach_kernel_driver(0)
151			except usb.USBError as e:
152			raise DriverException("Could not detach kernel driver")
153
154			# set the active configuration. With no arguments, the first
155			# configuration will be the active one
156			self._dev.set_configuration()
157
158			# get an endpoint instance
159			cfg = self._dev.get_active_configuration()
160			self._interfaceNumber = cfg[(0, 0)].bInterfaceNumber
161			interface = usb.util.find_descriptor(cfg, bInterfaceNumber=self._interfaceNumber,
162			bAlternateSetting=usb.control.get_interface(self._dev,
163			self._interfaceNumber))
164			usb.util.claim_interface(self._dev, self._interfaceNumber)
165
166			self._epOut = usb.util.find_descriptor(interface, custom_match=lambda e: usb.util.endpoint_direction(
167			e.bEndpointAddress) == usb.ENDPOINT_OUT)
168
169			self._ep_in = usb.util.find_descriptor(interface, custom_match=lambda e: usb.util.endpoint_direction(
170			e.bEndpointAddress) == usb.ENDPOINT_IN)
171
172			if self._epOut is None or self._ep_in is None:
173			raise DriverException("Could not initialize USB endpoint")
174			except IOError as e:
175			raise DriverException(str(e))
176
177			def _close(self):
178			usb.util.release_interface(self._dev, self._interfaceNumber)
179			usb.util.dispose_resources(self._dev)
180			self._dev = self._epOut = self._epIn = None
181
182			def _read(self, count):
183			return self._epIn.read(count)
184
185			def _write(self, data):
186			return self._epOut.write(data)
187

half2me / libant

GitHub Access Token became invalid

Push — master ( ca815e...75484e )

Driver.reOpen() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like