DummyDriver.__init__() - Code Metrics - Inspection of "Add logging" - half2me/libant - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#3)

unknown

created 2016-10-04 11:35 UTC

DummyDriver.init() A

↳ Parent: DummyDriver

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	4
Bugs	0	Features	0

Metric	Value
cc	4
c	4
b	0
f	0
dl	0
loc	13
rs	9.2

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

import math
from serial import Serial, SerialException, SerialTimeoutException

import usb
import time
from struct import *
import os

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, log = None):
        self._lock = Lock()
        self._log = log
        self._openTime = None

    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()
                self._openTime = time.time()
                if self._log:
                    # write pcap global header
                    magic_number = b'\xD4\xC3\xB2\xA1'
                    version_major = 2
                    version_minor = 4
                    thiszone = b'\x00\x00\x00\x00'
                    sigfigs = b'\x00\x00\x00\x00'
                    snaplen = b'\xFF\x00\x00\x00'
                    network = b'\x01\x00\x00\x00'

                    pcap_global_header = Struct('<4shh4s4s4s4s')

                    with open(self._log, 'wb') as log:
                        log.write(pcap_global_header.pack(magic_number, version_major, version_minor, thiszone, sigfigs, snaplen, network))

    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:
        # Splits the string into a list of tokens every n characters
        def splitN(str1, n):
            return [str1[start:start + n] for start in range(0, len(str1), n)]

        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
                length = self._read(1, timeout=timeout)[0]
                type = self._read(1, timeout=timeout)[0]
                data = self._read(length, timeout=timeout)
                chk = self._read(1, timeout=timeout)[0]
                msg = Message(type, data)
                if self._log:
                    logMsg = bytearray([sync, length, type])
                    logMsg.extend(data)
                    logMsg.append(chk)
                    timestamp = time.time() - self._openTime
                    frac, whole = math.modf(timestamp)

                    ts_sec = int(whole).to_bytes(4, byteorder='little')
                    ts_usec = int(frac * 1000 * 1000).to_bytes(4, byteorder='little')
                    incl_len = len(logMsg)
                    orig_len = incl_len

                    pcap_packet_header = Struct('<4s4sll')

                    with open(self._log, 'ab') as log:
                        log.write(pcap_packet_header.pack(ts_sec, ts_usec, incl_len, orig_len))
                        log.write(logMsg)

                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, log = None):
        super().__init__(log = log)
        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, log = None):
        super().__init__(log = None)
        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, log = None):
        super().__init__(log = log)
        self._isopen = False
        self._data = Queue()
        msg1 = Message(MESSAGE_CHANNEL_BROADCAST_DATA, b'\x00\x01\x02\x03\x04\x05\x06\x07').encode()
        for b in msg1:
            self._data.put(b)
        msg2 = Message(MESSAGE_CHANNEL_BROADCAST_DATA, b'\x00\xF1\xF2\xF3\xF4\xF5\xF6\xF7').encode()
        for b in msg2:
            self._data.put(b)
        msg3 = Message(MESSAGE_CHANNEL_BROADCAST_DATA, b'\x00\xF9\xFA\xFB\xFC\xFD\xFE\xFF').encode()
        for b in msg3:
            self._data.put(b)

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

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

    def _read(self, count: int, timeout=None) -> bytes:
        data = bytearray()
        for i in range(0, count):
            data.append(self._data.get(timeout=timeout))
        return bytes(data)

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

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

class PcapDriver(Driver):
    def __init__(self, pcap, log = None):
        super().__init__(log = log)
        self._isopen = False
        self._pcap = pcap
        self._buffer = Queue()

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

    def _open(self) -> None:
        self._isopen = True
        self._pcapfile = open(self._pcap, 'rb')
        self._EOF = os.stat(self._pcap).st_size
        # move file pointer to first packet header
        global_header_length = 24
        self._pcapfile.seek(global_header_length, 0)


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

    def _read(self, count: int, timeout=None) -> bytes:
        def read_next_packet_into_buffer():
            if self._pcapfile.tell() is self._EOF:
                print("EOF")
                return

            ts_sec, = unpack('i', self._pcapfile.read(4))
            ts_usec = unpack('i', self._pcapfile.read(4))[0]/1000000
            ts = ts_sec + ts_usec
            print("ts: ", ts)
            time.sleep(ts)

            packet_length = unpack('i', self._pcapfile.read(4))[0]
            print("packet length: ", packet_length)
            self._pcapfile.seek(4, 1)
            for i in range(0, packet_length):
                self._buffer.put(self._pcapfile.read(1))

        result = bytearray()

        print("Reading ", count, " byte(s)...")
        while len(result) < count:
            if self._buffer.empty():
                print("reading next packet into buffer...")
                read_next_packet_into_buffer()
            try:
                result.extend(self._buffer.get(block=True, timeout = timeout))
            except Empty:
                print("Timed out..")

        print("result: ", bytes(result))
        return bytes(result)

    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			import math
