BOSWatch/BW3-Core
1
0
Fork 0
mirror of https://github.com/BOSWatch/BW3-Core.git synced 2026-01-09 18:20:04 +01:00
Code Issues Actions 4 Packages Projects Releases Wiki Activity

first draft doubleFilter

Browse source
This commit is contained in:
B-Watch 2020-07-09 21:17:44 +02:00
parent 6b0a7822e9
commit 8b5a538c5f
1 changed files with 37 additions and 20 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

57
module/filter/___doubleFilter.py → module/filter/doubleFilter.py
View file

@ -10,33 +10,45 @@
by Bastian Schroll
@file: doubleFilter.py
@date: 15.01.2018
@author: Bastian Schroll
@description: Class to implement a filter for double alarms
@todo test, refactor and document / check_msg is not implemented yet
@date: 09.07.2020
@author: Bastian Schroll, b-watch
@description: Filter module for double packages
"""
import logging
from module.moduleBase import ModuleBase
# ###################### #
# Custom plugin includes #
import time
# ######################
logging.debug("- %s loaded", __name__)
class DoubleFilter:
"""!Double Filter Class"""
class BoswatchModule(ModuleBase):
"""!Description of the Module"""
def __init__(self, config):
"""!init"""
self._config = config
"""!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 filter(self, bwPacket):
def onLoad(self):
"""!Called by import of the plugin
Remove if not implemented"""
pass
def doWork(self, bwPacket):
"""!start an run of the module.
@param bwPacket: A BOSWatch packet instance"""
if bwPacket.get("mode") == "fms":
scanWord = "fms"
filterFields = ["fms"]
elif bwPacket.get("mode") == "pocsag":
scanWord = "ric"
filterFields = self.config.get("pocsagFields", default=["ric"])
elif bwPacket.get("mode") == "zvei":
scanWord = "zvei"
filterFields = ["zvei"]
else:
logging.error("No Filter for '%s'", bwPacket)
return False
@ -45,30 +57,35 @@ class DoubleFilter:
logging.debug("create new doubleFilter list for '%s'", bwPacket.get("mode"))
self._filterLists[bwPacket.get("mode")] = []
logging.debug("scanWord for '%s' is '%s'", bwPacket.get("mode"), scanWord)
logging.debug("filterFields for '%s' is '%s'", bwPacket.get("mode"), ", ".join(filterFields))
return self._check(bwPacket, scanWord)
return self._check(bwPacket, filterFields)
def _check(self, bwPacket, scanWord):
def onUnload(self):
"""!Called by destruction of the plugin
Remove if not implemented"""
pass
def _check(self, bwPacket, filterFields):
self._filterLists[bwPacket.get("mode")].insert(0, bwPacket)
# delete entries that are to old
counter = 0
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["ignoreTime"]):
if float(listPacket.get("timestamp")) < (time.time() - self.config.get("ignoreTime", default=10)):
self._filterLists[bwPacket.get("mode")].remove(listPacket)
counter += 1
if counter:
logging.debug("%d old entry(s) removed", counter)
# delete last entry if list is to big
if len(self._filterLists[bwPacket.get("mode")]) > self._config["maxEntry"]:
if len(self._filterLists[bwPacket.get("mode")]) > self.config.get("maxEntry", default=20):
logging.debug("MaxEntry reached - delete oldest")
self._filterLists[bwPacket.get("mode")].pop()
for listPacket in self._filterLists[bwPacket.get("mode")][1:]: # [1:] skip first entry, thats the new one
if listPacket.get(scanWord) is bwPacket.get(scanWord):
logging.debug("found duplicate: %s", bwPacket.get(scanWord))
if all(listPacket.get(x) == bwPacket.get(x) for x in filterFields):
logging.debug("found duplicate: %s", bwPacket.get("mode"))
return False
logging.debug("doubleFilter ok")