BOSWatch/BW3-Core
1
0
Fork 0
mirror of https://github.com/BOSWatch/BW3-Core.git synced 2026-03-17 10:14:45 +01:00
Code Issues Actions Packages Projects Releases Wiki Activity

fixes for pyLint

Browse source
This commit is contained in:
Bastian Schroll 2018-09-18 05:59:38 +02:00
parent 3564680867
commit 3f1eac88be
14 changed files with 35 additions and 34 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
boswatch/filter/doubeFilter.py
View file

@ -34,11 +34,11 @@ class DoubleFilter:
def filter(self, bwPacket): def filter(self, bwPacket):
if bwPacket.get("mode") is "fms": if bwPacket.get("mode") == "fms":
scanWord = "fms" scanWord = "fms"
elif bwPacket.get("mode") is "pocsag": elif bwPacket.get("mode") == "pocsag":
scanWord = "ric" scanWord = "ric"
elif bwPacket.get("mode") is "zvei": elif bwPacket.get("mode") == "zvei":
scanWord = "zvei" scanWord = "zvei"
else: else:
logging.error("No Filter for '%s'", bwPacket) logging.error("No Filter for '%s'", bwPacket)
@ -59,8 +59,8 @@ class DoubleFilter:
counter = 0 counter = 0
for listPacket in self._filterLists[bwPacket.get("mode")][1:]: # [1:] skip first entry, thats the new one for listPacket in self._filterLists[bwPacket.get("mode")][1:]: # [1:] skip first entry, thats the new one
if listPacket.get("timestamp") < (time.time() - self._config.getInt("doubleFilter", "IgnoreTime", "serverConfig")): if listPacket.get("timestamp") < (time.time() - self._config.getInt("doubleFilter", "IgnoreTime", "serverConfig")):
self._filterLists[bwPacket.get("mode")].remove(listPacket) self._filterLists[bwPacket.get("mode")].remove(listPacket)
counter += 1 counter += 1
if counter: if counter:
logging.debug("%d old entry(s) removed", counter) logging.debug("%d old entry(s) removed", counter)

12
boswatch/network/client.py
View file

@ -45,16 +45,16 @@ class TCPClient:
self._sock.setdefaulttimeout(self._timeout) self._sock.setdefaulttimeout(self._timeout)
self._sock = socket.create_connection((host, port)) self._sock = socket.create_connection((host, port))
logging.debug("connected to " + str(host) + ":" + str(port)) logging.debug("connected to %s:%s", host, port)
return True return True
except ConnectionRefusedError: except ConnectionRefusedError:
logging.error("cannot connect to %s:%s - connection refused", str(host), str(port)) logging.error("cannot connect to %s:%s - connection refused", host, port)
return False return False
except socket.timeout: # pragma: no cover except socket.timeout: # pragma: no cover
logging.warning("cannot connect to %s:%s - timeout after %s sec", str(host), str(port), self._timeout) logging.warning("cannot connect to %s:%s - timeout after %s sec", host, port, self._timeout)
return False return False
except: # pragma: no cover except: # pragma: no cover
logging.exception("cannot connect to %s:%s", str(host), str(port)) logging.exception("cannot connect to %s:%s", host, port)
return False return False
def disconnect(self): def disconnect(self):
@ -78,7 +78,7 @@ class TCPClient:
@param data: data to send to the server @param data: data to send to the server
@return True or False""" @return True or False"""
try: try:
logging.debug("transmitting: " + data) logging.debug("transmitting: %s", data)
self._sock.sendall(bytes(data + "\n", "utf-8")) self._sock.sendall(bytes(data + "\n", "utf-8"))
logging.debug("transmitted...") logging.debug("transmitted...")
return True return True
@ -98,7 +98,7 @@ class TCPClient:
@return received data""" @return received data"""
try: try:
received = str(self._sock.recv(1024), "utf-8") received = str(self._sock.recv(1024), "utf-8")
logging.debug("received: " + received) logging.debug("received: %d", received)
return received return received
except AttributeError: except AttributeError:
logging.error("cannot receive - no connection established") logging.error("cannot receive - no connection established")

12
boswatch/network/server.py
View file

