# -*- coding: utf-8 -*- """ Created on Fri Jan 24 15:18:31 2025 @author: Matthias Quintern """ from .measurement_device.base import VoltageMeasurementDevice from .led_control_device.base import LedControlDevice from .led_script import LedScript from .utility.prsdata import DataCollector import time import datetime from queue import Queue import logging log = logging.getLogger(__name__) def measure( vm_dev: VoltageMeasurementDevice, led_dev: LedControlDevice, led_script: LedScript, data: DataCollector, delta_t: float=0.1, flush_after:int|None=None, use_buffer=False, max_measurements: int=None, stop_on_script_end: bool=False, verbose: bool=False, command_queue: None|Queue=None, data_queue: None|Queue=None, add_measurement_info_to_metadata=True ): """ Perform a measurement Parameters ---------- vm_dev : VoltageMeasurementDevice DESCRIPTION. led_dev : LedControlDevice DESCRIPTION. led_script : LedScript DESCRIPTION. data : DataCollector DESCRIPTION. delta_t : float, optional Target interval between measurements and led updates. The default is 0.1. flush_after : int|None, optional If int, flush values to disk after . The default is None. use_buffer : TYPE, optional If True, use the buffer measurement mode. The default is False. max_measurements : int, optional Number of measurements to perform before returning. Note: If use_buffer=True, a few more than max_measurements might be performed The default is None. stop_on_script_end : bool, optional Stop when the script end is reached. verbose : bool, optional If True, print some messages. The default is False. command_queue : None|Connection, optional A queue to receive to commands from. Commands may be: "stop" -> stops the measurement ("led_script", ) a new led script to use The default is None. data_queue : None|Queue, optional A queue to put data in. The default is None. add_measurement_info_to_metadata : bool, optional If True, add measurement info to the metadata: time, measurement_interval, measurement_use_buffer, measurement_voltage_device, measurement_led_device The default is True. Returns ------- None. """ get_time = lambda: datetime.datetime.now().strftime("%Y-%m-%d_%H:%M:%S") if add_measurement_info_to_metadata: data.metadata["measurement_interval"] = str(delta_t) + " s" data.metadata["measurement_use_buffer"] = str(use_buffer) data.metadata["measurement_voltage_measurement_device"] = str(vm_dev) data.metadata["measurement_led_control_device"] = str(led_dev) led_name = led_dev.get_led_name() if led_name: data.metadata["measurement_led_lamp"] = led_name data.metadata["measurement_time_start"] = get_time() # write metadata to disk data.write_metadata() vm_dev.reset(True) if use_buffer: vm_dev.buffer_measure(delta_t, verbose=True) # allow 0 instead of None if max_measurements == 0: max_measurements = None if flush_after == 0: flush_after = None try: i = 0 led_val = led_script.start() try: led_dev.set_level(led_val) except Exception as e: log.error(f"Error setting led to {led_val:03}%: {e}") raise e t_iter_start = time.time() while True: # using while True and if, to be able to log the stop reason if max_measurements is not None and i >= max_measurements: log.info(f"Reached maximum number of measurements ({i}{max_measurements}), stopping measurement") break # 1) read value(s) if use_buffer: try: values = vm_dev.buffer_read_new_values() except ValueError as e: # print(f"Error in buffer measurement {i}:", e) values = [] else: values = [vm_dev.read_value()] # print(values) # 2) process value(s) for (tval, vval) in values: if i == 0: t0 = tval tval -= t0 current_data = (i, tval, vval, led_val) data.add_data(*current_data) # 3) write data if verbose: print(f"n = {i:6d}, t = {tval: .2f} s, U = {vval: .5f} V, LED = {led_val:03}%" + " "*10, end='\r') if flush_after is not None and (i+1) % flush_after == 0: data.flush(verbose=verbose) # if a queue was given, put the data if data_queue is not None: data_queue.put(current_data) i += 1 # if a pipe was given, check for messages if command_queue is not None and command_queue.qsize() > 0: recv = command_queue.get(block=False) if recv == "stop": log.info(f"Received 'stop', stopping measurement") break elif type(recv) == tuple and recv[0] == "led_script": log.info(f"Received 'led_script', replacing script") led_script = recv[1] elif type(recv) == tuple and recv[0] == "metadata": log.info(f"Received 'metadata', updating metadata") data.metadata |= recv[1] data.write_metadata() else: log.error(f"Received invalid message: '{recv}'") # 4) sleep # subtract the execution time from the sleep time for a more # accurate frequency dt_sleep = delta_t - (time.time() - t_iter_start) if dt_sleep > 0: # print(f"Sleeping for {dt_sleep}") time.sleep(dt_sleep) t_iter_start = time.time() # 5) update LED if stop_on_script_end and led_script.is_done(t_iter_start): log.info("Reached led script end, stopping measurement") break new_led_val = led_script.get_state(t_iter_start) if new_led_val != led_val: try: led_dev.set_level(new_led_val) led_val = new_led_val except Exception as e: log.error(f"Error setting led to {new_led_val:03}%: {e}") raise e except KeyboardInterrupt: log.info("Keyboard interrupt, stopping measurement") except Exception as e: log.critical(f"Unexpected error, stopping measurement. Error: {e}") if command_queue is not None: command_queue.put(("exception", e)) if add_measurement_info_to_metadata: data.metadata["measurement_time_stop"] = get_time() # Write again after having updated the stop time data.write_metadata() data.flush() led_dev.off()