Merge pull request #305 from JohnGriffiths/add_kf_device

Add kf device

Author: JohnGriffiths

eegnb/devices/eeg.py

@@ -9,6 +9,7 @@
 import time
 import logging
 from time import sleep
+from datetime import datetime
 from multiprocessing import Process
 
 import numpy as np
@@ -26,6 +27,9 @@
     EEG_CHANNELS,
 )
 
+import socket, json, logging, struct
+from time import time
+
 
 logger = logging.getLogger(__name__)
 
@@ -54,6 +58,7 @@
 ]
 
 
+
 class EEG:
     device_name: str
     stream_started: bool = False
@@ -66,8 +71,8 @@ def __init__(
         mac_addr=None,
         other=None,
         ip_addr=None,
-        ch_names=None
-    ):
+        ch_names=None,
+        make_logfile=False):
         """The initialization function takes the name of the EEG device and determines whether or not
         the device belongs to the Muse or Brainflow families and initializes the appropriate backend.
 
@@ -84,6 +89,7 @@ def __init__(
         self.mac_address = mac_addr
         self.ip_addr = ip_addr
         self.other = other
+        self.make_logfile = make_logfile # currently only used for kf
         self.backend = self._get_backend(self.device_name)
         self.initialize_backend()
         self.n_channels = len(EEG_INDICES[self.device_name])
@@ -92,23 +98,30 @@ def __init__(
         self.ch_names = ch_names
 
     def initialize_backend(self):
+        # run this at initialization to get some
+        # stream metadata into the eeg class
         if self.backend == "brainflow":
             self._init_brainflow()
             self.timestamp_channel = BoardShim.get_timestamp_channel(self.brainflow_id)
         elif self.backend == "muselsl":
             self._init_muselsl()
-            self._muse_get_recent()  # run this at initialization to get some
-            # stream metadata into the eeg class
+            self._muse_get_recent()
+        elif self.backend == "kernelflow":
+            self._init_kf()
 
     def _get_backend(self, device_name):
         if device_name in brainflow_devices:
             return "brainflow"
         elif device_name in ["muse2016", "muse2", "museS"]:
             return "muselsl"
+        elif device_name in ["kernelflow"]:
+            return "kernelflow"
+
 
     #####################
     #   MUSE functions  #
     #####################
+
     def _init_muselsl(self):
         # Currently there's nothing we need to do here. However keeping the
         # option open to add things with this init method.
@@ -186,9 +199,11 @@ def _muse_get_recent(self, n_samples: int = 256, restart_inlet: bool = False):
         df = pd.DataFrame(samples, index=timestamps, columns=ch_names)
         return df
 
+
     ##########################
     #   BrainFlow functions  #
     ##########################
+
     def _init_brainflow(self):
         """This function initializes the brainflow backend based on the input device name. It calls
         a utility function to determine the appropriate USB port to use based on the current operating system.
@@ -398,10 +413,111 @@ def _brainflow_get_recent(self, n_samples=256):
         # print (df)
         return df
 