6			from serial import Serial, SerialException, SerialTimeoutException
7
8			import usb
9			import time
10			from struct import *
11			import os
12
13			from libAnt.constants import MESSAGE_TX_SYNC, MESSAGE_CHANNEL_BROADCAST_DATA
14			from libAnt.message import Message, SystemResetMessage
15
16
17			class DriverException(Exception):
18			pass
19
20
21			class Driver:
22			"""
23			The driver provides an interface to read and write raw data to and from an ANT+ capable hardware device
24			"""
25
26			def __init__(self, log = None):
27			self._lock = Lock()
28			self._log = log
29			self._openTime = None
30
31			def __enter__(self):
32			self.open()
33			return self
34
35			def __exit__(self, exc_type, exc_val, exc_tb):
36			self.close()
37
38			def isOpen(self) -> bool:
39			with self._lock:
40			return self._isOpen()
41
42			def open(self) -> None:
43			with self._lock:
44			if not self._isOpen():
45			self._open()
46			self._openTime = time.time()
47			if self._log:
48			# write pcap global header
49			magic_number = b'\xD4\xC3\xB2\xA1'
50			version_major = 2
51			version_minor = 4
52			thiszone = b'\x00\x00\x00\x00'
53			sigfigs = b'\x00\x00\x00\x00'
54			snaplen = b'\xFF\x00\x00\x00'
55			network = b'\x01\x00\x00\x00'
56
57			pcap_global_header = Struct('<4shh4s4s4s4s')
58
59			with open(self._log, 'wb') as log:
60			log.write(pcap_global_header.pack(magic_number, version_major, version_minor, thiszone, sigfigs, snaplen, network))
61
62			def close(self) -> None:
63			with self._lock:
64			if self._isOpen:
65			self._close()
66
67			def reOpen(self) -> None:
68			with self._lock:
69			if self._isOpen():
70			self._close()
71			self._open()
72
73			def read(self, timeout=None) -> Message:
74			# Splits the string into a list of tokens every n characters
75			def splitN(str1, n):
76			return [str1[start:start + n] for start in range(0, len(str1), n)]
77
78			if not self.isOpen():
79			raise DriverException("Device is closed")
80
81			with self._lock:
82			while True:
83			sync = self._read(1, timeout=timeout)[0]
84			if sync is not MESSAGE_TX_SYNC:
85			continue
86			length = self._read(1, timeout=timeout)[0]
87			type = self._read(1, timeout=timeout)[0]
88			data = self._read(length, timeout=timeout)
89			chk = self._read(1, timeout=timeout)[0]
90			msg = Message(type, data)
91			if self._log:
92			logMsg = bytearray([sync, length, type])
93			logMsg.extend(data)
94			logMsg.append(chk)
95			timestamp = time.time() - self._openTime
96			frac, whole = math.modf(timestamp)
97
98			ts_sec = int(whole).to_bytes(4, byteorder='little')
99			ts_usec = int(frac * 1000 * 1000).to_bytes(4, byteorder='little')
100			incl_len = len(logMsg)
101			orig_len = incl_len
102
103			pcap_packet_header = Struct('<4s4sll')
104
105			with open(self._log, 'ab') as log:
106			log.write(pcap_packet_header.pack(ts_sec, ts_usec, incl_len, orig_len))
107			log.write(logMsg)
108
109			if msg.checksum() == chk:
110			return msg
111
112			def write(self, msg: Message) -> None:
113			if not self.isOpen():
114			raise DriverException("Device is closed")
115
116			with self._lock:
117			self._write(msg.encode())
118
119			@abstractmethod
120			def _isOpen(self) -> bool:
121			pass
122
123			@abstractmethod
124			def _open(self) -> None:
125			pass
126
127			@abstractmethod
128			def _close(self) -> None:
129			pass
130
131			@abstractmethod
132			def _read(self, count: int, timeout=None) -> bytes:
133			pass
134
135			@abstractmethod
136			def _write(self, data: bytes) -> None:
137			pass
138
139
140			class SerialDriver(Driver):
141			"""
142			An implementation of a serial ANT+ device driver
143			"""
144
145			def __init__(self, device: str, baudRate: int = 115200, log = None):
146			super().__init__(log = log)
147			self._device = device
148			self._baudRate = baudRate
149			self._serial = None
150
151			def __str__(self):
152			if self.isOpen():
153			return self._device + " @ " + str(self._baudRate)
154			return None
155
156			def _isOpen(self) -> bool:
157			return self._serial is None
158
159			def _open(self) -> None:
160			try:
161			self._serial = Serial(self._device, self._baudRate)
162			except SerialException as e:
163			raise DriverException(str(e))
164
165			if not self._serial.isOpen():
166			raise DriverException("Could not open specified device")
167
168			def _close(self) -> None:
169			self._serial.close()
170			self._serial = None
171
172			def _read(self, count: int, timeout=None) -> bytes:
173			return self._serial.read(count, timeout=timeout)
174
175			def _write(self, data: bytes) -> None:
176			try:
177			self._serial.write(data)
178			self._serial.flush()
179			except SerialTimeoutException as e:
180			raise DriverException(str(e))
181
182
183			class USBDriver(Driver):
184			"""
185			An implementation of a USB ANT+ device driver
186			"""
187
188			def __init__(self, vid, pid, log = None):
189			super().__init__(log = None)
190			self._idVendor = vid
191			self._idProduct = pid
192			self._dev = None
193			self._epOut = None
194			self._epIn = None
195			self._interfaceNumber = None
196			self._packetSize = 0x20
197			self._queue = None
198			self._loop = None
199			self._driver_open = False
200
201			def __str__(self):
202			if self.isOpen():
203			return str(self._dev)
204			return "Closed"
205
206			def _isOpen(self) -> bool:
207			return self._driver_open
208
209			def _open(self) -> None:
210			print('USB OPEN START')
211			try:
212			# find the first USB device that matches the filter
213			self._dev = usb.core.find(idVendor=self._idVendor, idProduct=self._idProduct)
214
215			if self._dev is None:
216			raise DriverException("Could not open specified device")
217
218			# Detach kernel driver
219			try:
220			if self._dev.is_kernel_driver_active(0):
221			try:
222			self._dev.detach_kernel_driver(0)
223			except usb.USBError as e:
224			raise DriverException("Could not detach kernel driver")
225			except NotImplementedError:
226			pass # for non unix systems
227
228			# set the active configuration. With no arguments, the first
229			# configuration will be the active one
230			self._dev.set_configuration()
231
232			# get an endpoint instance
233			cfg = self._dev.get_active_configuration()
234			self._interfaceNumber = cfg[(0, 0)].bInterfaceNumber
235			interface = usb.util.find_descriptor(cfg, bInterfaceNumber=self._interfaceNumber,
236			bAlternateSetting=usb.control.get_interface(self._dev,
237			self._interfaceNumber))
238			usb.util.claim_interface(self._dev, self._interfaceNumber)
239
240			self._epOut = usb.util.find_descriptor(interface, custom_match=lambda e: usb.util.endpoint_direction(
241			e.bEndpointAddress) == usb.ENDPOINT_OUT)
242
243			self._epIn = usb.util.find_descriptor(interface, custom_match=lambda e: usb.util.endpoint_direction(
244			e.bEndpointAddress) == usb.ENDPOINT_IN)
245
246			if self._epOut is None or self._epIn is None:
247			raise DriverException("Could not initialize USB endpoint")
248
249			self._queue = Queue()
250			self._loop = USBLoop(self._epIn, self._packetSize, self._queue)
251			self._loop.start()
252			self._driver_open = True
253			print('USB OPEN SUCCESS')
254			except IOError as e:
255			self._close()
256			raise DriverException(str(e))
257
258			def _close(self) -> None:
259			print('USB CLOSE START')
260			if self._loop is not None:
261			if self._loop.is_alive():
262			self._loop.stop()
263			self._loop.join()
264			self._loop = None
265			try:
266			self._dev.reset()
267			usb.util.dispose_resources(self._dev)
268			except:
269			pass
270			self._dev = self._epOut = self._epIn = None
271			self._driver_open = False
272			print('USB CLOSE END')
273
274			def _read(self, count: int, timeout=None) -> bytes:
275			data = bytearray()
276			for i in range(0, count):
277			b = self._queue.get(timeout=timeout)
278			if b is None:
279			self._close()
280			raise DriverException("Device is closed!")
281			data.append(b)
282			return bytes(data)
283
284			def _write(self, data: bytes) -> None:
285			return self._epOut.write(data)
286
287
288			class USBLoop(Thread):
289			def __init__(self, ep, packetSize: int, queue: Queue):
290			super().__init__()
291			self._stopper = Event()
292			self._ep = ep
293			self._packetSize = packetSize
294			self._queue = queue
295
296			def stop(self) -> None:
297			self._stopper.set()
298
299			def run(self) -> None:
300			while not self._stopper.is_set():
301			try:
302			data = self._ep.read(self._packetSize, timeout=1000)
303			for d in data:
304			self._queue.put(d)
305			except usb.core.USBError as e:
306			if e.errno not in (60, 110) and e.backend_error_code != -116: # Timout errors
307			self._stopper.set()
308			# We Put in an invalid byte so threads will realize the device is stopped
309			self._queue.put(None)
310
311			class DummyDriver(Driver):
312			def __init__(self, log = None):
313			super().__init__(log = log)
314			self._isopen = False
315			self._data = Queue()
316			msg1 = Message(MESSAGE_CHANNEL_BROADCAST_DATA, b'\x00\x01\x02\x03\x04\x05\x06\x07').encode()
317			for b in msg1:
318			self._data.put(b)
319			msg2 = Message(MESSAGE_CHANNEL_BROADCAST_DATA, b'\x00\xF1\xF2\xF3\xF4\xF5\xF6\xF7').encode()
320			for b in msg2:
321			self._data.put(b)
322			msg3 = Message(MESSAGE_CHANNEL_BROADCAST_DATA, b'\x00\xF9\xFA\xFB\xFC\xFD\xFE\xFF').encode()
323			for b in msg3:
324			self._data.put(b)
325
326			def _isOpen(self) -> bool:
327			return self._isopen
328
329			def _close(self) -> None:
330			self._isopen = False
331
332			def _read(self, count: int, timeout=None) -> bytes:
333			data = bytearray()
334			for i in range(0, count):
335			data.append(self._data.get(timeout=timeout))
336			return bytes(data)
337
338			def _open(self) -> None:
339			self._isopen = True
340
341			def _write(self, data: bytes) -> None:
342			pass
343
344			class PcapDriver(Driver):
345			def __init__(self, pcap, log = None):
346			super().__init__(log = log)
347			self._isopen = False
348			self._pcap = pcap
349			self._buffer = Queue()
350
351			def _isOpen(self) -> bool:
352			return self._isopen
353
354			def _open(self) -> None:
355			self._isopen = True
356			self._pcapfile = open(self._pcap, 'rb')
357			self._EOF = os.stat(self._pcap).st_size
358			# move file pointer to first packet header
359			global_header_length = 24
360			self._pcapfile.seek(global_header_length, 0)
361
362
363			def _close(self) -> None:
364			self._isopen = False
365			self._pcapfile.close()
366
367			def _read(self, count: int, timeout=None) -> bytes:
368			def read_next_packet_into_buffer():
369			if self._pcapfile.tell() is self._EOF:
370			print("EOF")
371			return
372
373			ts_sec, = unpack('i', self._pcapfile.read(4))
374			ts_usec = unpack('i', self._pcapfile.read(4))[0]/1000000
375			ts = ts_sec + ts_usec
376			print("ts: ", ts)
377			time.sleep(ts)
378
379			packet_length = unpack('i', self._pcapfile.read(4))[0]
380			print("packet length: ", packet_length)
381			self._pcapfile.seek(4, 1)
382			for i in range(0, packet_length):
383			self._buffer.put(self._pcapfile.read(1))
384
385			result = bytearray()
386
387			print("Reading ", count, " byte(s)...")
388			while len(result) < count:
389			if self._buffer.empty():
390			print("reading next packet into buffer...")
391			read_next_packet_into_buffer()
392			try:
393			result.extend(self._buffer.get(block=True, timeout = timeout))
394			except Empty:
395			print("Timed out..")
396
397			print("result: ", bytes(result))
398			return bytes(result)
399
400			def _write(self, data: bytes) -> None:
401			pass
402

half2me / libant

GitHub Access Token became invalid

Pull Request — master (#3)

DummyDriver.__init__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

DummyDriver.init() A