Threaded h5 template

.gitignore

@@ -1,3 +1,7 @@
+# CAEN stuff
+CAENBoard_GENERICLog.txt  
+CAENParametersLog.txt
+
 # Byte-compiled / optimized / DLL files
 __pycache__/
 *.py[cod]

CAENBoard_GENERICLog.txt

@@ -0,0 +1,2 @@
+[       0.009][EE][CAENBoard_GENERIC.c::2526]: Open digitizer failed due to a comm error: -1
+[       0.009][EE][CAENDPPLayer.c::118]: From c_newDigitizer: Error Calling CAENDigitizer API

configs/debug.conf

@@ -6,4 +6,10 @@ con_type         = 'A4818'
 link_num         = 0
 conet_node       = 0
 vme_base_address = 0
-dig_authority    = 'caen.internal'
+dig_authority    = 'caen.internal'
+
+[output_settings]
+
+file_name = 'output_file'   # this will always be appended with a timestamp and .h5
+overwrite = True            # overwrite the file if you want
+

configs/recording/five_ns_window.conf

@@ -4,6 +4,7 @@ record_length  = 4096  # ns
 pre_trigger    = 512   # ns
 trigger_mode   = 'SELFTRIG' # look into the differing methods, trigger on channel based on threshold is an option
                           # SWTRIG, SELFTRIG, (not yet implemented) <EXTTRIG>
+software_timeout = 0      # hard software timeout between each digitiser poll (s)
 
 [channel_settings]
 
@@ -16,5 +17,39 @@ ch1 =   {'enabled'     : False,
          'self_trigger': False,
          'threshold'   : 0,
          'polarity'    : 'positive'}
+
+ch2 =   {'enabled'     : True,
+         'self_trigger': True,
+         'threshold'   : 600,
+         'polarity'    : 'positive'}
+
+ch3 =   {'enabled'     : False,
+         'self_trigger': False,
+         'threshold'   : 0,
+         'polarity'    : 'positive'}
+
+ch4 =   {'enabled'     : False,
+         'self_trigger': False,
+         'threshold'   : 0,
+         'polarity'    : 'positive'}
+
+ch5 =   {'enabled'     : False,
+         'self_trigger': False,
+         'threshold'   : 0,
+         'polarity'    : 'positive'}
+
+ch6 =   {'enabled'     : False,
+         'self_trigger': False,
+         'threshold'   : 0,
+         'polarity'    : 'positive'}
+
+ch7 =   {'enabled'     : False,
+         'self_trigger': False,
+         'threshold'   : 0,
+         'polarity'    : 'positive'}
 
+[output_settings]
 
+file_name = 'data/output_file'    # this will always be appended with a timestamp and .h5
+overwrite = True                  # overwrite the file if you want
+h5_flush_size = 20                # number of values added to h5 files per write

core/controller.py

@@ -1,6 +1,7 @@
 import numpy as np
 import logging
 import time
+from datetime import datetime
 #from caen_felib import lib, device, error
 from typing import Optional
 
@@ -25,6 +26,7 @@
 from core.logging import setup_logging
 from core.commands import CommandType, Command
 from core.worker import AcquisitionWorker
+from core.writer import Writer
 from core.tracker import Tracker
 from felib.digitiser import Digitiser
 from ui import oscilloscope
