BOSWatch/BW3-Core
1
0
Fork 0
mirror of https://github.com/BOSWatch/BW3-Core.git synced 2025-12-06 07:12:04 +01:00
Code Issues Actions Packages Projects Releases Wiki Activity

Merge pull request #112 from Luflosi/fix-deprecation-warning

Browse source 
Avoid "DeprecationWarning: invalid escape sequence"
This commit is contained in:
Bastian Schroll 2024-01-08 07:49:11 +01:00 committed by GitHub
parent 1b95474bc2 d4dcc75711
commit b7ba7f8627
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
50 changed files with 327 additions and 327 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
FileHead.template.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -22,8 +22,8 @@ logging.debug("- %s loaded", __name__)
class ClassName:
"""!General class comment"""
r"""!General class comment"""
def __init__(self):
"""!init comment"""
r"""!init comment"""
pass

10
boswatch/configYaml.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -34,15 +34,15 @@ class ConfigYAML:
yield item
def __len__(self):
"""!returns the length of an config element"""
r"""!returns the length of an config element"""
return len(self._config)
def __str__(self):
"""!Returns the string representation of the internal config dict"""
r"""!Returns the string representation of the internal config dict"""
return str(self._config)
def loadConfigFile(self, configPath):
"""!loads a given configuration file
r"""!loads a given configuration file
@param configPath: Path to the config file
@return True or False"""
@ -59,7 +59,7 @@ class ConfigYAML:
return False
def get(self, *args, default=None):
"""!Get a single value from the config
r"""!Get a single value from the config
or a value set in a new configYAML class instance
@param *args: Config section (one ore more strings)

4
boswatch/decoder/decoder.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -27,7 +27,7 @@ class Decoder:
@staticmethod
def decode(data):
"""!Choose the right decoder and return a bwPacket instance
r"""!Choose the right decoder and return a bwPacket instance
@param data: data to decode
@return bwPacket instance"""

6
boswatch/decoder/fmsDecoder.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -24,7 +24,7 @@ logging.debug("- %s loaded", __name__)
class FmsDecoder:
"""!FMS decoder class
r"""!FMS decoder class
This class decodes FMS data.
First step is to validate the data and _check if the format is correct.
@ -32,7 +32,7 @@ class FmsDecoder:
@staticmethod
def decode(data):
"""!Decodes FMS
r"""!Decodes FMS
@param data: FMS for decoding
@return BOSWatch FMS packet or None"""

8
boswatch/decoder/pocsagDecoder.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -24,7 +24,7 @@ logging.debug("- %s loaded", __name__)
class PocsagDecoder:
"""!POCSAG decoder class
r"""!POCSAG decoder class
This class decodes POCSAG data.
First step is to validate the data and _check if the format is correct.
@ -32,7 +32,7 @@ class PocsagDecoder:
@staticmethod
def decode(data):
"""!Decodes POCSAG
r"""!Decodes POCSAG
@param data: POCSAG for decoding
@return BOSWatch POCSAG packet or None"""
@ -63,7 +63,7 @@ class PocsagDecoder:
@staticmethod
def _getBitrateRicSubric(data):
"""!Gets the Bitrate, Ric and Subric from data
r"""!Gets the Bitrate, Ric and Subric from data
@param data: POCSAG data string
@return bitrate

8
boswatch/decoder/zveiDecoder.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -24,7 +24,7 @@ logging.debug("- %s loaded", __name__)
class ZveiDecoder:
"""!ZVEI decoder class
r"""!ZVEI decoder class
This class decodes ZVEI data.
First step is to validate the data and _check if the format is correct.
@ -33,7 +33,7 @@ class ZveiDecoder:
@staticmethod
def decode(data):
"""!Decodes ZVEI
r"""!Decodes ZVEI
@param data: ZVEI for decoding
@return BOSWatch ZVEI packet or None"""
@ -51,7 +51,7 @@ class ZveiDecoder:
@staticmethod
def _solveDoubleTone(data):
"""!Remove the doubleTone sign (here its the 'E')
r"""!Remove the doubleTone sign (here its the 'E')
@param data: ZVEI for double tone sign replacement
@return Double Tone replaced ZVEI"""