@ -48,8 +48,8 @@ class TCPHandler(socketserver.BaseRequestHandler):
try: try:
while data: while data:
data = str(self.request.recv(1024).strip(), 'utf-8') data = str(self.request.recv(1024).strip(), 'utf-8')
if data is not "": if data != "":
logging.debug(req_name + " recv: " + data) logging.debug("%s recv: %s", req_name, data)
# add a new entry at first position (index 0) with client IP # add a new entry at first position (index 0) with client IP
# and the decoded data dict as an string in utf-8 and an timestamp # and the decoded data dict as an string in utf-8 and an timestamp
@ -57,14 +57,14 @@ class TCPHandler(socketserver.BaseRequestHandler):
_dataPackets.insert(0, (self.client_address[0], data, time.time())) # time() to calc time in queue _dataPackets.insert(0, (self.client_address[0], data, time.time())) # time() to calc time in queue
logging.debug("Add data to queue") logging.debug("Add data to queue")
logging.debug(req_name + " send: [ack]") logging.debug("%s send: [ack]", req_name)
self.request.sendall(bytes("[ack]", "utf-8")) self.request.sendall(bytes("[ack]", "utf-8"))
self.request.close() self.request.close()
except (ConnectionResetError, ConnectionAbortedError): # pragma: no cover except (ConnectionResetError, ConnectionAbortedError): # pragma: no cover
logging.debug(req_name + " connection closed") logging.debug("%s connection closed", req_name)
except: # pragma: no cover except: # pragma: no cover
logging.exception(req_name + " error while receiving") logging.exception("%s error while receiving", req_name)
finally: finally:
with _lockClients: with _lockClients:
del _clients[threading.current_thread().name] del _clients[threading.current_thread().name]
@ -100,7 +100,7 @@ class TCPServer(socketserver.ThreadingMixIn, socketserver.TCPServer):
self._server_thread.name = "Thread-BWServer" self._server_thread.name = "Thread-BWServer"
self._server_thread.daemon = True self._server_thread.daemon = True
self._server_thread.start() self._server_thread.start()
logging.debug("TCPServer started in Thread: " + self._server_thread.name) logging.debug("TCPServer started in Thread: %s", self._server_thread.name)
return True return True
except OSError: except OSError:
logging.exception("server always running?") logging.exception("server always running?")

9
boswatch/plugin/plugin.py
View file

@ -43,6 +43,7 @@ class Plugin:
self._setupTime = 0 self._setupTime = 0
self._alarmTime = 0 self._alarmTime = 0
self._teardownTime = 0 self._teardownTime = 0
self._endTime = 0
self._tmpTime = 0 self._tmpTime = 0
# for statistics # for statistics
@ -89,16 +90,16 @@ class Plugin:
self._tmpTime = time.time() self._tmpTime = time.time()
try: try:
if bwPacket.get("mode") is "fms": if bwPacket.get("mode") == "fms":
logging.debug("[%s] fms()", self._pluginName) logging.debug("[%s] fms()", self._pluginName)
self.fms(bwPacket) self.fms(bwPacket)
if bwPacket.get("mode") is "pocsag": if bwPacket.get("mode") == "pocsag":
logging.debug("[%s] pocsag()", self._pluginName) logging.debug("[%s] pocsag()", self._pluginName)
self.pocsag(bwPacket) self.pocsag(bwPacket)
if bwPacket.get("mode") is "zvei": if bwPacket.get("mode") == "zvei":
logging.debug("[%s] zvei()", self._pluginName) logging.debug("[%s] zvei()", self._pluginName)
self.zvei(bwPacket) self.zvei(bwPacket)
if bwPacket.get("mode") is "msg": if bwPacket.get("mode") == "msg":
logging.debug("[%s] msg()", self._pluginName) logging.debug("[%s] msg()", self._pluginName)
self.msg(bwPacket) self.msg(bwPacket)
except: except:

8
boswatch/utils/header.py
View file

@ -61,10 +61,10 @@ def infoToLog():
boswatch.version.date["day"], boswatch.version.date["day"],
boswatch.version.date["month"], boswatch.version.date["month"],
boswatch.version.date["year"]) boswatch.version.date["year"])
logging.debug("- Python version: " + platform.python_version()) logging.debug("- Python version: %s", platform.python_version())
logging.debug("- Python build: " + str(platform.python_build())) logging.debug("- Python build: %s", platform.python_build())
logging.debug("- System: " + platform.system()) logging.debug("- System: %s", platform.system())
logging.debug("- OS Version: " + platform.platform()) logging.debug("- OS Version: %s", platform.platform())
logging.debug("") logging.debug("")
return True return True
except: # pragma: no cover except: # pragma: no cover

2
test/test_ServerClient.py
View file

