Source code for semi_cr.core.lab.drivers.cqcc

import time

from pySerialTransfer import pySerialTransfer as txfer
from qcodes.instrument import Instrument

from semi_cr.setup.juno.cqcc.utils.parser import CQCC_V1




class QcodesCQCC(Instrument):
    def __init__(self, name: str, port: str):
        # Calls Instrument.__init__(name)
        super().__init__(name)
        self.cqcc = CQCC(settings_path=r"src\semi_cr\setup\juno\cqcc\conf\cqcc_conf\CQCC_V1.yaml")



    def close(self):
        try:
            self.cqcc.disconnect()
        finally:
            super().close()




    def get_idn(self) -> dict[str, str | None]:
        return {
            "vendor": "NQCP_Spin",
            "model": self.name,
            "serial": None,
            "firmware": None,
        }






class ConnectionError(Exception):
    pass





class SwitchSingleControlStruct:
    """
    Class for constructing data between PC and MCU
    This class is used for open single channel control
    """

    exp_type: str = "M"
    pin_list: list[int]
    level_list: list[int]
    num_pairs: int = 9  # this is always 9 for single control





class SwitchPairControlStruct:
    """
    Class for constructing data between PC and MCU
    This class is used for open pair channel control
    """

    exp_type: str = "K"
    pin_list: list[int]
    level_list: list[int]
    num_pairs: int = 18  # this is always 18 for pair control (9 for positive, 9 for GND
    tail: bool = False





class CQCC:
    def __init__(self, settings_path=r"src\semi_cr\setup\juno\cqcc\conf\cqcc_conf\CQCC_V1.yaml"):
        """
        Initialize cqcc driver

        Parameters:
        Settings_path: stores the hardware mapping

        """
        self.cqcc_parser = CQCC_V1(settings_path)
        self.connected = False
        self.arduino = None

    # Connects to arduino


    def connect(self, COM):
        try:
            self.arduino = txfer.SerialTransfer(COM)  # initialize the serial transfer
            self.arduino.open()
            time.sleep(2)  # wait for the connection to be established
        except txfer.InvalidSerialPort as e:
            raise ConnectionError(f"Connection error: Invalid serial port {COM}: {e}")


    # Sets the channel to be measured


    def set_channel(self, polarity: str, channel: int, state: bool):
        # Check arduino connection
        if not self.arduino.connection.is_open:
            raise ConnectionError("Not connected to the Arduino. Please call connect() first.")
        # Check polarity is specified and channel in valid range
        try:
            if polarity not in {"Positive", "GND"}:
                raise ValueError("Polarity must be either 'Positive' or 'GND'.")
            num_of_channels = 96
            if (channel > 0) and (channel < num_of_channels + 1):
                # Set the desired channel on the arduino
                signal = self.cqcc_parser.port_to_signal(f"dc{channel}", polarity)
                command = SwitchSingleControlStruct()
                command.exp_type = "M"  # let controller know now is set a single channel
                command.pin_list = [pin for pin, _ in signal.items()]
                command.level_list = [value for _, value in signal.items()]
                command.num_pairs = len(command.pin_list)
                self.send_switchSingleControlStruct_to_arduino(command)
                # Wait for signal from arduino to notify successful or unsuccessful connection
                ack = self.wait_for_ack()
                if ack:
                    print(f"Channel {channel} set to {'HIGH' if state else 'LOW'} successfully.")
                else:
                    raise ConnectionError(f"Failed to set channel {channel}. No ACK received from the Arduino.")
            else:
                raise ValueError("Channel number must be between 1 and 96.")
        except ValueError as e:
            raise ValueError(f"Value error: {e}")
        except ConnectionError as e:
            raise ConnectionError(f"Connection error: {e}")




    def disconnect(self):
        try:
            self.arduino.close()
        except Exception as e:
            raise ConnectionError(f"Connection error: Error closing the connection: {e}")


    # Wait for acknowledgement from MCU


    def wait_for_ack(self):
        # Need to be connected to arduino
        if not self.arduino.connection.is_open:
            raise ConnectionError("Not connected to the Arduino. Please call connect() first.")
        try:
            while True:
                # 1 - connection successful
                if self.arduino.available():
                    recSize = 0
                    ack = self.arduino.rx_obj(obj_type="c", start_pos=recSize)
                    ack = int.from_bytes(ack)
                    print(f"Received ACK: {ack}")
                    if ack == 1:
                        return ack
                # 0 - connection unsuccessful
                elif self.arduino.status.value <= 0:
                    if self.arduino.status == txfer.Status.CRC_ERROR:
                        print("ERROR: CRC_ERROR")
                    elif self.arduino.status == txfer.Status.PAYLOAD_ERROR:
                        print("ERROR: PAYLOAD_ERROR")
                    elif self.arduino.status == txfer.Status.STOP_BYTE_ERROR:
                        print("ERROR: STOP_BYTE_ERROR")
                    else:
                        print(f"ERROR: {txfer.Status.name}")
                    return None
        except Exception as e:
            raise ConnectionError(f"Connection error: Error waiting for ACK: {e}")




    def send_switchSingleControlStruct_to_arduino(self, data: SwitchSingleControlStruct):
        sendSize = 0
        sendSize = self.arduino.tx_obj(data.exp_type, start_pos=sendSize)
        sendSize = self.arduino.tx_obj(data.pin_list, start_pos=sendSize)
        sendSize = self.arduino.tx_obj(data.level_list, start_pos=sendSize)
        sendSize = self.arduino.tx_obj(data.num_pairs, start_pos=sendSize)
        self.arduino.send(sendSize)




    def send_switchPairControlStruct_to_arduino(self, data: SwitchPairControlStruct, header=True):
        sendSize = 0
        if header:
            sendSize = self.arduino.tx_obj(data.exp_type, start_pos=sendSize)
            self.arduino.send(sendSize)  # send header first to indicate the type of command
        sendSize = 0
        sendSize = self.arduino.tx_obj(data.pin_list, start_pos=sendSize)
        sendSize = self.arduino.tx_obj(data.level_list, start_pos=sendSize)
        sendSize = self.arduino.tx_obj(data.num_pairs, start_pos=sendSize)
        sendSize = self.arduino.tx_obj(data.tail, start_pos=sendSize)
        self.arduino.send(sendSize)