Helmholtz_Test_Bench/src/psu_device.py

# ABC is template used to create python abstract classes
import time
from abc import ABC, abstractmethod
import serial

from src.ps2000b import PS2000B  # Module containing all PS2000B HW functions and classes
from src.utility import ui_print


class PSUDevice(ABC):
    """The PSUDevice abstract class defines a generic interface for power supplies used with the test bench.
    It can be subclassed to easily support new hardware."""

    def __init__(self, com_port):
        """PSUDevice assumes a serial connection"""
        self.com_port = com_port
        self.cached_state = dict.fromkeys(self.valid_channels())  # Used for components that require state on demand

    @abstractmethod
    def enable_channel(self, channel_nr):
        """Most PSUs offer completely disabling or enabling a channel, beyond setting V=0, I=0.
        Can throw exceptions"""
        pass

    @abstractmethod
    def disable_channel(self, channel_nr):
        """Most PSUs offer completely disabling or enabling a channel, beyond setting V=0, I=0.
        Can throw exceptions"""
        pass

    # Warning: not abstract, does not need to be implemented
    def set_channel_state(self, channel_nr, enabled):
        """True: Enable channel, False: Disable channel
        Can throw exceptions"""
        if enabled:
            self.enable_channel(channel_nr)
        else:
            self.disable_channel(channel_nr)

    @abstractmethod
    def set_current(self, channel_nr, current):
        """Set the current limit. Actual current dependent on OVP or OCP mode.
        Can throw exceptions"""
        pass

    @abstractmethod
    def set_voltage(self, channel_nr, voltage):
        """Set the voltage limit. Actual voltage dependent on OVP or OCP mode.
        Can throw exceptions"""
        pass

    @abstractmethod
    def poll_channel_state(self, channel_nr):
        """Return a dictionary with the entries below. WARNING: this call is blocking and potentially slow!
        Can throw exceptions"""
        # Should also set self.cached_state
        # return {'active': False, 'remote_active': False,
        #         'actual_voltage':0, 'limit_voltage':0, 'actual_current':0, 'limit_current':0}
        pass

    def cached_channel_state(self, channel_nr):
        """Return a dictionary with the entries below. Uses the values obtained during last poll.
        May contain None-entries"""

        if self.cached_state[channel_nr]:
            return self.cached_state[channel_nr]
        else:
            return {'active': None, 'remote_active': None,
                    'actual_voltage': None,
                    'limit_voltage': None,
                    'actual_current': None,
                    'limit_current': None}

    def idle(self):
        """Zero all outputs but activate channels so commands can be sent."""
        for ch in self.valid_channels():
            self.set_current(ch, 0)
            self.set_voltage(ch, 0)
            self.enable_channel(ch)

    @abstractmethod
    def shutdown(self):
        """Shuts the PSU down safely and makes sure ALL outputs are off."""
        pass

    @abstractmethod
    def destroy(self):
        """Disconnects the device"""
        pass

    @staticmethod
    @abstractmethod
    def valid_channels():
        """Returns a list containing valid channel numbers"""
        pass


class PSUDevicePS2000B(PSUDevice):
    """Provides HW support for the Elektro-Automatik PS2000B power supply.
    This class is an adapter for the already existing PS2000B library, to have a consistent interface"""

    MIN_DELAY = 0.05

    def __init__(self, com_port):
        super().__init__(com_port)
        """Can fail; Check for serial.SerialException"""
        self.dev = PS2000B.PS2000B(com_port)
        # dev_info is a class which contains hw specific constants, such as nominal voltage and current
        self.dev_info = self.dev.get_device_information()  # Cache this result
        time.sleep(self.MIN_DELAY)

    @staticmethod
    def valid_channels():
        # Dependent on PSU, the PS2000B has 2 channels
        return [0, 1]

    def enable_channel(self, channel_nr):
        self.dev.enable_output(channel_nr)
        time.sleep(self.MIN_DELAY)

    def disable_channel(self, channel_nr):
        self.dev.enable_output(channel_nr)
        time.sleep(self.MIN_DELAY)

    def set_current(self, channel_nr, current):
        self.dev.set_current(current, channel_nr)
        time.sleep(self.MIN_DELAY)

    def set_voltage(self, channel_nr, voltage):
        self.dev.set_voltage(voltage, channel_nr)
        time.sleep(self.MIN_DELAY)

    def poll_channel_state(self, channel_nr):
        self.dev.update_device_information(channel_nr)  # update the information in the device object
        time.sleep(self.MIN_DELAY)
        dev_status = self.dev.get_device_status_information(channel_nr)  # get object with new status info

        # This is more efficient than the dev.get_voltage() call, since it does not require another update_device_info
        voltage = dev_status.actual_voltage_percent * self.dev_info.nominal_voltage  # Extracted from PS2000B library
        voltage_setp = self.dev.get_voltage_setpoint(channel_nr)
        time.sleep(self.MIN_DELAY)
        current = dev_status.actual_current_percent * self.dev_info.nominal_current
        current_setp = self.dev.get_current_setpoint(channel_nr)
        time.sleep(self.MIN_DELAY)

        # Format should match the provided template in abstract PSUDevice class.
        self.cached_state[channel_nr] = {'active': dev_status.output_active,
                                         'remote_active': dev_status.remote_control_active,
                                         'voltage': voltage, 'voltage_setpoint': voltage_setp,
                                         'current': current, 'current_setpoint': current_setp}
        return self.cached_state[channel_nr]

    def shutdown(self):
        for ch in self.valid_channels():
            self.disable_channel(ch)
            self.set_current(ch, 0)
            self.set_voltage(ch, 0)

    def destroy(self):
        self.dev.serial.close()


