add filter middleware

box-pc/plc-connector/inputs/replay_influxdb.py

@@ -3,6 +3,15 @@ from datetime import datetime, timedelta
 from influxdb_client import InfluxDBClient
 
 from inputs.common import Input
+from structures.measurement import Measurement24v, Measurement480v
+from structures.plant import CompactLogixState, S7State
+
+dataclasses = [
+  Measurement24v,
+  Measurement480v,
+  CompactLogixState,
+  S7State
+]
 
 logger = logging.getLogger(__name__)
 
@@ -15,7 +24,6 @@ class Replay(Input):
 
     self.query_api = self.client.query_api()
     self.current_time = datetime.strptime(start_time, "%d.%m.%Y %H:%M:%S")
-
   
   def read_handler(self):
     start = self.current_time
@@ -31,12 +39,16 @@ class Replay(Input):
     result = self.query_api.query(query=query)
     results = []
     for table in result:
+      for dataclass in dataclasses:
+        if table in dataclass.__name__.lower():
+          break
+
       for record in table.records:
-        res = {
-          'series': table,
-          'timestamp': record.get_time(),
-        }
-        res.extend(record.values)
+        res = dataclass(
+          series = table,
+          timestamp = record.get_time(),
+          **record.values
+        )
         results.append(res)
     return results
 

box-pc/plc-connector/middlewares/common.py

@@ -0,0 +1,10 @@
+
+class MatchSeries:
+  def __init__(self, series) -> None:
+    self._series = series
+  
+  def get_series(self, measurement):
+    if getattr(measurement, 'series', None) == self._series:
+      return measurement
+    else:
+      return getattr(measurement, self._series, None)

box-pc/plc-connector/middlewares/filter.py

@@ -0,0 +1,27 @@
+
+from .common import MatchSeries
+
+class Filter(MatchSeries):
+  def __init__(self, field = None, value = None, **kwargs) -> None:
+    super().__init__(**kwargs)
+    self._field = field
+    self._value = value
+
+  def execute(self, values):
+    for measurement in values:
+      dataset = self.get_series(measurement)
+
+      if not dataset:
+        continue
+
+      if self._field:
+        value = getattr(dataset, self._field, None)
+        if not value:
+          continue
+        if self._value and self._value != value:
+          continue
+      
+      yield measurement
+      
+      
+