measurement_device_control.py

"""This module enables connecting to a KEYSIGHT N6075B power measurement device and measuring power in a certain
interval

Right now, this assumes that the device is already set to be a voltage priority power source on output @1.
"""

import socket

class MeasurementDevice:

    def set_marker_2_position_relative_to_trigger_point(self, distance_to_trigger_point):
        self.send_command(b':SENSe:DLOG:MARKer2:POINt ' +
                          bytes(str(distance_to_trigger_point), 'ascii'))

    def set_marker_1_position_relative_to_trigger_point(self, distance_to_trigger_point):
        self.send_command(b':SENSe:DLOG:MARKer1:POINt ' +
                          bytes(str(distance_to_trigger_point), 'ascii'))

    def enable_current_measurement(self):
        self.send_command(b':SENSe:DLOG:FUNCtion:CURRent 1,(@1)')

    def trigger_immediately(self):
        self.send_command(b':TRIGger:DLOG:IMMediate')

    def start_data_logging(self, log_file_name="current"):
        self.send_command(b':INITiate:IMMediate:DLOG "internal:/' +
                          bytes(log_file_name, 'ascii') + b'.dlog"')

    def abort_data_logging(self):
        self.send_command(b'ABORt:DLOG')

    def get_average_current_in_interval(self):
        """Return the average current between both markers in amps. This might not return if logging did not run long
        enough so that the second marker was reached"""
        self.send_command(b':FETCh:DLOG:CURRent:DC? (@1)')
        data = self.receive_reply(4096)
        return float(data)

    def receive_reply(self, size):
        data = self.socket.recv(size)
        return data

    def send_command(self, command):
        self.socket.sendall(command + b"\n")


def connect_to_device(address):
    """Connect to the measurement device found at <address> and return a MeasurementDevice object holding the status."""
    device = MeasurementDevice()
    device.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    device.socket.connect((address, 5025))
    return device


def start_measurement(device):
    """Convenience method for starting logging on device, start triggering immediately, enabling measurement of current
    and set marker 1 to the same position as the trigger"""
    device.start_data_logging()
    device.trigger_immediately()
    device.enable_current_measurement()
    device.set_marker_1_position_relative_to_trigger_point(0)


def get_average_current_to_time_delta(device, time_delta):
    """Convenience method for getting the average current between marker 1 and a time_delta in seconds. Data logging
    must run at least as long as time_delta is, or otherwise this might not return."""
    device.abort_data_logging()
    device.set_marker_2_position_relative_to_trigger_point(time_delta)
    return device.get_average_current_in_interval()


__author__ = "Roland Kunkel"
__license__ = "Apache-2.0"