class PSUDeviceQL355TP(PSUDevice):
    """HW interface for QL355TP from AIM-TTi Instruments"""

    def __init__(self, com_port):
        """Can fail; Check for serial.SerialException"""
        super().__init__(com_port)
        self._serial_object = serial.Serial(
            port=self.com_port,
            baudrate=19200,
            timeout=0.5,
            parity=serial.PARITY_NONE,
            stopbits=serial.STOPBITS_ONE,
            bytesize=serial.EIGHTBITS
        )
        self.set_output_range(0, 0)  # Put the PSU into the 15V/5A range
        self.set_output_range(1, 0)
        self.reset_breaker()  # Reset the breaker in case we are coming from an unclean state
        self.cached_state = dict.fromkeys(self.valid_channels())  # Used for components that require state on demand

    @staticmethod
    def valid_channels():
        # Dependent on PSU, the QL355TP has 2 normal channels. The auxiliary channel is not usable for our purpose
        return [1, 2]

    def enable_channel(self, channel_nr):
        # The serial interface is documented in the QL355TP handbook
        self._serial_object.write("OP{} 1\n".format(channel_nr).encode())

    def disable_channel(self, channel_nr):
        # The serial interface is documented in the QL355TP handbook
        self._serial_object.write("OP{} 0\n".format(channel_nr).encode())

    def set_current(self, channel_nr, current):
        # The serial interface is documented in the QL355TP handbook
        self._serial_object.write("I{} {}\n".format(channel_nr, current).encode())

    def set_voltage(self, channel_nr, voltage):
        # The serial interface is documented in the QL355TP handbook
        self._serial_object.write("V{} {}\n".format(channel_nr, voltage).encode())

    def poll_channel_state(self, channel_nr):
        # Request channel state
        self._serial_object.write("OP{}?\n".format(channel_nr).encode())
        resp = self._serial_object.read_until()
        output_active = resp.decode().rstrip()

        # Request current current
        self._serial_object.write("I{}O?\n".format(channel_nr).encode())
        resp = self._serial_object.read_until()
        # Trim whitespace and units
        current = float(resp.decode().rstrip()[:-1])

        # Request current setpoint
        self._serial_object.write("I{}?\n".format(channel_nr).encode())
        resp = self._serial_object.read_until()
        # Trim whitespace, prefix and units
        current_setp = float(resp.decode().rstrip()[3:-1])

        # Request current voltage
        self._serial_object.write("V{}O?\n".format(channel_nr).encode())
        resp = self._serial_object.read_until()
        # Trim whitespace and units
        voltage = float(resp.decode().rstrip()[:-1])

        # Request voltage setpoint
        self._serial_object.write("V{}?\n".format(channel_nr).encode())
        resp = self._serial_object.read_until()
        # Trim whitespace and units
        voltage_setp = float(resp.decode().rstrip()[3:-1])

        # Format should match the provided template in abstract PSUDevice class.
        # The remote_active property is assumed to always be True since it cant be read
        # (it should be since we are talking to it)
        self.cached_state[channel_nr] = {'active': output_active, 'remote_active': True,
                                         'voltage': voltage, 'voltage_setpoint': voltage_setp,
                                         'current': current, 'current_setpoint': current_setp}
        return self.cached_state[channel_nr]

    def set_output_range(self, channel_nr, value_range):
        """The QL355TP supports various output ranges. We require the 15V/5A mode to achieve the greatest range."""
        # range 0: 15V/5A, 1=35V/3A, 2=35V/500mA
        self._serial_object.write("RANGE{} {}\n".format(channel_nr, value_range).encode())

    def reset_breaker(self):
        """The QL355TP includes an internal breaker, which must be reset if the
        maximum voltage or current is exceeded"""
        self._serial_object.write("TRIPRST\n".encode())

    def shutdown(self):
        for ch in self.valid_channels():
            self.disable_channel(ch)
            self.set_current(ch, 0)
            self.set_voltage(ch, 0)

    def destroy(self):
        self._serial_object.close()