diff --git a/__deprecated/__dep__pluginManager.py b/__deprecated/__dep__pluginManager.py deleted file mode 100644 index 471c9cc..0000000 --- a/__deprecated/__dep__pluginManager.py +++ /dev/null @@ -1,111 +0,0 @@ -#!/usr/bin/python -# -*- coding: utf-8 -*- -"""! - ____ ____ ______ __ __ __ _____ - / __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ / - / __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ < - / /_/ / /_/ /___/ /| |/ |/ / /_/ / /_/ /__/ / / / ___/ / -/_____/\____//____/ |__/|__/\__,_/\__/\___/_/ /_/ /____/ - German BOS Information Script - by Bastian Schroll - -@file: pluginManager.py -@date: 08.01.2018 -@author: Bastian Schroll -@description: Plugin manager class to load and call the plugin -@todo must be mostly refactored -""" -import logging -import os -import time -import importlib - -from boswatch import configYaml -from boswatch.utils import paths - -logging.debug("- %s loaded", __name__) - - -class PluginManager: - """!Plugin manager class to load, manage and call the plugin - - @todo refactor the class and add documentation""" - - def __init__(self): - """!init comment""" - self._config = configYaml.loadConfigSharepoint("serverConfig") - self._pluginList = [] - - def searchPluginDir(self): - logging.debug("search for plugin in: %s", paths.PLUGIN_PATH) - for name in os.listdir(paths.PLUGIN_PATH): - location = os.path.join(paths.PLUGIN_PATH, name) - - # Skip if Path.isdir() or no File DIR_NAME.py is found - if not os.path.isdir(location) or not name + ".py" in os.listdir(location): - continue - - pluginPriority = self._config["plugin"][name] - - if pluginPriority is None: - logging.warning("no entry in server config for plugin: %s", name) - continue - elif pluginPriority > 0: - self._pluginList.append({"pluginName": name, "pluginPriority": pluginPriority}) - logging.debug("[ENABLED ] %s [%3d]", name, pluginPriority) - elif pluginPriority <= 0: - logging.debug("[DISABLED] %s ", name) - - # sort pluginList on pluginPriority descending (High to Low) - self._pluginList.sort(key=lambda x: x['pluginPriority'], reverse=True) - - def importAllPlugins(self): - logging.debug("importing all plugin") - for item in self._pluginList: - importPlugin = self._importPlugin(item["pluginName"]) - if importPlugin: - item["pluginImport"] = importPlugin - - @staticmethod - def _importPlugin(pluginName): - logging.debug("import plugin: %s", pluginName) - try: - return importlib.import_module("plugin." + pluginName + "." + pluginName) - except: - logging.exception("error while loading plugin: %s", pluginName) - return False - - def loadAllPlugins(self): - logging.debug("loading all plugin") - for item in self._pluginList: - item["pluginObject"] = None # todo del or none ??? - item["pluginObject"] = item["pluginImport"].BoswatchPlugin() - - def runAllPlugins(self, bwPacket): - logging.info("ALARM - %0.3f sec. since radio reception", time.time() - bwPacket.get("timestamp")) - for item in self._pluginList: - item["pluginObject"]._run(bwPacket) - item["pluginStatistics"] = item["pluginObject"]._getStatistics() - self.printEndStats() - - def unloadAllPlugins(self): - logging.debug("unload all plugin") - for item in self._pluginList: - # todo del or None ??? - del item["pluginObject"] # delete plugin object to force __del__() running - - def printEndStats(self): - logging.debug("Plugin run statistics:") - logging.debug("Plugin | runs | tRUN | tCUM | tSET | tALA | tTRD | eSET | eALA | eTRD") - for item in self._pluginList: - logging.debug("- %-12s | %4d | %0.2f | %6.1f | %0.2f | %0.2f | %0.2f | %4d | %4d | %4d", - item["pluginName"], - item["pluginStatistics"]["runCount"], - item["pluginStatistics"]["sumTime"], - item["pluginStatistics"]["cumTime"], - item["pluginStatistics"]["setupTime"], - item["pluginStatistics"]["alarmTime"], - item["pluginStatistics"]["teardownTime"], - item["pluginStatistics"]["setupErrorCount"], - item["pluginStatistics"]["alarmErrorCount"], - item["pluginStatistics"]["teardownErrorCount"]) diff --git a/__deprecated/descriptor.py b/__deprecated/descriptor.py deleted file mode 100644 index af97495..0000000 --- a/__deprecated/descriptor.py +++ /dev/null @@ -1,111 +0,0 @@ -#!/usr/bin/python -# -*- coding: utf-8 -*- -"""! - ____ ____ ______ __ __ __ _____ - / __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ / - / __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ < - / /_/ / /_/ /___/ /| |/ |/ / /_/ / /_/ /__/ / / / ___/ / -/_____/\____//____/ |__/|__/\__,_/\__/\___/_/ /_/ /____/ - German BOS Information Script - by Bastian Schroll - -@file: descriptor.py -@date: 07.01.2018 -@author: Bastian Schroll -@description: Descriptor to load Descriptions from csv files -""" -import logging -import csv -import re - -from boswatch.utils import paths - -logging.debug("- %s loaded", __name__) - - -class Descriptor: - """!CSV Descriptor class to load specific descriptions from CSV files, manage and serve them""" - - def __init__(self): - """!Initialise a private list for the DescriptionList objects""" - self._lists = {} - - def loadDescription(self, csvType): - """!Build a new description list from DescriptionList class - - @param csvType: Name of the CSV file without `.csv` - @return True or False""" - bwDescriptionList = DescriptionList() - if bwDescriptionList.loadCSV(csvType): - self._lists[csvType] = bwDescriptionList - return True - return False - - def addDescriptions(self, bwPacket): - """!Add the short and long description to a bwPacket - - @param bwPacket: bwPacket instance to add descriptions - @return True or False""" - logging.debug("add descriptions to bwPacket") - try: - bwPacket.set("shortDescription", - self._lists[bwPacket.get("mode")].getShortDescription(bwPacket.get(bwPacket.get("mode")))) - - bwPacket.set("longDescription", - self._lists[bwPacket.get("mode")].getLongDescription(bwPacket.get(bwPacket.get("mode")))) - return True - except: # pragma: no cover - logging.exception("error while adding descriptions") - return False - - -class DescriptionList: - def __init__(self): - """!Loads the given CSV file into internal list""" - logging.debug("create new descriptionList") - self._descriptionList = {} - - def getShortDescription(self, checkId): - """!Returns the short description of given id - - @return short description or empty string""" - try: - return self._descriptionList[str(checkId)]["shortDescription"] - except: - return "" - - def getLongDescription(self, checkId): - """!Returns the long description of given id - - @return long description or empty string""" - try: - return self._descriptionList[str(checkId)]["longDescription"] - except: - return "" - - def loadCSV(self, csvType): - """!Load descriptions from an csv file - - @param csvType: Name of the CSV file without `.csv` - @return True or False""" - count = 0 - logging.debug("loading csv file: %s", csvType) - csvPath = paths.CSV_PATH + csvType + ".csv" - try: - csvFile = open(csvPath, 'r', -1, 'utf-8') - reader = csv.DictReader(csvFile) - - for line in reader: - if re.match("^[0-9]+[A-D]?$", line["id"], re.IGNORECASE): - self._descriptionList[line["id"]] = {"shortDescription": line["shortDescription"], "longDescription": line["longDescription"]} - logging.debug("- %s", line) - count += 1 - - logging.debug("%s entry's loaded", count) - return True - except FileNotFoundError: - logging.error("csv file not found: %s", csvPath) - return False - except: # pragma: no cover - logging.exception("error while loading descriptions") - return False diff --git a/__deprecated/fms.csv b/__deprecated/fms.csv deleted file mode 100644 index f077c98..0000000 --- a/__deprecated/fms.csv +++ /dev/null @@ -1,12 +0,0 @@ -id,shortDescription,longDescription -# -# BOSWatch CSV file for describing FMS-Addresses -# -# For each FMS-Address you could set a description-text -# Use the structure: fms,"Description-Text" -# -# !!! DO NOT delete the first line !!! -# -12345678,"FMS testdata äöüß" -56487234,"Test Nummer 18","" -68734123,"Und noch einer ganz schnell","" \ No newline at end of file diff --git a/__deprecated/pocsag.csv b/__deprecated/pocsag.csv deleted file mode 100644 index 1b9f59a..0000000 --- a/__deprecated/pocsag.csv +++ /dev/null @@ -1,18 +0,0 @@ -id,shortDescription,longDescription -# -# BOSWatch CSV file for describing POCSAG-Addresses -# -# For each RIC-Address you could set a description-text -# Use the structure: ric,"Description-Text" -# -# You can even define specific subrics, therefore you -# 1. need to specify a main RIC: 1234567, "Unit One" -# 2. specify a certain subric: 1234567B, "Subunit Bravo" -# The result for 1234567B will be "Unit One Subunit Bravo" -# - Be sure having defined the main RIC (step one)! - -# -# !!! DO NOT delete the first line !!! -# -1234567,"POCSAG testdata äöüß" -4323424,"Test Nummer 18","" -6873423,"Und noch einer ganz schnell","" diff --git a/__deprecated/test_descriptor.py b/__deprecated/test_descriptor.py deleted file mode 100644 index 3b1e0b8..0000000 --- a/__deprecated/test_descriptor.py +++ /dev/null @@ -1,72 +0,0 @@ -#!/usr/bin/python -# -*- coding: utf-8 -*- -"""! - ____ ____ ______ __ __ __ _____ - / __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ / - / __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ < - / /_/ / /_/ /___/ /| |/ |/ / /_/ / /_/ /__/ / / / ___/ / -/_____/\____//____/ |__/|__/\__,_/\__/\___/_/ /_/ /____/ - German BOS Information Script - by Bastian Schroll - -@file: test_descriptor.py -@date: 07.01.2017 -@author: Bastian Schroll -@description: Unittests for BOSWatch. File have to run as "pytest" unittest -""" -import logging -import pytest - -# from module.descriptor import Descriptor -# from module.descriptor import DescriptionList -from boswatch.packet import Packet - - -@pytest.mark.skip -class Test_Descriptor: - """!Unittests for the descriptor""" - - def setup_method(self, method): - logging.debug("[TEST] %s.%s", method.__module__, method.__name__) - - def test_loadCsvNotExist(self): - """!read CSV file where not exist direct per DescriptionList class""" - descList = DescriptionList() - assert not descList.loadCSV("boswatch") - - def test_loadCsv(self): - """!read CSV file direct per DescriptionList class""" - descList = DescriptionList() - assert descList.loadCSV("zvei") - - def test_descriptorLoadFailed(self): - """!read CSV file where not exist""" - bwDescriptor = Descriptor() - assert not bwDescriptor.loadDescription("boswatch") - - def test_descriptorLoad(self): - """!read CSV file""" - bwDescriptor = Descriptor() - assert bwDescriptor.loadDescription("zvei") - - def test_loadDescriptionsNotSet(self): - """!load descriptions where not set to an bwPacket""" - bwDescriptor = Descriptor() - assert bwDescriptor.loadDescription("zvei") - bwPacket = Packet() - bwPacket.set("mode", "zvei") - bwPacket.set("zvei", "54321") - assert bwDescriptor.addDescriptions(bwPacket) - assert bwPacket.get("shortDescription") is "" - assert bwPacket.get("longDescription") is "" - - def test_loadDescriptions(self): - """!load descriptions to an bwPacket""" - bwDescriptor = Descriptor() - assert bwDescriptor.loadDescription("zvei") - bwPacket = Packet() - bwPacket.set("mode", "zvei") - bwPacket.set("zvei", "12345") - assert bwDescriptor.addDescriptions(bwPacket) - assert bwPacket.get("shortDescription") is not "" - assert bwPacket.get("longDescription") is not "" diff --git a/__deprecated/zvei.csv b/__deprecated/zvei.csv deleted file mode 100644 index ea596dc..0000000 --- a/__deprecated/zvei.csv +++ /dev/null @@ -1,12 +0,0 @@ -id,shortDescription,longDescription -# -# BOSWatch CSV file for describing ZVEI-Addresses -# -# For each ZVEI-Address you could set a description-text -# Use the structure: zvei,"Description-Text" -# -# !!! DO NOT delete the first line !!! -# -12345,"ZVEI testdata äöüß","Hier nochmal eine echt Lange Info die keinen Sinn macht" -56487,"Test Nummer 18","" -68734,"Und noch einer ganz schnell",""