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

Completed

Pull Request — master (#3)

by Benjamin

created 2017-01-25 12:31 UTC

Logger.close() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	1	Features	0

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

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

import math
from serial import Serial, SerialException, SerialTimeoutException

import usb
import time
from struct import *

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


class DriverException(Exception):
    pass

class Logger:
    def __init__(self, logFile: str):
        self._logFile = logFile
        self._log = None

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

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

    def open(self):
        def validate(logFile: str) -> str:
            if '.' in logFile:
                name, ext = logFile.split('.', 2)
                ext = '.' + ext
            else:
                name = logFile
                ext = ''
            num = 0
            exists = True
            while(exists):
                logFile = name + '-' + str(num) + ext
                try:
                    with open(logFile):
                            pass
                except IOError:
                    return logFile
                num += 1

        if self._log is not None:
            self.close()
        self._logFile = validate(self._logFile)
        self._log = open(self._logFile, 'wb')
        self.onOpen()

    def close(self):
        if self._log is not None:
            self.beforeClose()
            self._log.close()
            self.afterClose()

    def log(self, data: bytes):
        self._log.write(self.encodeData(data))

    def onOpen(self):
        pass

    def beforeClose(self):
        pass

    def afterClose(self):
        pass

    def encodeData(self, data):
        return data

class PcapLogger(Logger):
    def onOpen(self):
        # 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')
        self._log.write(
            pcap_global_header.pack(magic_number, version_major, version_minor, thiszone, sigfigs,
                                    snaplen, network))

    def encodeData(self, data):
        timestamp = time.time()
        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(data)
        orig_len = incl_len

        pcap_packet_header = Struct('<4s4sll').pack(ts_sec, ts_usec, incl_len, orig_len)
        return pcap_packet_header + data

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

    def __init__(self, logger: Logger = None):
        self._lock = Lock()
        self._logger = logger
        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._openTime = time.time()
                if self._logger is not None:
                    self._logger.open()
                self._open()

    def close(self) -> None:
        with self._lock:
            if self._isOpen:
                self._close()
                if self._logger is not None:
                    self._logger.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
                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._logger:
                    logMsg = bytearray([sync, length, type])
                    logMsg.extend(data)
                    logMsg.append(chk)

                    self._logger.log(bytes(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, logger: Logger = None):
        super().__init__(logger=logger)
        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, logger: Logger = None):
        super().__init__(logger=logger)
        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"

    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)

    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 = self.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 PcapDriver(Driver):
    def __init__(self, pcap, logger: Logger = None):
        super().__init__(logger=logger)
        self._isopen = False
        self._pcap = pcap
        self._buffer = Queue()

        self._loop = None

    class PcapLoop(Thread):
        def __init__(self, pcap, buffer: Queue):
            super().__init__()
            self._stopper = Event()
            self._pcap = pcap
            self._buffer = buffer

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

        def run(self) -> None:
            self._pcapfile = open(self._pcap, 'rb')
            # move file pointer to first packet header
            global_header_length = 24
            self._pcapfile.seek(global_header_length, 0)

            first_ts = 0
            start_time = time.time()
            while not self._stopper.is_set():
                try:
                    ts_sec, = unpack('i', self._pcapfile.read(4))
                except error:
                    break
                ts_usec = unpack('i', self._pcapfile.read(4))[0] / 1000000

                if first_ts is 0:
                    first_ts = ts_sec + ts_usec

                ts = ts_sec + ts_usec
                send_time = ts - first_ts
                elapsed_time = time.time() - start_time
                if send_time > (elapsed_time):
                    sleep_time = send_time - elapsed_time
                    time.sleep(sleep_time)

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

            self._pcapfile.close()

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

    def _open(self) -> None:
        self._isopen = True
        self._loop = self.PcapLoop(self._pcap, self._buffer)
        self._loop.start()

    def _close(self) -> None:
        self._isopen = False
        if self._loop is not None:
            if self._loop.is_alive():
                self._loop.stop()
                self._loop.join()
        self._loop = None

    def _read(self, count: int, timeout=None) -> bytes:
        result = bytearray()

        while len(result) < count:
            result += self._buffer.get(block=True, timeout=timeout)

        return bytes(result)

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


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

half2me / libant

GitHub Access Token became invalid

Pull Request — master (#3)

Logger.close() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like