Pump.run() - Code Metrics - Inspection of "decoding" - half2me/libant - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( d5449d...f0eee5 )

by Benjamin

created 2016-09-04 21:43 UTC

Pump.run() B

↳ Parent: Pump

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	3
Bugs	0	Features	0

Metric	Value
c	3
b	0
f	0
dl	0
loc	13
rs	8
cc	6

import threading
from time import sleep

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


class Network:
    def __init__(self, key=b'\x00' * 8, name=None):
        self.key = key
        self.name = name
        self.number = 0

    def __str__(self):
        return self.name + self.key


class Pump(threading.Thread):
    def __init__(self, driver):
        super().__init__()
        self._stop = threading.Event()
        self._driver = driver

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

    def stopped(self):
        return self._stop.isSet()

    def run(self):
        with self._driver as driver:
            while not self._stop.is_set():
                sync = driver.read(1)  # search for sync
                if sync == MESSAGE_TX_SYNC or sync == MESSAGE_TX_SYNC_LEGACY:
                    size = driver.read(1) # get size of message
                    try:
                        raw = bytearray(sync)
                        raw.extend(size)
                        raw.extend(driver.read(size + 2))
                        message = Message.decode(raw)
                    except Exception as e:
                        pass


class Node:
    def __init__(self, driver, name=None):
        self._driver = driver
        self._name = name
        self._pump = Pump(driver)

    def __enter__(self):
        return self

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

    def start(self):
        if not self.isRunning():
            self._pump.start()

    def stop(self):
        if self.isRunning():
            self._pump.stop()
            self._pump.join()

    def isRunning(self):
        return self._pump.is_alive()

    def reset(self):
        self.stop()
        # TODO: reset device

    def getCapabilities(self):
        pass


1			import threading
2			from time import sleep
3
4			from libAnt.constants import MESSAGE_TX_SYNC, MESSAGE_TX_SYNC_LEGACY
5			from libAnt.message import Message
6
7
8			class Network:
9			def __init__(self, key=b'\x00' * 8, name=None):
10			self.key = key
11			self.name = name
12			self.number = 0
13
14			def __str__(self):
15			return self.name + self.key
16
17
18			class Pump(threading.Thread):
19			def __init__(self, driver):
20			super().__init__()
21			self._stop = threading.Event()
22			self._driver = driver
23
24			def stop(self):
25			self._stop.set()
26
27			def stopped(self):
28			return self._stop.isSet()
29
30			def run(self):
31			with self._driver as driver:
32			while not self._stop.is_set():
33			sync = driver.read(1) # search for sync
34			if sync == MESSAGE_TX_SYNC or sync == MESSAGE_TX_SYNC_LEGACY:
35			size = driver.read(1) # get size of message
36			try:
37			raw = bytearray(sync)
38			raw.extend(size)
39			raw.extend(driver.read(size + 2))
40			message = Message.decode(raw)
41			except Exception as e:
42			pass
43
44
45			class Node:
46			def __init__(self, driver, name=None):
47			self._driver = driver
48			self._name = name
49			self._pump = Pump(driver)
50
51			def __enter__(self):
52			return self
53
54			def __exit__(self, exc_type, exc_val, exc_tb):
55			self.stop()
56
57			def start(self):
58			if not self.isRunning():
59			self._pump.start()
60
61			def stop(self):
62			if self.isRunning():
63			self._pump.stop()
64			self._pump.join()
65
66			def isRunning(self):
67			return self._pump.is_alive()
68
69			def reset(self):
70			self.stop()
71			# TODO: reset device
72
73			def getCapabilities(self):
74			pass
75

half2me / libant

GitHub Access Token became invalid

Push — master ( d5449d...f0eee5 )

Pump.run() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like