Source code for semi_cr.core.lab.drivers.cqccV2
# ----------------------------------------------------
# Note
# ----------------------------------------------------
"""
Need to sort out the storage of variables in general
here
"""
# ----------------------------------------------------
# Imports
# ----------------------------------------------------
import time
from pySerialTransfer import pySerialTransfer as txfer
from semi_cr.setup.juno.cqcc.utils.parser import CQCC_V1
# ----------------------------------------------------
# Custom Errors / Helper Classes
# ----------------------------------------------------
# TODO: PUT THEM INTO A FILE AND USE FOR CQCC AND THIS FILE ???
[docs]
class ConnectionError(Exception):
pass
[docs]
class DisconnectionError(Exception):
pass
[docs]
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
[docs]
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
# ----------------------------------------------------
# CQCC Class
# ----------------------------------------------------
[docs]
class CQCC:
# ----------------------------------------------------
# Init, Connect, Disconnect, Check Connection
# ----------------------------------------------------
def __init__(self):
self._settings_path = r"src\semi_cr\setup\juno\cqcc\conf\cqcc_conf\CQCC_V1.yaml"
self.cqcc_parser = CQCC_V1(self._settings_path)
self._status = None
self.connected = False
[docs]
def connect(self, port):
try:
self._status = txfer.SerialTransfer(port)
self._status.open()
time.sleep(2)
self.connected = True
except txfer.InvalidSerialPort as e:
raise ConnectionError(f"Error connecting to Arduino: {e}")
[docs]
def disconnect(self):
try:
self._status.close()
self.connected = False
except Exception as e:
raise DisconnectionError(f"Error disconnecting from Arduino: {e}")
[docs]
def check_connection(self):
if self._status is None:
raise ConnectionError("arduino not connected.")
# ----------------------------------------------------
# Set Channel
# ----------------------------------------------------
# Sets the channel to be measured
[docs]
def set_channel(self, polarity: str, channel: int, state: bool):
# Check arduino connection
if not self._status.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}")
# ----------------------------------------------------
# Helper Functions
# ----------------------------------------------------
# TODO: Add a function description ...
[docs]
def send_switchSingleControlStruct_to_arduino(self, data: SwitchSingleControlStruct):
sendSize = 0
sendSize = self._status.tx_obj(data.exp_type, start_pos=sendSize)
sendSize = self._status.tx_obj(data.pin_list, start_pos=sendSize)
sendSize = self._status.tx_obj(data.level_list, start_pos=sendSize)
sendSize = self._status.tx_obj(data.num_pairs, start_pos=sendSize)
self._status.send(sendSize)
[docs]
def send_switchPairControlStruct_to_arduino(self, data: SwitchPairControlStruct, header=True):
sendSize = 0
if header:
sendSize = self._status.tx_obj(data.exp_type, start_pos=sendSize)
self._status.send(sendSize) # send header first to indicate the type of command
sendSize = 0
sendSize = self._status.tx_obj(data.pin_list, start_pos=sendSize)
sendSize = self._status.tx_obj(data.level_list, start_pos=sendSize)
sendSize = self._status.tx_obj(data.num_pairs, start_pos=sendSize)
sendSize = self._status.tx_obj(data.tail, start_pos=sendSize)
self._status.send(sendSize)
# Wait for acknowledgement from MCU
[docs]
def wait_for_ack(self):
# Need to be connected to arduino
if not self._status.connection.is_open:
raise ConnectionError("Not connected to the _status. Please call connect() first.")
try:
while True:
# 1 - connection successful
if self._status.available():
recSize = 0
ack = self._status.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._status.status.value <= 0:
if self._status.status == txfer.Status.CRC_ERROR:
print("ERROR: CRC_ERROR")
elif self._status.status == txfer.Status.PAYLOAD_ERROR:
print("ERROR: PAYLOAD_ERROR")
elif self._status.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}")