Implemented a high-pass filter for input data

lasp/device/lasp_device_common.py

@@ -2,6 +2,7 @@ __all__ = ['DaqChannel']
 import json, logging
 from dataclasses import dataclass, field
 from typing import List
+import numpy as np
 from dataclasses_json import dataclass_json
 from ..lasp_common import Qty, SIQtys
 
@@ -20,10 +21,21 @@ class DaqChannel:
     def __post_init__(self):
         # logging.debug(f'__post_init__({self.channel_name})')
         self._qty = SIQtys.default()
-        if len(self.channel_metadata) > 0:
+
+        # Whether a digital high-pass filter should be used on input data.
+        # Negative means disabled. A positive number corresponds to the cut-on
+        # frequency of the installed highpass filter.
+        self._highpass = -1.0
+        try:
             meta = json.loads(self.channel_metadata)
             if 'qty' in meta:
+                # The quantity itself is stored as a JSON string, in the JSON
+                # object called channel_metadata.
                 self._qty = Qty.from_json(meta['qty'])
+            if 'highpass' in meta:
+                self._highpass = meta['highpass']
+        except json.JSONDecodeError:
+            logging.debug('No JSON data found in DaqChannel {self.channel_name}')
 
     @property
     def qty(self):
@@ -32,15 +44,25 @@ class DaqChannel:
     @qty.setter
     def qty(self, newqty):
         self._qty = newqty
-        self._store('qty', newqty)
+        self._store('qty', newqty.to_json())
+
+    @property
+    def highpass(self):
+        return self._highpass
+
+    @highpass.setter
+    def highpass(self, newvalue: float):
+        newvalue = float(newvalue)
+        self._highpass = newvalue
+        self._store('highpass', newvalue)
 
     def _store(self, name, val):
-        if len(self.channel_metadata) > 0:
+        try:
             meta = json.loads(self.channel_metadata)
-        else:
+        except json.JSONDecodeError:
             meta = {}
 
-        meta[name] = val.to_json()
+        meta[name] = val
         self.channel_metadata = json.dumps(meta)
 
     def __eq__(self, other):
@@ -53,5 +75,6 @@ class DaqChannel:
                 self.range_index == other.range_index and 
                 self.ACCoupling_enabled == other.ACCoupling_enabled and 
                 self.IEPE_enabled == other.IEPE_enabled and 
-                self.channel_metadata == other.channel_metadata)
+                self.channel_metadata == other.channel_metadata and
+                np.isclose(self.highpass,other.highpass))
 

lasp/lasp_avstream.py

@@ -19,6 +19,8 @@ from .device import Daq, DaqChannel, DaqConfiguration, DeviceInfo
 from .lasp_atomic import Atomic
 from .lasp_common import AvType
 from .lasp_multiprocessingpatch import apply_patch
+from .filter import highpass
+from .wrappers import SosFilterBank
 
 apply_patch()
 
@@ -133,21 +135,33 @@ class AudioStream:
         self.daq = Daq(device, daqconfig)
         en_in_ch = daqconfig.getEnabledInChannels(include_monitor=True)
         en_out_ch = daqconfig.getEnabledOutChannels()
+
         if en_in_ch == 0 and en_out_ch == 0:
             raise RuntimeError('No enabled input / output channels')
 
         logging.debug('Ready to start device...')
-
         samplerate = self.daq.start(self.streamCallback)
+
+        # Create required Highpass filters for incoming data
+        self.hpfs = [None]*len(en_in_ch)
+        for i, ch in enumerate(en_in_ch):
+            # Simple filter with a single bank and one section
+            if ch.highpass > 0:
+                fb = SosFilterBank(1, 1)
+                hpf = highpass(samplerate, ch.highpass, Q=np.sqrt(2))
+                fb.setFilter(0, hpf)
+                self.hpfs[i] = fb
+
         self.streammetadata = StreamMetaData(
             fs=samplerate,
-            in_ch=daqconfig.getEnabledInChannels(),
-            out_ch=daqconfig.getEnabledOutChannels(),
+            in_ch=en_in_ch,
+            out_ch=en_out_ch,
             blocksize=self.daq.nFramesPerBlock,
             dtype=self.daq.getNumpyDataType(),
         )
         self.running = True
 
+
     def streamCallback(self, indata, outdata, nframes):
         """
         This is called (from a separate thread) for each block
@@ -155,7 +169,7 @@ class AudioStream:
         """
         if not self.running:
             return 1
-        self.aframectr += 1                
+        self.aframectr += 1
 
         # TODO: Fix this. This gives bug on Windows, the threading lock does
         # give a strange erro.
@@ -165,7 +179,29 @@ class AudioStream:
         except Exception as e:
             print(e)
 
-        rv = self.processCallback(self, indata, outdata)
+        if indata is not None:
+            indata_filtered = np.empty_like(indata)
+            nchannels = indata.shape[1]
+
+            for i in range(nchannels):
+                # Filter each channel to the optional high-pass, which could also
+                # be an empty filter
+                if self.hpfs[i] is not None:
+                    indata_float = indata[:, [i]].astype(np.float)
+                    filtered_ch_float = self.hpfs[i].filter_(
+                       indata_float 
+                    )
+
+                    indata_filtered[:, i] = filtered_ch_float.astype(
+                        self.streammetadata.dtype)[:, 0]
+                else:
+                    # One-to-one copy
+                    indata_filtered[:, i] = indata[:, i]
+        else:
+            indata_filtered = indata
+
+        # rv = self.processCallback(self, indata, outdata)
+        rv = self.processCallback(self, indata_filtered, outdata)
         if rv != 0:
             self.running <<= False
         return rv