+
+
+    ###########################
+    #   Kernel Flow functions #
+    ###########################
+
+
+    def _init_kf(self):
+        
+        self._notes = None #muse_recent_inlet = None
+
+        # Grab the init time for tracking
+        dtstr = str(datetime.now()).replace(' ', '_').split('.')[0].replace(':', '-') 
+
+        # Initiate connection to trigger-recording port
+        host = 'localhost'  # could be another computer on network with a visible IP address?
+        port = 6767
+        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        sock.connect((host, port))
+        self.kf_host = host
+        self.kf_port = port
+        self.kf_socket = sock
+
+        # Triggers history list
+        self.kf_triggers_history = []
+        self.kf_triggers_history.append('Initiated connection: %s' %dtstr)
+
+        # Optionally maintain a logfile
+        if self.make_logfile:
+            self.kf_logfile_fname = 'eegexpy_kf_logfile__%s.txt' % dtstr
+            self.kf_logfile_handle = open(self.kf_logfile_fname, 'a')
+            self.kf_logfile_handle.write('KF TRIGGERS LOGFILE %s \n\n' %dtstr)
+             
+
+    def _start_kf(self): #:, duration):
+
+        kf_start_timestamp = int(time()*1e6)
+
+        # Send first data packet 
+        self.kf_evnum = 0
+        data_to_send = {"id": self.kf_evnum,
+                        "timestamp": kf_start_timestamp,
+                        "event": "start_experiment",
+                        "value": "0"
+                        }
+        self._kf_sendeventinfo(data_to_send)
+        
+        # Update logfile text
+        self.kf_triggers_history.append({'kf_evnum': '%s' %self.kf_evnum,
+                                         'kf_start_timestamp': kf_start_timestamp,
+                                         'packet_sent': data_to_send})
+
+    
+    def _kf_push_sample(self, timestamp, marker, marker_name):
+   
+        # Send trigger
+        self.kf_evnum+=1
+        kf_trigger_timestamp = int(time()*1e6)
+        data_to_send = {
+                         "id": self.kf_evnum, #event_id,
+                         "timestamp": kf_trigger_timestamp, # timestamp
+                         "event": 'start_trial', #marker_name, #event_name,
+                         "value": str(marker), #str(marker_name), 
+                        }
+        self._kf_sendeventinfo(data_to_send)
+        
+        # Update logfile text
+        self.kf_triggers_history.append({'kf_evnum': '%s' %self.kf_evnum,
+                                         'kf_trigger_timestamp': kf_trigger_timestamp,
+                                         'experiment_timestamp': timestamp,
+                                         'packet_sent': data_to_send})
+    
+    
+    def _stop_kf(self):
+        
+        self.kf_evnum+=1
+        kf_stop_timestamp = int(time()*1e6)
+        data_to_send = {
+                        "id": 3, #self.kf_evnum,
+                        "timestamp": kf_stop_timestamp,
+                        "event": "end_experiment",
+                        "value": "5"
+                        }
+        self._kf_sendeventinfo(data_to_send)
+
+        self.kf_triggers_history.append({'kf_evnum': '%s' % self.kf_evnum,
+                                         'kf_stop_timestamp': kf_stop_timestamp,
+                                         'packet_sent': data_to_send})
+        
+        if self.make_logfile:
+            self.kf_logfile_handle.write(self.kf_triggers_history)
+            self.kf_logfile_handle.close()
+
+
+    def _kf_sendeventinfo(self, event_info):
+
+        event_info_pack = json.dumps(event_info).encode("utf-8")
+        msg = struct.pack("!I", len(event_info_pack)) + event_info_pack
+        self.kf_socket.sendall(msg)
+    
+    
     #################################
     #   Highlevel device functions  #
     #################################
-
+    
     def start(self, fn, duration=None):
         """Starts the EEG device based on the defined backend.
 
@@ -416,8 +532,10 @@ def start(self, fn, duration=None):
             self.markers = []
         elif self.backend == "muselsl":
             self._start_muse(duration)
+        elif self.backend == "kernelflow":
+            self._start_kf()
 
-    def push_sample(self, marker, timestamp):
+    def push_sample(self, marker, timestamp, marker_name=None):
         """
         Universal method for pushing a marker and its timestamp to store alongside the EEG data.
 
@@ -429,12 +547,16 @@ def push_sample(self, marker, timestamp):
             self._brainflow_push_sample(marker=marker)
         elif self.backend == "muselsl":
             self._muse_push_sample(marker=marker, timestamp=timestamp)
+        elif self.backend == "kernelflow":
+           self._kf_push_sample(marker=marker,timestamp=timestamp, marker_name=marker_name)
 
     def stop(self):
         if self.backend == "brainflow":
             self._stop_brainflow()
         elif self.backend == "muselsl":
             pass
+        elif self.backend == "kernelflow":
+          self._stop_kf()
 
     def get_recent(self, n_samples: int = 256):
         """

eegnb/devices/utils.py

@@ -27,6 +27,7 @@
     "notion2": BoardShim.get_eeg_names(BoardIds.NOTION_2_BOARD.value),
     "crown": BoardShim.get_eeg_names(BoardIds.CROWN_BOARD.value),
     "freeeeg32": [f"eeg_{i}" for i in range(0, 32)],
+    "kernelflow": []
 }
 
 BRAINFLOW_CHANNELS = {
@@ -56,7 +57,8 @@
     "notion2": BoardShim.get_eeg_channels(BoardIds.NOTION_2_BOARD.value),
     "crown": BoardShim.get_eeg_channels(BoardIds.CROWN_BOARD.value),
     "freeeeg32": BoardShim.get_eeg_channels(BoardIds.FREEEEG32_BOARD.value),
-}
+    "kernelflow": [],
+    }
 
 SAMPLE_FREQS = {
     "muse2016": 256,
@@ -78,7 +80,8 @@
     "notion2": BoardShim.get_sampling_rate(BoardIds.NOTION_2_BOARD.value),
     "crown": BoardShim.get_sampling_rate(BoardIds.CROWN_BOARD.value),
     "freeeeg32": BoardShim.get_sampling_rate(BoardIds.FREEEEG32_BOARD.value),
-}
+    "kernelflow": [],
+    }
 
 
 def create_stim_array(timestamps, markers):