""" DAQ Interface Minimum Functionality Spec """
from daq.operator import *
[docs]class DAQ:
"""
Parent class for device interface layer
* This only exists as documentation
* All methods should be overwritten by inheritors.
"""
def __init__(self, device, config_path=None):
"""
All one-time configuration should be done during initialization
:param device: Handle to low-level interface with device.
:param config_path: Path to config.json file defining all one-time configuration for the Interface
"""
self._device = device
self.config = dict()
self.device_info = dict()
self.channels = list()
self.mode = "control" # "control" or "stream"
self._stream_mode = False # For fast comparisons in the loop
self._stream_buffer = None
# self._configure(config_path) # One-time configuration
def _configure(self, fp):
"""
Load one-time config from disk and apply to hardware.
"""
with open(fp) as f:
config = json.load(f)
self.config = config # Example
self._stream_mode = True if config["mode"] == "stream" else False
[docs] def read(self):
"""
Read values from device.
* Channel list is defined during one-time configuration.
* No inputs are given to this function.
"""
if self._stream_mode: # Stream mode: Return whatever comes out of the device drivers
read_values = self._read_stream()
else: # Control mode: Return an iterable of float64 elements of length len(self.channels)
read_values = [0.0] * len(self.channels)
return read_values
[docs] def command(self, assembled_commands):
"""
Send commands to device low-level interface.
* As much processing as possible is done prior to this step.
* Mapping from user input to device-level interface is done during initialization.
"""
pass
[docs] def stream(self):
"""Open high-rate read stream"""
self._stream_buffer = None
[docs] def close(self):
"""
1. End stream
2. Return to default settings
3. Close connection to device
"""
pass
[docs] def assemble_command_table(self, command_table):
"""
Assemble a human-readable command block into device-friendly commands
:param command_table: Time-series table of commands
:type command_table: pandas.DataFrame
"""
pass
[docs] def process_stream_data(self, stream_data):
"""Reformat stream output after run"""
stream_data_processed = stream_data # Do something to convert to a dataframe, apply calibrations, etc
return stream_data_processed
[docs] def get_config(self):
"""Return current config as json"""
config_json = None
return config_json
def _read_stream(self):
"""Get values from stream buffer"""
if self._stream_buffer is not None:
stream_buffer_values = self._read_stream() # Take output as whatever format computationally intensive
else:
stream_buffer_values = None
return stream_buffer_values
def __str__(self):
"""Return a unique identifier for the DAQ hardware"""
return "{}_SN_{}".format(type(self), self.device_info["Serial Number"])