@@ -48,26 +50,51 @@ def __init__(self,
         # Digitiser configuration
         self.dig_config = dig_config
         self.rec_config = rec_config
+        self.dig_dict = read_config_file(self.dig_config)
+        self.rec_dict = read_config_file(self.rec_config)
+
+        # initialise a universal event counter for sanity purposes
+        self.event_counter = 0
 
         # Thread-safe communication channels
         self.cmd_buffer = Queue(maxsize=10)
         self.display_buffer = Queue(maxsize=1024)
-        self.stop_event = Event()
+        self.worker_stop_event = Event()
+        self.writer_stop_event = Event()
+        self.recording = False
 
         # Acquisition worker
+        self.sw_timeout = self.rec_dict['software_timeout']
         self.worker = AcquisitionWorker(
             cmd_buffer=self.cmd_buffer,
             display_buffer=self.display_buffer,
-            stop_event=self.stop_event,
+            stop_event=self.worker_stop_event,
+            sw_timeout = self.sw_timeout
         )
 
         # Set the callback to the controller's data_handling method
         self.worker.data_ready_callback = self.data_handling
 
-        # Start thread and log
+        # Start acquisition worker thread and log
         self.worker.start()
         logging.info("Acquisition worker thread started.")
 
+        # Multi channel writes to h5 
+        self.ch_mapping = self.get_ch_mapping()
+        self.num_ch = len(self.ch_mapping)
+        self.h5_flush_size = self.rec_dict['h5_flush_size']
+        self.writer_buffers = [Queue(maxsize=1024) for _ in range(self.num_ch)]
+        self.writers = [Writer(
+                            ch=curr_ch,
+                            flush_size=self.h5_flush_size,
+                            write_buffer=self.writer_buffers[i],
+                            stop_event=self.writer_stop_event,
+                            rec_config = read_config_file(self.rec_config),
+                            dig_config = read_config_file(self.dig_config),
+                            TIMESTAMP = datetime.now().strftime("%H:%M:%S")
+                        )
+                        for curr_ch, i in self.ch_mapping.items()]
+
         # gui second
         self.app = QApplication([])
         self.main_window = oscilloscope.MainWindow(controller = self)
@@ -78,6 +105,20 @@ def __init__(self,
 
         self.connect_digitiser()
 
+    def get_ch_mapping(self):
+        '''
+        Extract what channels are being used map them: ch -> index
+        '''
+        mapping = {}
+        i = 0
+        for entry in self.rec_dict:
+            if 'ch' in entry:
+                if self.rec_dict[entry]['enabled']:
+                    ch = int(entry[2:])
+                    mapping[ch] = i
+                    i += 1
+
+        return mapping
 
     def data_handling(self):
         '''
@@ -91,14 +132,26 @@ def data_handling(self):
                 break
 
             try:
-                wf_size, ADCs = data
+                # you must pass wf_size and ADCs through. 
+                wf_size, ADCs, ch = data
 
                 # update visuals
                 self.main_window.screen.update_ch(np.arange(0, wf_size, dtype=wf_size.dtype), ADCs)
 
                 # ping the tracker (make this optional)
                 self.tracker.track(ADCs.nbytes)
 
+                # push data to writer buffer 
+                if self.recording:
+                    write_data = wf_size, ADCs, self.event_counter
+
+                    # multi channel writing
+                    ch = int(ch)
+                    i = int(self.ch_mapping[ch])
+                    self.writer_buffers[i].put(write_data)
+
+                self.event_counter += 1
+
             except Exception as e:
                 logging.exception(f"Error updating display: {e}")
 
@@ -148,15 +201,59 @@ def stop_acquisition(self):
         logging.info("Stopping acquisition.")
         self.cmd_buffer.put(Command(CommandType.STOP))
 
+    def start_recording(self):
+        '''
+        Start recording data.
+        '''
+        self.recording = True
+        for w in self.writers:
+            if not w.is_alive():
+                w.start()
+                logging.info(f"Writer (channel {w.ch}) thread started.")
+
+        logging.info("Starting recording.")
+
+    def stop_recording(self):
+        '''
+        Stop recording data.
+        '''
+        self.recording = False
+        self.writer_stop_event.set()
+
+        for w in self.writers:
+            w.join(timeout=2) 
+
+        logging.info("Stopping recording.")
+
     def shutdown(self):
         '''
         Carefully shut down acquisition and worker thread.
         '''
         logging.info("Shutting down controller.")
+
+        # Acquisition Worker thread
         self.cmd_buffer.put(Command(CommandType.EXIT))
-        self.stop_event.set()
+        self.worker_stop_event.set()
         self.worker.join(timeout=2)
+
+        # Writer threads
+        self.writer_stop_event.set()
+        for w in self.writers:
+            w.join(timeout=2) 
+
+        clean_shutdown = True
+
         if self.worker.is_alive():
+            clean_shutdown = False
             logging.warning("AcquisitionWorker did not stop cleanly.")
-        else:
+
+        for w in self.writers:
+            if w.is_alive():
+                clean_shutdown = False
+                logging.warning(f"Writer (channel {w.ch}) did not stop cleanly.")
+
+        if clean_shutdown:
             logging.info("Controller shutdown complete.")
+
+        else:
+            logging.info("Controller shutdown failed.")

core/df_classes.py

@@ -0,0 +1,31 @@
+import tables as tb
+from typing import Type
+
+
+class config_class(tb.IsDescription):
+     '''
+     Holds dictionary (key, value) pairs
+     '''
+     key   = tb.StringCol(90) # 90 character string maximum, you've been warned!
+     value = tb.StringCol(90)
+
+
+
+def return_rwf_class(WD_version : str, shape : int) -> Type[tb.IsDescription]:
+    '''
+    Based on MULE shapes, expect output to be formatted as such, for forwards compatibility.
+    '''
+    if WD_version == 1:
+        class rwf_df(tb.IsDescription):
+            evt_no    = tb.UInt32Col()
+            channel   = tb.UInt32Col()
+            timestamp = tb.UInt64Col()
+            rwf       = tb.UInt16Col(shape=(shape,))
+    elif WD_version == 2:
+        class rwf_df(tb.IsDescription):
+            evt_no    = tb.UInt32Col()
+            channel   = tb.UInt32Col()
+            timestamp = tb.UInt64Col()
+            rwf       = tb.Float32Col(shape = (shape,))
+
+    return rwf_df

core/io.py

@@ -1,4 +1,6 @@
 import pandas as pd
+import tables as tb
+import core.df_classes as df_class 
 
 import ast
 import configparser
@@ -42,4 +44,31 @@ def read_config_file(file_path  :  str) -> dict:
             # we can setup stricter rules at some other time
             arg_dict[key] = ast.literal_eval(config[section][key])
 
-    return arg_dict
+    return arg_dict
+
+
+def create_config_table(h5file : tb.File, dictionary : dict, name : str, description = "config"):
+
+        # create config node if it doesnt exist already
+        try:
+                group = h5file.get_node("/", "config")
+        except tb.NoSuchNodeError:
+                group = h5file.create_group("/", "config", "Config parameters")
+
+        # create table
+        table = h5file.create_table(group, name, df_class.config_class, description)
+        # assign the rows by component
+        config_details = table.row
+        for key, values in dictionary.items():
+            if type(values) is dict:
+                # single nest only! any more and you've made the config too complicated
+                for key_2, value_2 in values.items():
+                    #print(key_2)
+                    config_details['key'] = f'{key}/{key_2}'
+                    config_details['value'] = value_2
+                    config_details.append()
+            else:
+                config_details['key']   = key
+                config_details['value'] = values
+                config_details.append()
+        table.flush()

core/worker.py

@@ -14,7 +14,7 @@ class AcquisitionWorker(Thread):
     All commands and data flow through thread-safe mechanisms (queue, locks, events).
     '''
 
-    def __init__(self, cmd_buffer: Queue, display_buffer: Queue, stop_event: Event):
+    def __init__(self, cmd_buffer: Queue, display_buffer: Queue, stop_event: Event, sw_timeout: float):
         super().__init__(daemon=True)
         self.digitiser = None
         self.stop_event = stop_event
@@ -24,6 +24,7 @@ def __init__(self, cmd_buffer: Queue, display_buffer: Queue, stop_event: Event):
         self.data_ready_callback = None  # set by Controller
         self.dig_config = None
         self.rec_config = None
+        self.sw_timeout = sw_timeout     # set in config file (s)
 
     def enqueue_cmd(self, cmd_type: CommandType, *args):
         '''
@@ -142,7 +143,7 @@ def run(self):
                         logging.exception(f"Acquisition error: {e}")
 
                 # to avoid busy digitiser - add software timeout as member variable
-                time.sleep(1)
+                time.sleep(self.sw_timeout)
 
         except Exception as e:
             logging.exception(f"Fatal error in AcquisitionWorker: {e}")