SerialDriver.__str__() - Code Metrics - Inspection of "pcap logging" - half2me/libant - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — add-logging ( b569b9...b98437 )

unknown

created 2016-10-04 11:34 UTC

SerialDriver.str() A

↳ Parent: SerialDriver

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
c	0
b	0
f	0
dl	0
loc	4
rs	10
cc	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

Push — add-logging ( b569b9...b98437 )

SerialDriver.__str__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

SerialDriver.str() A