@ -26,7 +26,7 @@ class Test_ServerClient:
"""!Unittest for the server/client environment""" """!Unittest for the server/client environment"""
def setup_method(self, method): def setup_method(self, method):
logging.debug("[TEST] %s.%s" % (type(self).__name__, method.__name__)) logging.debug("[TEST] %s.%s", (type(self).__name__, method.__name__))
@pytest.fixture(scope="function") @pytest.fixture(scope="function")
def useServer(self): def useServer(self):

2
test/test_config.py
View file

@ -24,7 +24,7 @@ class Test_Config:
"""!Unittests for the config""" """!Unittests for the config"""
def setup_method(self, method): def setup_method(self, method):
logging.debug("[TEST] %s.%s" % (type(self).__name__, method.__name__)) logging.debug("[TEST] %s.%s", (type(self).__name__, method.__name__))
def test_loadLocalConfig(self): def test_loadLocalConfig(self):
"""!load a local config file""" """!load a local config file"""

2
test/test_decoder.py
View file

@ -23,7 +23,7 @@ class Test_Decoder:
"""!Unittests for the decoder""" """!Unittests for the decoder"""
def setup_method(self, method): def setup_method(self, method):
logging.debug("[TEST] %s.%s" % (type(self).__name__, method.__name__)) logging.debug("[TEST] %s.%s", (type(self).__name__, method.__name__))
def test_decoderNoData(self): def test_decoderNoData(self):
"""!Test a empty string""" """!Test a empty string"""

2
test/test_descriptor.py
View file

@ -25,7 +25,7 @@ class Test_Descriptor:
"""!Unittests for the descriptor""" """!Unittests for the descriptor"""
def setup_method(self, method): def setup_method(self, method):
logging.debug("[TEST] %s.%s" % (type(self).__name__, method.__name__)) logging.debug("[TEST] %s.%s", (type(self).__name__, method.__name__))
def test_loadCsvNotExist(self): def test_loadCsvNotExist(self):
"""!read CSV file where not exist direct per DescriptionList class""" """!read CSV file where not exist direct per DescriptionList class"""

2
test/test_doubleFilter.py
View file

@ -22,7 +22,7 @@ import boswatch.filter.doubeFilter
class Test_DoubleFilter: class Test_DoubleFilter:
def setup_method(self, method): def setup_method(self, method):
logging.debug("[TEST] %s.%s" % (type(self).__name__, method.__name__)) logging.debug("[TEST] %s.%s", (type(self).__name__, method.__name__))
def test_none(self): def test_none(self):
pass pass

2
test/test_header.py
View file

@ -23,7 +23,7 @@ class Test_Header:
"""!Unittests for the header""" """!Unittests for the header"""
def setup_method(self, method): def setup_method(self, method):
logging.debug("[TEST] %s.%s" % (type(self).__name__, method.__name__)) logging.debug("[TEST] %s.%s", (type(self).__name__, method.__name__))
def test_logoToLog(self): def test_logoToLog(self):
"""!Test logo to log""" """!Test logo to log"""

2
test/test_packet.py
View file

@ -24,7 +24,7 @@ class Test_Packet:
"""!Unittests for the BOSWatch packet""" """!Unittests for the BOSWatch packet"""
def setup_method(self, method): def setup_method(self, method):
logging.debug("[TEST] %s.%s" % (type(self).__name__, method.__name__)) logging.debug("[TEST] %s.%s", (type(self).__name__, method.__name__))
@pytest.fixture(scope="function") @pytest.fixture(scope="function")
def buildPacket(self): def buildPacket(self):

2
test/test_paths.py
View file

@ -24,7 +24,7 @@ class Test_Config:
"""!Unittests for the paths""" """!Unittests for the paths"""
def setup_method(self, method): def setup_method(self, method):
logging.debug("[TEST] %s.%s" % (type(self).__name__, method.__name__)) logging.debug("[TEST] %s.%s", (type(self).__name__, method.__name__))
def test_fileExists(self): def test_fileExists(self):
"""!load a local config file""" """!load a local config file"""

2
test/test_watchdog.py
View file

@ -22,7 +22,7 @@ import logging
class Test_Watchdog: class Test_Watchdog:
def setup_method(self, method): def setup_method(self, method):
logging.debug("[TEST] %s.%s" % (type(self).__name__, method.__name__)) logging.debug("[TEST] %s.%s", (type(self).__name__, method.__name__))
def test_none(self): def test_none(self):
pass pass