14
boswatch/inputSource/inputBase.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -26,10 +26,10 @@ logging.debug("- %s loaded", __name__)
class InputBase(ABC):
"""!Base class for handling inout sources"""
r"""!Base class for handling inout sources"""
def __init__(self, inputQueue, inputConfig, decoderConfig):
"""!Build a new InputSource class
r"""!Build a new InputSource class
@param inputQueue: Python queue object to store input data
@param inputConfig: ConfigYaml object with the inoutSource config
@ -41,7 +41,7 @@ class InputBase(ABC):
self._decoderConfig = decoderConfig
def start(self):
"""!Start the input source thread"""
r"""!Start the input source thread"""
logging.debug("starting input thread")
self._isRunning = True
self._inputThread = threading.Thread(target=self._runThread, name="inputThread",
@ -51,10 +51,10 @@ class InputBase(ABC):
@abstractmethod
def _runThread(self, dataQueue, sdrConfig, decoderConfig):
"""!Thread routine of the input source has to be inherit"""
r"""!Thread routine of the input source has to be inherit"""
def shutdown(self):
"""!Stop the input source thread"""
r"""!Stop the input source thread"""
if self._isRunning:
logging.debug("wait for stopping the input thread")
self._isRunning = False
@ -62,7 +62,7 @@ class InputBase(ABC):
logging.debug("input thread stopped")
def addToQueue(self, data):
"""!Decode and add alarm data to the queue for further processing during boswatch client"""
r"""!Decode and add alarm data to the queue for further processing during boswatch client"""
bwPacket = Decoder.decode(data)
if bwPacket is not None:
self._inputQueue.put_nowait((bwPacket, time.time()))

4
boswatch/inputSource/lineInInput.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -23,7 +23,7 @@ logging.debug("- %s loaded", __name__)
class LineInInput(InputBase):
"""!Class for the line-in input source"""
r"""!Class for the line-in input source"""
def _runThread(self, dataQueue, lineInConfig, decoderConfig):
lineInProc = None

4
boswatch/inputSource/pulseaudioInput.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -23,7 +23,7 @@ logging.debug("- %s loaded", __name__)
class PulseAudioInput(InputBase):
"""!Class for the PulseAudio input source"""
r"""!Class for the PulseAudio input source"""
def _runThread(self, dataQueue, PulseAudioConfig, decoderConfig):
PulseAudioProc = None

4
boswatch/inputSource/sdrInput.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -24,7 +24,7 @@ logging.debug("- %s loaded", __name__)
class SdrInput(InputBase):
"""!Class for the sdr input source"""
r"""!Class for the sdr input source"""
def _runThread(self, dataQueue, sdrConfig, decoderConfig):
sdrProc = None

24
boswatch/network/broadcast.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -22,10 +22,10 @@ logging.debug("- %s loaded", __name__)
class BroadcastClient:
"""!BroadcastClient class"""
r"""!BroadcastClient class"""
def __init__(self, port=5000):
"""!Create an BroadcastClient instance
r"""!Create an BroadcastClient instance
@param port: port to send broadcast packets (5000)"""
self._broadcastPort = port
@ -39,7 +39,7 @@ class BroadcastClient:
self._socket.settimeout(3)
def getConnInfo(self, retry=0):
"""!Get the connection info from server over udp broadcast
r"""!Get the connection info from server over udp broadcast
This function will send broadcast package(s)
to get connection info from the server.
@ -73,20 +73,20 @@ class BroadcastClient:
@property
def serverIP(self):
"""!Property to get the server IP after successful broadcast"""
r"""!Property to get the server IP after successful broadcast"""
return self._serverIP
@property
def serverPort(self):
"""!Property to get the server Port after successful broadcast"""
r"""!Property to get the server Port after successful broadcast"""
return self._serverPort
class BroadcastServer:
"""!BroadcastServer class"""
r"""!BroadcastServer class"""
def __init__(self, servePort=8080, listenPort=5000):
"""!Create an BroadcastServer instance
r"""!Create an BroadcastServer instance
@param servePort: port to serve as connection info (8080)
@param listenPort: port to listen for broadcast packets (5000)"""
@ -106,7 +106,7 @@ class BroadcastServer:
pass
def start(self):
"""!Start the broadcast server in a new thread
r"""!Start the broadcast server in a new thread
@return True or False"""
if not self.isRunning:
@ -121,7 +121,7 @@ class BroadcastServer:
return True
def stop(self):
"""!Stop the broadcast server
r"""!Stop the broadcast server
Due to the timeout of the socket,
stopping the thread can be delayed by two seconds.
@ -138,7 +138,7 @@ class BroadcastServer:
return True
def _listen(self):
"""!Broadcast server worker thread
r"""!Broadcast server worker thread
This function listen for magic packets on broadcast
address and send the connection info to the clients.
@ -161,7 +161,7 @@ class BroadcastServer:
@property
def isRunning(self):
"""!Property of broadcast server running state"""
r"""!Property of broadcast server running state"""
if self._serverThread:
return True
return False

16
boswatch/network/client.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -24,17 +24,17 @@ HEADERSIZE = 10
class TCPClient:
"""!TCP client class"""
r"""!TCP client class"""
def __init__(self, timeout=3):
"""!Create a new instance
r"""!Create a new instance
@param timeout: timeout for the client in sec. (3)"""
socket.setdefaulttimeout(timeout)
self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
def connect(self, host="localhost", port=8080):
"""!Connect to the server
r"""!Connect to the server
@param host: Server IP address ("localhost")
@param port: Server Port (8080)
@ -52,7 +52,7 @@ class TCPClient:
return False
def disconnect(self):
"""!Disconnect from the server
r"""!Disconnect from the server
@return True or False"""
try:
@ -68,7 +68,7 @@ class TCPClient:
return False
def transmit(self, data):
"""!Send a data packet to the server
r"""!Send a data packet to the server
@param data: data to send to the server
@return True or False"""
@ -84,7 +84,7 @@ class TCPClient:
return False
def receive(self, timeout=1):
"""!Receive data from the server
r"""!Receive data from the server
@param timeout: to wait for incoming data in seconds
@return received data"""
@ -109,7 +109,7 @@ class TCPClient:
@property
def isConnected(self):
"""!Property of client connected state"""
r"""!Property of client connected state"""
try:
if self._sock:
_, write, _ = select.select([], [self._sock], [], 0.1)

8
boswatch/network/netCheck.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -21,10 +21,10 @@ logging.debug("- %s loaded", __name__)
class NetCheck:
"""!Worker class to check internet connection"""
r"""!Worker class to check internet connection"""
def __init__(self, hostname="https://www.google.com/", timeout=1):
"""!Create a new NetCheck instance
r"""!Create a new NetCheck instance
@param hostname: host against connection check is running ("https://www.google.com/")
@param timeout: timeout for connection check in sec. (1)"""
@ -34,7 +34,7 @@ class NetCheck:
self.checkConn() # initiate a first check
def checkConn(self):
"""!Check the connection
r"""!Check the connection
@return True or False"""
try:

22
boswatch/network/server.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -27,10 +27,10 @@ HEADERSIZE = 10
class _ThreadedTCPRequestHandler(socketserver.BaseRequestHandler):
"""!ThreadedTCPRequestHandler class for our TCPServer class."""
r"""!ThreadedTCPRequestHandler class for our TCPServer class."""
def handle(self):
"""!Handles the request from an single client in a own thread
r"""!Handles the request from an single client in a own thread
Insert a request in the clients[] list and send a [ack]"""
with self.server.clientsConnectedLock: # because our list is not threadsafe
@ -79,15 +79,15 @@ class _ThreadedTCPRequestHandler(socketserver.BaseRequestHandler):
class _ThreadedTCPServer(socketserver.ThreadingMixIn, socketserver.TCPServer):
"""!ThreadedTCPServer class for our TCPServer class."""
r"""!ThreadedTCPServer class for our TCPServer class."""
pass
class TCPServer:
"""!TCP server class"""
r"""!TCP server class"""
def __init__(self, alarmQueue, timeout=3):
"""!Create a new instance
r"""!Create a new instance
@param alarmQueue: python queue instance
@param timeout: server timeout in sec (3)
@ -105,7 +105,7 @@ class TCPServer:
self.stop()
def start(self, port=8080):
"""!Start a threaded TCP socket server
r"""!Start a threaded TCP socket server
Start a TCP Socket Server in a new thread that will
then start one more thread for each client request.
@ -139,7 +139,7 @@ class TCPServer:
return True
def stop(self):
"""!Stops the TCP socket server
r"""!Stops the TCP socket server
@return True or False"""
if self.isRunning:
@ -155,14 +155,14 @@ class TCPServer:
return True
def countClientsConnected(self):
"""!Number of currently connected Clients
r"""!Number of currently connected Clients
@return Connected clients"""
with self._clientsConnectedLock: # because our list is not threadsafe
return len(self._clientsConnected)
def getClientsConnected(self):
"""!A list of all connected clients
r"""!A list of all connected clients
with their IP address and last seen timestamp
_clients[ThreadName] = {"address", "timestamp"}
@ -173,7 +173,7 @@ class TCPServer:
@property
def isRunning(self):
"""!Property of server running state"""
r"""!Property of server running state"""
if self._server:
return True
return False

14
boswatch/packet.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -21,10 +21,10 @@ logging.debug("- %s loaded", __name__)
class Packet:
"""!Class implementation of an BOSWatch packet"""
r"""!Class implementation of an BOSWatch packet"""
def __init__(self, bwPacket=None):
"""!Build a new BOSWatch packet or copy existing data in it
r"""!Build a new BOSWatch packet or copy existing data in it
@param bwPacket: Existing data to copy"""
if bwPacket is None:
@ -35,18 +35,18 @@ class Packet:
self._packet = eval(str(bwPacket.strip()))
def __str__(self):
"""!Return the intern _packet dict as string"""
r"""!Return the intern _packet dict as string"""
return str(self._packet)
def set(self, fieldName, value):
"""!Set a field in the intern _packet dict
r"""!Set a field in the intern _packet dict
@param fieldName: Name of the data to set
@param value: Value to set"""
self._packet[fieldName] = str(value)
def get(self, fieldName):
"""!Returns the value from a single field.
r"""!Returns the value from a single field.
If field not existing `None` is returned
@param fieldName: Name of the field
@ -58,7 +58,7 @@ class Packet:
return None
def printInfo(self):
"""!Print a info message to the log on INFO level.
r"""!Print a info message to the log on INFO level.
Contains the most useful info about this packet.
@todo not complete yet - must be edit to print nice formatted messages on console
"""

30
boswatch/processManager.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -21,7 +21,7 @@ logging.debug("- %s loaded", __name__)
class ProcessManager:
"""!class to manage a extern sub process"""
r"""!class to manage a extern sub process"""
def __init__(self, process, textMode=False):
logging.debug("create process instance %s - textMode: %s", process, textMode)
self._args = []
@ -33,7 +33,7 @@ class ProcessManager:
self._textMode = textMode
def addArgument(self, arg):
"""!add a new argument
r"""!add a new argument
@param arg: argument to add as string"""
logging.debug("add argument to process: %s -> %s", self._args[0], arg)
@ -41,11 +41,11 @@ class ProcessManager:
self._args.append(splitArg)
def clearArguments(self):
"""!clear all arguments"""
r"""!clear all arguments"""
self._args = self._args[0:1] # kept first element (process name)
def start(self):
"""!start the new process
r"""!start the new process
@return: True or False"""
logging.debug("start new process: %s %s", self._args[0], self._args[1:])
@ -67,7 +67,7 @@ class ProcessManager:
return False
def stop(self):
"""!Stop the process by sending SIGTERM and wait for ending"""
r"""!Stop the process by sending SIGTERM and wait for ending"""
logging.debug("stopping process: %s", self._args[0])
if self.isRunning:
self._processHandle.terminate()
@ -76,7 +76,7 @@ class ProcessManager:
logging.debug("process %s returned %d", self._args[0], self._processHandle.returncode)
def readline(self):
"""!Read one line from stdout stream
r"""!Read one line from stdout stream
@return singe line or None"""
if self.isRunning and self._stdout is not None:
@ -88,7 +88,7 @@ class ProcessManager:
return None
def skipLines(self, lineCount=1):
"""!Skip given number of lines from the output
r"""!Skip given number of lines from the output
@param lineCount: number of lines to skip
"""
@ -98,7 +98,7 @@ class ProcessManager:
lineCount -= 1
def skipLinesUntil(self, matchText):
"""!Skip lines from the output until the given string is in it
r"""!Skip lines from the output until the given string is in it
@param matchText: string to search for in output
"""
@ -109,30 +109,30 @@ class ProcessManager:
pass
def setStdin(self, stdin):
"""!Set the stdin stream instance"""
r"""!Set the stdin stream instance"""
self._stdin = stdin
def setStdout(self, stdout):
"""!Set the stdout stream instance"""
r"""!Set the stdout stream instance"""
self._stdout = stdout
def setStderr(self, stderr):
"""!Set the stderr stream instance"""
r"""!Set the stderr stream instance"""
self._stderr = stderr
@property
def stdout(self):
"""!Property to get the stdout stream"""
r"""!Property to get the stdout stream"""
return self._processHandle.stdout
@property
def stderr(self):
"""!Property to get the stderr stream"""
r"""!Property to get the stderr stream"""
return self._processHandle.stderr
@property
def isRunning(self):
"""!Property to get process running state
r"""!Property to get process running state
@return True or False"""
if self._processHandle:

6
boswatch/router/route.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -21,9 +21,9 @@ logging.debug("- %s loaded", __name__)
class Route:
"""!Class for single routing points"""
r"""!Class for single routing points"""
def __init__(self, name, callback, statsCallback=None, cleanupCallback=None):
"""!Create a instance of an route point
r"""!Create a instance of an route point
@param name: name of the route point
@param callback: instance of the callback function

12
boswatch/router/router.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -22,9 +22,9 @@ logging.debug("- %s loaded", __name__)
class Router:
"""!Class for the Router"""
r"""!Class for the Router"""
def __init__(self, name):
"""!Create a new router
r"""!Create a new router
@param name: name of the router"""
self.name = name
@ -40,7 +40,7 @@ class Router:
logging.debug("[%s] add new router", self.name)
def addRoute(self, route):
"""!Adds a route point to the router
r"""!Adds a route point to the router
@param route: instance of the Route class
"""
@ -48,7 +48,7 @@ class Router:
self.routeList.append(route)
def runRouter(self, bwPacket):
"""!Run the router
r"""!Run the router
@param bwPacket: instance of Packet class
@return a instance of Packet class
@ -79,7 +79,7 @@ class Router:
return bwPacket
def _getStatistics(self):
"""!Returns statistical information's from last router run
r"""!Returns statistical information's from last router run
@return Statistics as pyton dict"""
stats = {"type": "router",

16
boswatch/router/routerManager.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -27,16 +27,16 @@ logging.debug("- %s loaded", __name__)
class RouterManager:
"""!Class to manage all routers"""
r"""!Class to manage all routers"""
def __init__(self):
"""!Create new router"""
r"""!Create new router"""
self._routerDict = {}
self._startTime = int(time.time())
# if there is an error, router list would be empty (see tmp variable)
def buildRouters(self, config):
"""!Initialize Routers from given config file
r"""!Initialize Routers from given config file
@param config: instance of ConfigYaml class
@return True or False"""
@ -103,7 +103,7 @@ class RouterManager:
return True
def runRouters(self, routerRunList, bwPacket):
"""!Run given Routers
r"""!Run given Routers
@param routerRunList: string or list of router names in string form
@param bwPacket: instance of Packet class"""
@ -119,7 +119,7 @@ class RouterManager:
self._saveStats() # write stats to stats file
def cleanup(self):
"""!Run cleanup routines for all loaded route points"""
r"""!Run cleanup routines for all loaded route points"""
for name, routerObject in self._routerDict.items():
logging.debug("Start cleanup for %s", name)
for routePoint in routerObject.routeList:
@ -127,7 +127,7 @@ class RouterManager:
routePoint.cleanup()
def _showRouterRoute(self):
"""!Show the routes of all routers"""
r"""!Show the routes of all routers"""
for name, routerObject in self._routerDict.items():
logging.debug("Route for %s", name)
counter = 0
@ -136,7 +136,7 @@ class RouterManager:
logging.debug(" %d. %s", counter, routePoint.name)
def _saveStats(self):
"""!Save current statistics to file"""
r"""!Save current statistics to file"""
lines = []
for name, routerObject in self._routerDict.items():
lines.append("[" + name + "]")

14
boswatch/timer.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -24,7 +24,7 @@ logging.debug("- %s loaded", __name__)
class RepeatedTimer:
def __init__(self, interval, targetFunction, *args, **kwargs):
"""!Create a new instance of the RepeatedTimer
r"""!Create a new instance of the RepeatedTimer
@param interval: interval in sec. to recall target function
@param targetFunction: function to call on timer event
@ -43,7 +43,7 @@ class RepeatedTimer:
self._thread = None
def start(self):
"""!Start a new timer worker thread
r"""!Start a new timer worker thread
@return True or False"""
if self._thread is None:
@ -58,7 +58,7 @@ class RepeatedTimer:
return True
def stop(self):
"""!Stop the timer worker thread
r"""!Stop the timer worker thread
@return True or False"""
if self._thread is not None:
@ -71,7 +71,7 @@ class RepeatedTimer:
return True
def _target(self):
"""!Runs the target function with his arguments in own thread"""
r"""!Runs the target function with his arguments in own thread"""
self._start = time.time()
while not self._event.wait(self.restTime):
logging.debug("work")
@ -96,12 +96,12 @@ class RepeatedTimer:
@property
def isRunning(self):
"""!Property for repeatedTimer running state"""
r"""!Property for repeatedTimer running state"""
if self._thread:
return True
return False
@property
def restTime(self):
"""!Property to get remaining time till next call"""
r"""!Property to get remaining time till next call"""
return self._interval - ((time.time() - self._start) % self._interval)

22
boswatch/utils/header.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -23,22 +23,22 @@ logging.debug("- %s loaded", __name__)
def logoToLog():
"""!Prints the BOSWatch logo to the log at debug level
r"""!Prints the BOSWatch logo to the log at debug level
@return True or False on error"""
logging.debug(" ____ ____ ______ __ __ __ _____ ")
logging.debug(" / __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ / ")
logging.debug(" / __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ < ")
logging.debug(" / /_/ / /_/ /___/ /| |/ |/ / /_/ / /_/ /__/ / / / ___/ / ")
logging.debug("/_____/\____//____/ |__/|__/\__,_/\__/\___/_/ /_/ /____/ ")
logging.debug(" German BOS Information Script ")
logging.debug(" by Bastian Schroll ")
logging.debug("")
logging.debug(r" ____ ____ ______ __ __ __ _____ ")
logging.debug(r" / __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ / ")
logging.debug(r" / __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ < ")
logging.debug(r" / /_/ / /_/ /___/ /| |/ |/ / /_/ / /_/ /__/ / / / ___/ / ")
logging.debug(r"/_____/\____//____/ |__/|__/\__,_/\__/\___/_/ /_/ /____/ ")
logging.debug(r" German BOS Information Script ")
logging.debug(r" by Bastian Schroll ")
logging.debug(r"")
return True
def infoToLog():
"""!Prints the BOSWatch and OS information to log at debug level
r"""!Prints the BOSWatch and OS information to log at debug level
@return True or False on error"""
logging.debug("BOSWatch and environment information")

6
boswatch/utils/misc.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -21,7 +21,7 @@ logging.debug("- %s loaded", __name__)
def addClientDataToPacket(bwPacket, config):
"""!Add the client information to the decoded data
r"""!Add the client information to the decoded data
This function adds the following data to the bwPacket:
- clientName
@ -40,7 +40,7 @@ def addClientDataToPacket(bwPacket, config):
def addServerDataToPacket(bwPacket, config):
"""!Add the server information to the decoded data
r"""!Add the server information to the decoded data
This function adds the following data to the bwPacket:
- serverName

4
boswatch/utils/paths.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -34,7 +34,7 @@ def fileExist(filePath):
def makeDirIfNotExist(dirPath):
"""!Checks if an directory is existing and create it if not
r"""!Checks if an directory is existing and create it if not
@param dirPath: Path of the directory
@return Path of the directory or False"""

2
boswatch/utils/version.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <

6
boswatch/wildcard.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -25,7 +25,7 @@ _additionalWildcards = {}
def registerWildcard(wildcard, bwPacketField):
"""!Register a new additional wildcard
r"""!Register a new additional wildcard
@param wildcard: New wildcard string with format: '{WILDCARD}'
@param bwPacketField: Field of the bwPacket which is used for wildcard replacement"""
@ -37,7 +37,7 @@ def registerWildcard(wildcard, bwPacketField):
def replaceWildcards(message, bwPacket):
"""!Replace the wildcards in a given message
r"""!Replace the wildcards in a given message
@param message: Message in which wildcards should be replaced
@param bwPacket: bwPacket instance with the replacement information

2
bw_client.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <

2
bw_server.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <

12
module/descriptor.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -26,19 +26,19 @@ logging.debug("- %s loaded", __name__)
class BoswatchModule(ModuleBase):
"""!Adds descriptions to bwPackets"""
r"""!Adds descriptions to bwPackets"""
def __init__(self, config):
"""!Do not change anything here!"""
r"""!Do not change anything here!"""
super().__init__(__name__, config) # you can access the config class on 'self.config'
def onLoad(self):
"""!Called by import of the plugin"""
r"""!Called by import of the plugin"""
for descriptor in self.config:
if descriptor.get("wildcard", default=None):
self.registerWildcard(descriptor.get("wildcard"), descriptor.get("descrField"))
def doWork(self, bwPacket):
"""!start an run of the module.
r"""!start an run of the module.
@param bwPacket: A BOSWatch packet instance"""
for descriptor in self.config:
@ -54,5 +54,5 @@ class BoswatchModule(ModuleBase):
return bwPacket
def onUnload(self):
"""!Called by destruction of the plugin"""
r"""!Called by destruction of the plugin"""
pass

12
module/filter/doubleFilter.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -26,21 +26,21 @@ logging.debug("- %s loaded", __name__)
class BoswatchModule(ModuleBase):
"""!Description of the Module"""
r"""!Description of the Module"""
def __init__(self, config):
"""!Do not change anything here!"""
r"""!Do not change anything here!"""
super().__init__(__name__, config) # you can access the config class on 'self.config'
self._filterLists = {}
logging.debug("Configured ignoreTime: %d", self.config.get("ignoreTime", default=10))
logging.debug("Configured maxEntry: %d", self.config.get("maxEntry", default=10))
def onLoad(self):
"""!Called by import of the plugin
r"""!Called by import of the plugin
Remove if not implemented"""
pass
def doWork(self, bwPacket):
"""!start an run of the module.
r"""!start an run of the module.
@param bwPacket: A BOSWatch packet instance"""
if bwPacket.get("mode") == "fms":
@ -62,7 +62,7 @@ class BoswatchModule(ModuleBase):
return self._check(bwPacket, filterFields)
def onUnload(self):
"""!Called by destruction of the plugin
r"""!Called by destruction of the plugin
Remove if not implemented"""
pass

12
module/filter/modeFilter.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -26,17 +26,17 @@ logging.debug("- %s loaded", __name__)
class BoswatchModule(ModuleBase):
"""!Filter of specific bwPacket mode"""
r"""!Filter of specific bwPacket mode"""
def __init__(self, config):
"""!Do not change anything here!"""
r"""!Do not change anything here!"""
super().__init__(__name__, config) # you can access the config class on 'self.config'
def onLoad(self):
"""!Called by import of the plugin"""
r"""!Called by import of the plugin"""
pass
def doWork(self, bwPacket):
"""!start an run of the module.
r"""!start an run of the module.
@param bwPacket: A BOSWatch packet instance"""
@ -48,5 +48,5 @@ class BoswatchModule(ModuleBase):
return False
def onUnload(self):
"""!Called by destruction of the plugin"""
r"""!Called by destruction of the plugin"""
pass

12
module/filter/regexFilter.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -26,17 +26,17 @@ logging.debug("- %s loaded", __name__)
class BoswatchModule(ModuleBase):
"""!Regex based filter mechanism"""
r"""!Regex based filter mechanism"""
def __init__(self, config):
"""!Do not change anything here!"""
r"""!Do not change anything here!"""
super().__init__(__name__, config) # you can access the config class on 'self.config'
def onLoad(self):
"""!Called by import of the plugin"""
r"""!Called by import of the plugin"""
pass
def doWork(self, bwPacket):
"""!start an run of the module.
r"""!start an run of the module.
@param bwPacket: A BOSWatch packet instance"""
for regexFilter in self.config:
@ -61,5 +61,5 @@ class BoswatchModule(ModuleBase):
return False # False -> Router will stop further processing
def onUnload(self):
"""!Called by destruction of the plugin"""
r"""!Called by destruction of the plugin"""
pass

10
module/geocoding.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -27,13 +27,13 @@ logging.debug("- %s loaded", __name__)
class BoswatchModule(ModuleBase):
"""!Description of the Module"""
r"""!Description of the Module"""
def __init__(self, config):
"""!Do not change anything here!"""
r"""!Do not change anything here!"""
super().__init__(__name__, config) # you can access the config class on 'self.config'
def doWork(self, bwPacket):
"""!start an run of the module.
r"""!start an run of the module.
@param bwPacket: A BOSWatch packet instance"""
if bwPacket.get("mode") == "pocsag":
@ -42,7 +42,7 @@ class BoswatchModule(ModuleBase):
return bwPacket
def geocode(self, bwPacket):
"""!find address in message and get latitude and longitude
r"""!find address in message and get latitude and longitude
@param bwPacket: A BOSWatch packet instance"""
try:

20
module/moduleBase.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -24,12 +24,12 @@ logging.debug("- %s loaded", __name__)
class ModuleBase(ABC):
"""!Main module class"""
r"""!Main module class"""
_modulesActive = []
def __init__(self, moduleName, config):
"""!init preload some needed locals and then call onLoad() directly"""
r"""!init preload some needed locals and then call onLoad() directly"""
self._moduleName = moduleName
self.config = config
self._modulesActive.append(self)
@ -46,13 +46,13 @@ class ModuleBase(ABC):
self.onLoad()
def _cleanup(self):
"""!Cleanup routine calls onUnload() directly"""
r"""!Cleanup routine calls onUnload() directly"""
logging.debug("[%s] onUnload()", self._moduleName)
self._modulesActive.remove(self)
self.onUnload()
def _run(self, bwPacket):
"""!start an run of the module.
r"""!start an run of the module.
@param bwPacket: A BOSWatch packet instance
@return bwPacket or False"""
@ -75,7 +75,7 @@ class ModuleBase(ABC):
return bwPacket
def _getStatistics(self):
"""!Returns statistical information's from last module run
r"""!Returns statistical information's from last module run
@return Statistics as pyton dict"""
stats = {"type": "module",
@ -86,25 +86,25 @@ class ModuleBase(ABC):
return stats
def onLoad(self):
"""!Called by import of the module
r"""!Called by import of the module
can be inherited"""
pass
def doWork(self, bwPacket):
"""!Called module run
r"""!Called module run
can be inherited
@param bwPacket: bwPacket instance"""
logging.warning("no functionality in module %s", self._moduleName)
def onUnload(self):
"""!Called on shutdown of boswatch
r"""!Called on shutdown of boswatch
can be inherited"""
pass
@staticmethod
def registerWildcard(newWildcard, bwPacketField):
"""!Register a new wildcard
r"""!Register a new wildcard
@param newWildcard: wildcard where parser searching for
@param bwPacketField: field from bwPacket where holds replacement data"""

12
module/template_module.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -26,18 +26,18 @@ logging.debug("- %s loaded", __name__)
class BoswatchModule(ModuleBase):
"""!Description of the Module"""
r"""!Description of the Module"""
def __init__(self, config):
"""!Do not change anything here!"""
r"""!Do not change anything here!"""
super().__init__(__name__, config) # you can access the config class on 'self.config'
def onLoad(self):
"""!Called by import of the plugin
r"""!Called by import of the plugin
Remove if not implemented"""
pass
def doWork(self, bwPacket):
"""!start an run of the module.
r"""!start an run of the module.
@param bwPacket: A BOSWatch packet instance"""
if bwPacket.get("mode") == "fms":
@ -52,6 +52,6 @@ class BoswatchModule(ModuleBase):
return bwPacket
def onUnload(self):
"""!Called by destruction of the plugin
r"""!Called by destruction of the plugin
Remove if not implemented"""
pass

14
plugin/divera.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -28,13 +28,13 @@ logging.debug("- %s loaded", __name__)
class BoswatchPlugin(PluginBase):
"""!Description of the Plugin"""
r"""!Description of the Plugin"""
def __init__(self, config):
"""!Do not change anything here!"""
r"""!Do not change anything here!"""
super().__init__(__name__, config) # you can access the config class on 'self.config'
def fms(self, bwPacket):
"""!Called on FMS alarm
r"""!Called on FMS alarm
@param bwPacket: bwPacket instance
Remove if not implemented"""
@ -52,7 +52,7 @@ class BoswatchPlugin(PluginBase):
self._makeRequests(apipath, apicall)
def pocsag(self, bwPacket):
"""!Called on POCSAG alarm
r"""!Called on POCSAG alarm
@param bwPacket: bwPacket instance
Remove if not implemented"""
@ -68,7 +68,7 @@ class BoswatchPlugin(PluginBase):
self._makeRequests(apipath, apicall)
def zvei(self, bwPacket):
"""!Called on ZVEI alarm
r"""!Called on ZVEI alarm
@param bwPacket: bwPacket instance
Remove if not implemented"""
@ -84,7 +84,7 @@ class BoswatchPlugin(PluginBase):
self._makeRequests(apipath, apicall)
def msg(self, bwPacket):
"""!Called on MSG packet
r"""!Called on MSG packet
@param bwPacket: bwPacket instance
Remove if not implemented"""

14
plugin/http.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -27,13 +27,13 @@ logging.debug("- %s loaded", __name__)
class BoswatchPlugin(PluginBase):
"""!Description of the Plugin"""
r"""!Description of the Plugin"""
def __init__(self, config):
"""!Do not change anything here!"""
r"""!Do not change anything here!"""
super().__init__(__name__, config) # you can access the config class on 'self.config'
def fms(self, bwPacket):
"""!Called on FMS alarm
r"""!Called on FMS alarm
@param bwPacket: bwPacket instance
Remove if not implemented"""
@ -41,7 +41,7 @@ class BoswatchPlugin(PluginBase):
self._makeRequests(urls)
def pocsag(self, bwPacket):
"""!Called on POCSAG alarm
r"""!Called on POCSAG alarm
@param bwPacket: bwPacket instance
Remove if not implemented"""
@ -49,7 +49,7 @@ class BoswatchPlugin(PluginBase):
self._makeRequests(urls)
def zvei(self, bwPacket):
"""!Called on ZVEI alarm
r"""!Called on ZVEI alarm
@param bwPacket: bwPacket instance
Remove if not implemented"""
@ -57,7 +57,7 @@ class BoswatchPlugin(PluginBase):
self._makeRequests(urls)
def msg(self, bwPacket):
"""!Called on MSG packet
r"""!Called on MSG packet
@param bwPacket: bwPacket instance
Remove if not implemented"""

22
plugin/mysql.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -28,14 +28,14 @@ logging.debug("- %s loaded", __name__)
class BoswatchPlugin(PluginBase):
"""!Description of the Plugin"""
r"""!Description of the Plugin"""
def __init__(self, config):
"""!Do not change anything here!"""
r"""!Do not change anything here!"""
super().__init__(__name__, config) # you can access the config class on 'self.config'
def onLoad(self):
"""!Called by import of the plugin
r"""!Called by import of the plugin
Remove if not implemented"""
self.sqlInserts = {
"pocsag": "INSERT INTO boswatch (packetTimestamp, packetMode, pocsag_ric, pocsag_subric, pocsag_subricText, pocsag_message, pocsag_bitrate, serverName, serverVersion, serverBuildDate, serverBranch, clientName, clientIP, clientVersion, clientBuildDate, clientBranch, inputSource, frequency) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)",
@ -63,7 +63,7 @@ class BoswatchPlugin(PluginBase):
self.cursor.close()
def setup(self):
"""!Called before alarm
r"""!Called before alarm
Remove if not implemented"""
try:
self.connection.ping(reconnect=True, attempts=3, delay=2)
@ -74,7 +74,7 @@ class BoswatchPlugin(PluginBase):
self.cursor = self.connection.cursor()
def fms(self, bwPacket):
"""!Called on FMS alarm
r"""!Called on FMS alarm
@param bwPacket: bwPacket instance
Remove if not implemented"""
@ -105,7 +105,7 @@ class BoswatchPlugin(PluginBase):
self.cursor.execute(self.sqlInserts.get("fms"), val)
def pocsag(self, bwPacket):
"""!Called on POCSAG alarm
r"""!Called on POCSAG alarm
@param bwPacket: bwPacket instance
Remove if not implemented"""
@ -132,7 +132,7 @@ class BoswatchPlugin(PluginBase):
self.cursor.execute(self.sqlInserts.get("pocsag"), val)
def zvei(self, bwPacket):
"""!Called on ZVEI alarm
r"""!Called on ZVEI alarm
@param bwPacket: bwPacket instance
Remove if not implemented"""
@ -155,7 +155,7 @@ class BoswatchPlugin(PluginBase):
self.cursor.execute(self.sqlInserts.get("pocsag"), val)
def msg(self, bwPacket):
"""!Called on MSG packet
r"""!Called on MSG packet
@param bwPacket: bwPacket instance
Remove if not implemented"""
@ -177,12 +177,12 @@ class BoswatchPlugin(PluginBase):
self.cursor.execute(self.sqlInserts.get("msg"), val)
def teardown(self):
"""!Called after alarm
r"""!Called after alarm
Remove if not implemented"""
self.connection.commit()
self.cursor.close()
def onUnload(self):
"""!Called by destruction of the plugin
r"""!Called by destruction of the plugin
Remove if not implemented"""
self.connection.close()

30
plugin/pluginBase.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -24,12 +24,12 @@ logging.debug("- %s loaded", __name__)
class PluginBase(ABC):
"""!Main plugin class"""
r"""!Main plugin class"""
_pluginsActive = []
def __init__(self, pluginName, config):
"""!init preload some needed locals and then call onLoad() directly"""
r"""!init preload some needed locals and then call onLoad() directly"""
self._pluginName = pluginName
self.config = config
self._pluginsActive.append(self)
@ -54,13 +54,13 @@ class PluginBase(ABC):
self.onLoad()
def _cleanup(self):
"""!Cleanup routine calls onUnload() directly"""
r"""!Cleanup routine calls onUnload() directly"""
logging.debug("[%s] onUnload()", self._pluginName)
self._pluginsActive.remove(self)
self.onUnload()
def _run(self, bwPacket):
"""!start an complete running turn of an plugin.
r"""!start an complete running turn of an plugin.
Calls setup(), alarm() and teardown() in this order.
The alarm() method serves the BOSWatch packet to the plugin.
@ -121,7 +121,7 @@ class PluginBase(ABC):
return None
def _getStatistics(self):
"""!Returns statistical information's from last plugin run
r"""!Returns statistical information's from last plugin run
@return Statistics as pyton dict"""
stats = {"type": "plugin",
@ -137,55 +137,55 @@ class PluginBase(ABC):
return stats
def onLoad(self):
"""!Called by import of the plugin
r"""!Called by import of the plugin
can be inherited"""
pass
def setup(self):
"""!Called before alarm
r"""!Called before alarm
can be inherited"""
pass
def fms(self, bwPacket):
"""!Called on FMS alarm
r"""!Called on FMS alarm
can be inherited
@param bwPacket: bwPacket instance"""
logging.warning("ZVEI not implemented in %s", self._pluginName)
def pocsag(self, bwPacket):
"""!Called on POCSAG alarm
r"""!Called on POCSAG alarm
can be inherited
@param bwPacket: bwPacket instance"""
logging.warning("POCSAG not implemented in %s", self._pluginName)
def zvei(self, bwPacket):
"""!Called on ZVEI alarm
r"""!Called on ZVEI alarm
can be inherited
@param bwPacket: bwPacket instance"""
logging.warning("ZVEI not implemented in %s", self._pluginName)
def msg(self, bwPacket):
"""!Called on MSG packet
r"""!Called on MSG packet
can be inherited
@param bwPacket: bwPacket instance"""
logging.warning("MSG not implemented in %s", self._pluginName)
def teardown(self):
"""!Called after alarm
r"""!Called after alarm
can be inherited"""
pass
def onUnload(self):
"""!Called on shutdown of boswatch
r"""!Called on shutdown of boswatch
can be inherited"""
pass
def parseWildcards(self, msg):
"""!Return the message with parsed wildcards"""
r"""!Return the message with parsed wildcards"""
if self._bwPacket is None:
logging.warning("wildcard replacing not allowed - no bwPacket set")
return msg

16
plugin/telegram.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -51,14 +51,14 @@ class MQBot(telegram.bot.Bot):
class BoswatchPlugin(PluginBase):
"""!Description of the Plugin"""
r"""!Description of the Plugin"""
def __init__(self, config):
"""!Do not change anything here!"""
r"""!Do not change anything here!"""
super().__init__(__name__, config) # you can access the config class on 'self.config'
def onLoad(self):
"""!Called by import of the plugin"""
r"""!Called by import of the plugin"""
if self.config.get("queue", default=True):
q = mq.MessageQueue()
request = Request(con_pool_size=8)
@ -69,14 +69,14 @@ class BoswatchPlugin(PluginBase):
print('normal')
def fms(self, bwPacket):
"""!Called on FMS alarm
r"""!Called on FMS alarm
@param bwPacket: bwPacket instance"""
msg = self.parseWildcards(self.config.get("message_fms", default="{FMS}"))
self._sendMessage(msg)
def pocsag(self, bwPacket):
"""!Called on POCSAG alarm
r"""!Called on POCSAG alarm
@param bwPacket: bwPacket instance"""
msg = self.parseWildcards(self.config.get("message_pocsag", default="{RIC}({SRIC})\n{MSG}"))
@ -88,14 +88,14 @@ class BoswatchPlugin(PluginBase):
self._sendLocation(lat, lon)
def zvei(self, bwPacket):
"""!Called on ZVEI alarm
r"""!Called on ZVEI alarm
@param bwPacket: bwPacket instance"""
msg = self.parseWildcards(self.config.get("message_zvei", default="{TONE}"))
self._sendMessage(msg)
def msg(self, bwPacket):
"""!Called on MSG packet
r"""!Called on MSG packet
@param bwPacket: bwPacket instance"""
msg = self.parseWildcards(self.config.get("message_msg"))

22
plugin/template_plugin.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -26,55 +26,55 @@ logging.debug("- %s loaded", __name__)
class BoswatchPlugin(PluginBase):
"""!Description of the Plugin"""
r"""!Description of the Plugin"""
def __init__(self, config):
"""!Do not change anything here!"""
r"""!Do not change anything here!"""
super().__init__(__name__, config) # you can access the config class on 'self.config'
def onLoad(self):
"""!Called by import of the plugin
r"""!Called by import of the plugin
Remove if not implemented"""
pass
def setup(self):
"""!Called before alarm
r"""!Called before alarm
Remove if not implemented"""
pass
def fms(self, bwPacket):
"""!Called on FMS alarm
r"""!Called on FMS alarm
@param bwPacket: bwPacket instance
Remove if not implemented"""
pass
def pocsag(self, bwPacket):
"""!Called on POCSAG alarm
r"""!Called on POCSAG alarm
@param bwPacket: bwPacket instance
Remove if not implemented"""
pass
def zvei(self, bwPacket):
"""!Called on ZVEI alarm
r"""!Called on ZVEI alarm
@param bwPacket: bwPacket instance
Remove if not implemented"""
pass
def msg(self, bwPacket):
"""!Called on MSG packet
r"""!Called on MSG packet
@param bwPacket: bwPacket instance
Remove if not implemented"""
pass
def teardown(self):
"""!Called after alarm
r"""!Called after alarm
Remove if not implemented"""
pass
def onUnload(self):
"""!Called by destruction of the plugin
r"""!Called by destruction of the plugin
Remove if not implemented"""
pass

42
test/boswatch/test_ServerClient.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -32,13 +32,13 @@ def setup_function(function):
@pytest.fixture
def getClient():
"""!Build and serve a TCPCLient"""
r"""!Build and serve a TCPCLient"""
return TCPClient()
@pytest.fixture
def getServer():
"""!Build and serve a TCPServer"""
r"""!Build and serve a TCPServer"""
dataQueue = queue.Queue()
testServer = TCPServer(dataQueue)
return testServer
@ -46,7 +46,7 @@ def getServer():
@pytest.fixture
def getRunningServer(getServer):
"""!Build and serve a still running TCPServer"""
r"""!Build and serve a still running TCPServer"""
logging.debug("start server")
assert getServer.start()
while not getServer.isRunning:
@ -58,40 +58,40 @@ def getRunningServer(getServer):
def test_clientConnectFailed(getClient):
"""!Connect to a non available server"""
r"""!Connect to a non available server"""
assert not getClient.connect()
def test_clientDisconnectFailed(getClient):
"""!Disconnect while no connection is established"""
r"""!Disconnect while no connection is established"""
assert getClient.disconnect()
def test_clientTransmitFailed(getClient):
"""!Transmit while no connection is established"""
r"""!Transmit while no connection is established"""
assert not getClient.transmit("test")
def test_clientReceiveFailed(getClient):
"""!Receive while no connection is established"""
r"""!Receive while no connection is established"""
assert not getClient.receive()
def test_clientConnect(getClient, getRunningServer):
"""!Connect to a server"""
r"""!Connect to a server"""
assert getClient.connect()
assert getClient.disconnect()
def test_doubleConnect(getClient, getRunningServer):
"""!Connect to a server twice"""
r"""!Connect to a server twice"""
assert getClient.connect()
assert getClient.connect()
assert getClient.disconnect()
def test_clientReconnect(getClient, getRunningServer):
"""!Try a reconnect after a established connection"""
r"""!Try a reconnect after a established connection"""
assert getClient.connect()
assert getClient.disconnect()
assert getClient.connect()
@ -99,7 +99,7 @@ def test_clientReconnect(getClient, getRunningServer):
def test_clientMultiConnect(getClient, getRunningServer):
"""!Connect with 2 clients to the server"""
r"""!Connect with 2 clients to the server"""
assert getClient.connect()
testClient2 = TCPClient()
assert testClient2.connect()
@ -112,7 +112,7 @@ def test_clientMultiConnect(getClient, getRunningServer):
def test_clientCommunicate(getClient, getRunningServer):
"""!Try to send data to the server and check on '[ack]'"""
r"""!Try to send data to the server and check on '[ack]'"""
assert getClient.connect()
assert getClient.transmit("test")
assert getClient.receive() == "[ack]"
@ -121,7 +121,7 @@ def test_clientCommunicate(getClient, getRunningServer):
@pytest.mark.skip("needs fixture for more than one client")
def test_clientMultiCommunicate(getServer):
"""!Try to send data to the server with 3 clients and check on '[ack]'"""
r"""!Try to send data to the server with 3 clients and check on '[ack]'"""
# connect all
testClient1 = TCPClient()
assert testClient1.connect()
@ -146,26 +146,26 @@ def test_clientMultiCommunicate(getServer):
def test_serverRestart(getRunningServer):
"""!Test a stop and restart of the server"""
r"""!Test a stop and restart of the server"""
assert getRunningServer.stop()
assert getRunningServer.start()
assert getRunningServer.stop()
def test_serverStopFailed(getServer):
"""!Test to stop a stopped server"""
r"""!Test to stop a stopped server"""
assert getServer.stop()
def test_serverDoubleStart(getServer):
"""!Test to start the server twice"""
r"""!Test to start the server twice"""
assert getServer.start()
assert getServer.start()
assert getServer.stop()
def test_serverStartTwoInstances():
"""!Test to start two server different server instances"""
r"""!Test to start two server different server instances"""
dataQueue = queue.Queue()
testServer1 = TCPServer(dataQueue)
testServer2 = TCPServer(dataQueue)
@ -179,7 +179,7 @@ def test_serverStartTwoInstances():
def test_serverStopsWhileConnected(getRunningServer, getClient):
"""!Shutdown server while client is connected"""
r"""!Shutdown server while client is connected"""
getClient.connect()
getRunningServer.stop()
timeout = 5
@ -193,7 +193,7 @@ def test_serverStopsWhileConnected(getRunningServer, getClient):
@pytest.mark.skip("needs fixture for more than one client")
def test_serverGetOutput(getRunningServer):
"""!Send data to server with 2 clients, check '[ack]' and data on server queue"""
r"""!Send data to server with 2 clients, check '[ack]' and data on server queue"""
# connect all
testClient1 = TCPClient()
assert testClient1.connect()
@ -217,7 +217,7 @@ def test_serverGetOutput(getRunningServer):
def test_serverHighLoad(getRunningServer):
"""!High load server test with 10 send threads each will send 100 msg with 324 bytes size"""
r"""!High load server test with 10 send threads each will send 100 msg with 324 bytes size"""
logging.debug("start sendThreads")
threads = []
for thr_id in range(10):

16
test/boswatch/test_broadcast.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -29,7 +29,7 @@ def setup_function(function):
@pytest.fixture()
def broadcastServer():
"""!Server a BroadcastServer instance"""
r"""!Server a BroadcastServer instance"""
broadcastServer = BroadcastServer()
yield broadcastServer
if broadcastServer.isRunning:
@ -40,36 +40,36 @@ def broadcastServer():
@pytest.fixture()
def broadcastClient():
"""!Server a BroadcastClient instance"""
r"""!Server a BroadcastClient instance"""
return BroadcastClient()
def test_serverStartStop(broadcastServer):
"""!Start a BroadcastServer, check if running and stop it"""
r"""!Start a BroadcastServer, check if running and stop it"""
assert broadcastServer.start()
assert broadcastServer.isRunning
assert broadcastServer.stop()
def test_serverDoubleStart(broadcastServer):
"""!Try to start a BroadcastServer twice"""
r"""!Try to start a BroadcastServer twice"""
assert broadcastServer.start()
assert broadcastServer.start()
assert broadcastServer.stop()
def test_serverStopNotStarted(broadcastServer):
"""!Try to stop a BroadcastServer where is not running"""
r"""!Try to stop a BroadcastServer where is not running"""
assert broadcastServer.stop()
def test_clientWithoutServer(broadcastClient):
"""!Use BroadcastClient with no server"""
r"""!Use BroadcastClient with no server"""
assert not broadcastClient.getConnInfo(1)
def test_serverClientFetchConnInfo(broadcastClient, broadcastServer):
"""!Fetch connection info from BroadcastServer"""
r"""!Fetch connection info from BroadcastServer"""
assert broadcastServer.start()
assert broadcastClient.getConnInfo()
assert broadcastServer.stop()

24
test/boswatch/test_config.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -29,41 +29,41 @@ def setup_function(function):
@pytest.fixture
def getConfig():
"""!Build a config object"""
r"""!Build a config object"""
return ConfigYAML()
@pytest.fixture
def getFilledConfig():
"""!Build a config object and fill it with the config data"""
r"""!Build a config object and fill it with the config data"""
filledConfig = ConfigYAML()
assert filledConfig.loadConfigFile(paths.TEST_PATH + "test_config.yaml") is True
return filledConfig
def test_loadConfigFile(getConfig):
"""!load a config file"""
r"""!load a config file"""
assert getConfig.loadConfigFile(paths.TEST_PATH + "test_config.yaml") is True
def test_loadConfigFileFailed(getConfig):
"""!load a config file with syntax error"""
r"""!load a config file with syntax error"""
assert getConfig.loadConfigFile(paths.TEST_PATH + "test_configFailed.yaml") is False
def test_loadConfigFileNotFound(getConfig):
"""!load a config file where is not available"""
r"""!load a config file where is not available"""
assert getConfig.loadConfigFile(paths.TEST_PATH + "test_configNotFound.yaml") is False
def test_getConfigAsString(getFilledConfig):
"""!Get the string representation of the config"""
r"""!Get the string representation of the config"""
assert type(str(getFilledConfig)) is str
logging.debug(getFilledConfig)
def test_getTypes(getFilledConfig):
"""!Get and check different data types in config"""
r"""!Get and check different data types in config"""
assert type(getFilledConfig.get("types")) is ConfigYAML
assert type(getFilledConfig.get("types", "string")) is str
assert type(getFilledConfig.get("types", "bool")) is bool
@ -72,19 +72,19 @@ def test_getTypes(getFilledConfig):
def test_getDefaultValue(getFilledConfig):
"""!Get the default value of an not existent entry"""
r"""!Get the default value of an not existent entry"""
assert getFilledConfig.get("notExistent", default="defaultValue") == "defaultValue"
def test_getNestedConfig(getFilledConfig):
"""!Work with nested sub-config elements"""
r"""!Work with nested sub-config elements"""
nestedConfig = getFilledConfig.get("types")
assert type(nestedConfig) is ConfigYAML
assert nestedConfig.get("string") == "Hello World"
def test_configIterationList(getFilledConfig):
"""!Try to iterate over a list in the config"""
r"""!Try to iterate over a list in the config"""
counter = 0
for item in getFilledConfig.get("list"):
assert type(item) is str
@ -93,7 +93,7 @@ def test_configIterationList(getFilledConfig):
def test_configIterationListWithNestedList(getFilledConfig):
"""!Try to iterate over a list in the config where its elements are lists itself"""
r"""!Try to iterate over a list in the config where its elements are lists itself"""
listCnt = 0
strCnt = 0
for item in getFilledConfig.get("list1"):

20
test/boswatch/test_decoder.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -26,19 +26,19 @@ def setup_function(function):
def test_decoderNoData():
"""!Test a empty string"""
r"""!Test a empty string"""
assert Decoder.decode("") is None
def test_decoderZveiValid():
"""!Test valid ZVEI"""
r"""!Test valid ZVEI"""
assert not Decoder.decode("ZVEI1: 12345") is None
assert not Decoder.decode("ZVEI1: 12838") is None
assert not Decoder.decode("ZVEI1: 34675") is None
def test_decoderZveiDoubleTone():
"""!Test doubleTone included ZVEI"""
r"""!Test doubleTone included ZVEI"""
assert not Decoder.decode("ZVEI1: 6E789") is None
assert not Decoder.decode("ZVEI1: 975E7") is None
assert not Decoder.decode("ZVEI1: 2E87E") is None
@ -54,7 +54,7 @@ def test_decoderZveiInvalid():
def test_decoderPocsagValid():
"""!Test valid POCSAG"""
r"""!Test valid POCSAG"""
assert not Decoder.decode("POCSAG512: Address: 1000000 Function: 0") is None
assert not Decoder.decode("POCSAG512: Address: 1000001 Function: 1") is None
assert not Decoder.decode("POCSAG1200: Address: 1000002 Function: 2") is None
@ -62,7 +62,7 @@ def test_decoderPocsagValid():
def test_decoderPocsagText():
"""!Test POCSAG with text"""
r"""!Test POCSAG with text"""
assert not Decoder.decode("POCSAG512: Address: 1000000 Function: 0 Alpha: test") is None
assert not Decoder.decode("POCSAG512: Address: 1000001 Function: 1 Alpha: test") is None
assert not Decoder.decode("POCSAG1200: Address: 1000002 Function: 2 Alpha: test") is None
@ -70,7 +70,7 @@ def test_decoderPocsagText():
def test_decoderPocsagShortRic():
"""!Test short POCSAG"""
r"""!Test short POCSAG"""
assert not Decoder.decode("POCSAG512: Address: 3 Function: 0 Alpha: test") is None
assert not Decoder.decode("POCSAG512: Address: 33 Function: 0 Alpha: test") is None
assert not Decoder.decode("POCSAG1200: Address: 333 Function: 0 Alpha: test") is None
@ -81,14 +81,14 @@ def test_decoderPocsagShortRic():
def test_decoderPocsagInvalid():
"""!Test invalid POCSAG"""
r"""!Test invalid POCSAG"""
assert Decoder.decode("POCSAG512: Address: 333333F Function: 0 Alpha: invalid") is None
assert Decoder.decode("POCSAG512: Address: 333333F Function: 1 Alpha: invalid") is None
assert Decoder.decode("POCSAG512: Address: 3333333 Function: 4 Alpha: invalid") is None
def test_decoderFmsValid():
"""!Test valid FMS"""
r"""!Test valid FMS"""
assert not Decoder.decode("""FMS: 43f314170000 (9=Rotkreuz 3=Bayern 1 Ort 0x25=037FZG 7141Status 3=Einsatz Ab 0=FZG->LST 2=I (ohneNA,ohneSIGNAL)) CRC correct""") is None
assert not Decoder.decode("""FMS: 43f314170000 (9=Rotkreuz 3=Bayern 1 Ort 0x25=037FZG 7141Status 3=Einsatz Ab 1=LST->FZG 2=I (ohneNA,ohneSIGNAL)) CRC correct""") is None
assert not Decoder.decode("""FMS: 43f314170000 (9=Rotkreuz 3=Bayern 1 Ort 0x25=037FZG 7141Status 3=Einsatz Ab 0=FZG->LST 2=II (ohneNA,mit SIGNAL)) CRC correct""") is None
@ -97,7 +97,7 @@ def test_decoderFmsValid():
def test_decoderFmsInvalid():
"""!Test invalid FMS"""
r"""!Test invalid FMS"""
assert Decoder.decode("""FMS: 14170000 (9=Rotkreuz 3=Bayern 1 Ort 0x25=037FZG 7141Status 3=Einsatz Ab 1=LST->FZG 2=III(mit NA,ohneSIGNAL)) CRC correct""") is None
assert Decoder.decode("""FMS: 43f314170000 (9=Rotkreuz 3=Bayern 1 Ort 0x25=037FZG 7141Sta 3=Einsatz Ab 0=FZG->LST 2=IV (mit NA,mit SIGNAL)) CRC correct""") is None
assert Decoder.decode("""FMS: 14170000 (9=Rotkreuz 3=Bayern 1 Ort 0x25=037FZG 7141Status 3=Einsatz Ab 1=LST->FZG 2=III(mit NA,ohneSIGNAL)) CRC incorrect""") is None

6
test/boswatch/test_header.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -26,10 +26,10 @@ def setup_function(function):
def test_logoToLog():
"""!Test logo to log"""
r"""!Test logo to log"""
assert header.logoToLog()
def test_infoToLog():
"""!Test info to log"""
r"""!Test info to log"""
assert header.infoToLog()

14
test/boswatch/test_packet.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -28,33 +28,33 @@ def setup_function(function):
@pytest.fixture()
def buildPacket():
"""!Build a BOSWatch packet and serve it to each test"""
r"""!Build a BOSWatch packet and serve it to each test"""
return Packet()
def test_createPacket(buildPacket):
"""!Create a packet"""
r"""!Create a packet"""
assert buildPacket != ""
def test_copyPacket(buildPacket):
"""!Copy a packet to an new instance"""
r"""!Copy a packet to an new instance"""
bwCopyPacket = Packet(buildPacket.__str__())
assert bwCopyPacket != ""
def test_getPacketString(buildPacket):
"""!get the intern packet dict as string"""
r"""!get the intern packet dict as string"""
assert type(buildPacket.__str__()) is str
assert buildPacket.__str__() != ""
def test_getNotSetField(buildPacket):
"""!try to get a not set field"""
r"""!try to get a not set field"""
assert not buildPacket.get("testfield")
def test_setGetField(buildPacket):
"""!set and get a field"""
r"""!set and get a field"""
buildPacket.set("testField", "test")
assert buildPacket.get("testField") == "test"

10
test/boswatch/test_paths.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -27,23 +27,23 @@ def setup_function(function):
def test_fileExists():
"""!load a local config file"""
r"""!load a local config file"""
assert paths.fileExist("README.md")
def test_fileNotExists():
"""!load a local config file"""
r"""!load a local config file"""
assert not paths.fileExist("notFound.txt")
def test_makeDirNotExisting():
"""!load a local config file"""
r"""!load a local config file"""
assert paths.makeDirIfNotExist("UnItTeSt")
os.removedirs("UnItTeSt")
def test_makeDirExisting():
"""!load a local config file"""
r"""!load a local config file"""
paths.makeDirIfNotExist("UnItTeSt")
assert paths.makeDirIfNotExist("UnItTeSt")
os.removedirs("UnItTeSt")

24
test/boswatch/test_timer.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -28,12 +28,12 @@ def setup_function(function):
def testTargetFast():
"""!Fast worker thread"""
r"""!Fast worker thread"""
logging.debug("run testTargetFast")
def testTargetSlow():
"""!Slow worker thread"""
r"""!Slow worker thread"""
logging.debug("run testTargetSlow start")
time.sleep(0.51)
logging.debug("run testTargetSlow end")
@ -41,7 +41,7 @@ def testTargetSlow():
@pytest.fixture()
def useTimerFast():
"""!Server a RepeatedTimer instance with fast worker"""
r"""!Server a RepeatedTimer instance with fast worker"""
testTimer = RepeatedTimer(0.1, testTargetFast)
yield testTimer
if testTimer.isRunning:
@ -50,7 +50,7 @@ def useTimerFast():
@pytest.fixture()
def useTimerSlow():
"""!Server a RepeatedTimer instance slow worker"""
r"""!Server a RepeatedTimer instance slow worker"""
testTimer = RepeatedTimer(0.1, testTargetSlow)
yield testTimer
if testTimer.isRunning:
@ -58,32 +58,32 @@ def useTimerSlow():
def test_timerStartStop(useTimerFast):
"""!Try to start and stop a timer"""
r"""!Try to start and stop a timer"""
assert useTimerFast.start()
assert useTimerFast.stop()
def test_timerDoubleStart(useTimerFast):
"""!Try to start a timer twice"""
r"""!Try to start a timer twice"""
assert useTimerFast.start()
assert useTimerFast.start()
assert useTimerFast.stop()
def test_timerStopNotStarted(useTimerFast):
"""!Try to stop a timer where is not started"""
r"""!Try to stop a timer where is not started"""
assert useTimerFast.stop()
def test_timerIsRunning(useTimerFast):
"""!Check if a timer is running"""
r"""!Check if a timer is running"""
assert useTimerFast.start()
assert useTimerFast.isRunning
assert useTimerFast.stop()
def test_timerRun(useTimerFast):
"""!Run a timer and check overdue and lostEvents"""
r"""!Run a timer and check overdue and lostEvents"""
assert useTimerFast.start()
time.sleep(0.2)
assert useTimerFast.stop()
@ -92,7 +92,7 @@ def test_timerRun(useTimerFast):
def test_timerOverdue(useTimerSlow):
"""!Run a timer and check overdue and lostEvents"""
r"""!Run a timer and check overdue and lostEvents"""
assert useTimerSlow.start()
time.sleep(0.2)
assert useTimerSlow.stop()
@ -101,7 +101,7 @@ def test_timerOverdue(useTimerSlow):
def test_timerOverdueLong(useTimerSlow):
"""!Run a timer and check overdue and lostEvents"""
r"""!Run a timer and check overdue and lostEvents"""
assert useTimerSlow.start()
time.sleep(1)
assert useTimerSlow.stop()

14
test/module/test_descriptor.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
@ -31,7 +31,7 @@ def setup_method(method):
@pytest.fixture
def makeDescriptor():
"""!Build a descriptor object with loaded configuration"""
r"""!Build a descriptor object with loaded configuration"""
config = ConfigYAML()
assert config.loadConfigFile(paths.TEST_PATH + "test_config.yaml") is True
descriptor = Descriptor(config.get("descriptor_test"))
@ -40,33 +40,33 @@ def makeDescriptor():
@pytest.fixture
def makePacket():
"""!Build a BW Packet object"""
r"""!Build a BW Packet object"""
packet = Packet()
return packet
def test_descriptorFoundFirst(makeDescriptor, makePacket):
"""!Run descriptor on the first entry in list"""
r"""!Run descriptor on the first entry in list"""
makePacket.set("tone", "12345")
makePacket = makeDescriptor.doWork(makePacket)
assert makePacket.get("description") == "Test 12345"
def test_descriptorFoundSecond(makeDescriptor, makePacket):
"""!Run descriptor on the second entry in list"""
r"""!Run descriptor on the second entry in list"""
makePacket.set("tone", "23456")
makePacket = makeDescriptor.doWork(makePacket)
assert makePacket.get("description") == "Test 23456"
def test_descriptorNotFound(makeDescriptor, makePacket):
"""!Run descriptor no matching data found"""
r"""!Run descriptor no matching data found"""
makePacket.set("tone", "99999")
makePacket = makeDescriptor.doWork(makePacket)
assert makePacket.get("description") == "99999"
def test_descriptorScanFieldNotAvailable(makeDescriptor, makePacket):
"""!Run descriptor on a non existent scanField"""
r"""!Run descriptor on a non existent scanField"""
makePacket = makeDescriptor.doWork(makePacket)
assert makePacket.get("description") is None

2
test/test_template.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""!
r"""!
____ ____ ______ __ __ __ _____
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <