chroma电子负载驱动开发自动化编程

本文实现了针对 Chroma 63004 系列（150V/60A/350W）可编程直流电子负载的完整 SCPI 驱动库。驱动封装了 CC/CV/CR/CP等多种工作模式，提供直观的 Python API，支持多种通信总线。

import time
import pyvisa as visa

# ==================== 硬件规格常量 ====================
MAX_CURR = 60.0 # 最大电流60A
MAX_POWER = 350.0 # 最大功率350W
MAX_VOLT = 150.0 # 最大电压150V
MAX_RES_HIGH = 2500 # CR高档最大电阻
MAX_LS = 20e-6 # CZ最大串联电感 20uH
MAX_CL = 50000e-6 # CZ最大并联电容 50000uF
SAMPLE_POINT_MAX = 15000 # 数字化最大采样点

# 手册标准合法模式集合
VALID_MODES = [
"CCL", "CCM", "CCH", # 恒流 低/中/高
"CRL", "CRM", "CRH", # 恒阻 低/中/高
"CVL", "CVM", "CVH", # 恒压 低/中/高
"CPL", "CPM", "CPH", # 恒功率 低/中/高
"CCDL", "CCDM", "CCDH", # 动态恒流
"CRDL", "CRDM", "CRDH", # 动态恒阻
"CZL", "CZM", "CZH", # 需要可拓展，恒阻抗 低/中/高
"BATT", "OCPH", "OPPH", "PROG", "UDW" # 后续需要可补充
]

SIGNAL_STATE = ["OFF", "ON"] # 信号状态：0=OFF, 1=ON
MEAS_INPUT = ["LOAD", "UUT"] # 测量源：0=负载端, 1=待测物端
VOLT_RESP = ["SLOW", "NORMAL", "FAST"] # CV模式响应速度
# 故障位映射（FETCh:STATus? 返回值解析）
PROTECT_BIT_MAP = {
0: "OV1 轻度过压",
1: "OV2 过峰值电压",
2: "REV 输入反接",
3: "OCP1 硬件过流",
4: "OCP2 过峰值电流",
5: "OCP3 用户自定义过流",
6: "OPP1 硬件过功率",
7: "OPP3 用户自定义过功率",
9: "OTP 过温保护",
10: "FAN 风扇故障",
15: "RMT_INH 远程抑制"
}

# 日志工具函数，兼容外部logging
def log_print(level, tag, msg, logger=None):
if logger is not None:
if level == "ERR":
logger.error(f"{tag:24s} | {msg}")
elif level == "WRN":
logger.warning(f"{tag:24s} | {msg}")
else:
logger.info(f"{tag:24s} | {msg}")
else:
print(f"[{level}] {tag:24s} | {msg}")

class Chroma63004(object):
"""
Chroma 63004-150-60 完整SCPI驱动
支持模式：CC/CR/CV/CP/CCD/CRD/BATT/OCP/OPP/CZ/UDW/PROG
总线：USB VISA / GPIB / TCPIP Socket
使用示例：
with Chroma63004() as eload:
eload.open("TCPIP0::192.168.1.100::2101::SOCKET")
eload.Load_Up_Down("ON", "CCH", 5.0)
v = eload.measure_V()
eload.Load_Up_Down("OFF")
"""
instrContext = 'Chroma63004'
MSG_ERR = '%-24s : ' % (instrContext.upper() + '-ERR')
MSG_WRN = '%-24s : ' % (instrContext.upper() + '-WRN')
MSG_INF = '%-24s : ' % (instrContext.upper() + '-INF')

def __init__(self, logger=None):
self.logger = logger
self.rm = None
self.inst = None
try:
self.rm = visa.ResourceManager()
self.D = list(self.rm.list_resources())
log_print(self.MSG_INF, "VISA资源扫描完成", f"设备列表:{self.D}", self.logger)
except Exception as ex:
self.import_error(ex)

# 模式/状态映射表
self.mode = VALID_MODES
self.input_port = MEAS_INPUT
self.curr_subsystem = ["STATic", "DYNamic"]
self.state = SIGNAL_STATE
self.volt_mode = VOLT_RESP

def resource_list(self):
"""返回扫描到的全部VISA设备地址"""
return self.D

# ==================== 连接与断开管理 ====================
def open(self, inst_resource: str = "", reset=True, idn=True, timeout=3000, retry=2):
attempt = 0
while attempt <= retry:
try:
self.inst = self.rm.open_resource(inst_resource)
self.inst.write_termination = "\\n"
self.inst.read_termination = "\\n"
self.inst.timeout = timeout
break
except Exception as e:
attempt += 1
log_print(self.MSG_WRN, f"连接尝试{attempt}失败", str(e), self.logger)
time.sleep(0.5)
if attempt > retry:
self.exceptionhandler(e)
return
if len(self.D) == 0:
log_print(self.MSG_WRN, "设备扫描", "未检测到任何VISA仪器", self.logger)
if reset:
self.reset()
if idn:
idn_str = self.inst_Q()
log_print(self.MSG_INF, "设备连接成功", f"{idn_str}", self.logger)
self.write('CONF:VOLT:ON 350mV') # 默认开启拉载电压

def close(self):
"""关闭仪器会话，先切断负载输出"""
try:
if self.inst:
self.load_subsystem(0)
idn_str = self.inst_Q().strip()
self.inst.close()
log_print(self.MSG_INF, "仪器断开", f"{idn_str} 会话已关闭", self.logger)
self.inst = None
except Exception as ex:
self.exceptionhandler(ex)

def inst_close(self):
"""全局VISA资源管理器销毁，程序结束调用"""
if self.rm:
self.rm.close()
log_print(self.MSG_INF, "资源释放", "VISA管理器已关闭", self.logger)

# 上下文管理器 with 语法支持
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.close()

# ==================== 统一异常处理 ====================
def import_error(self, ex):
log_print(self.MSG_ERR, "初始化异常", f"VISA库加载失败: {str(ex)}", self.logger)
def exceptionhandler(self, exception):
log_print(self.MSG_ERR, "运行异常", str(exception), self.logger)
def _safe_float(self, raw_str):
"""数值读取容错，非法返回0"""
if raw_str is None:
return 0.0
try:
return float(raw_str.strip())
except (ValueError, AttributeError):
log_print(self.MSG_WRN, "数值解析失败", f"原始返回: {raw_str}", self.logger)
return 0.0

# ==================== IEEE488.2 通用标准指令 ====================
def reset(self):
"""整机复位、清空寄存器、清除保护锁"""
try:
self.write("*RST")
self.default_setup()
self.clear_protect()
time.sleep(0.1)
self.write("*OPC?")
except Exception as ex:
self.exceptionhandler(ex)
def default_setup(self):
"""手册推荐事件寄存器配置"""
self.write("*ESE 60;*SRE 48;*CLS")
def inst_Q(self):
"""读取设备型号序列号 *IDN?"""
return self.write("*IDN?")
def common_Q(self, cmd_str: str):
"""通用任意查询指令"""
return self.write(cmd_str)
def get_sys_err(self):
"""读取仪器内部错误 (错误码, 描述文本)"""
rsp = self.write("SYST:ERR?")
if not rsp:
return 0, "No Error"
code, msg = rsp.split(",", 1)
return int(code), msg.strip('"')
def clear_protect(self):
"""清除过压/过流/过温保护锁"""
self.write("LOAD:PROT:CLE")
def parse_protect_status(self, status_code: int) -> list:
"""数字故障码转中文告警列表"""
err_list = []
for bit, desc in PROTECT_BIT_MAP.items():
if status_code & (1 << bit):
err_list.append(desc)
return err_list
def save_state(self, slot=0):
"""*SAV 保存配置到存储槽 0~120"""
if 0 <= slot <= 120:
self.write(f"*SAV {slot}")
log_print(self.MSG_INF, "状态保存", f"存储槽{slot}", self.logger)
def recall_state(self, slot=0):
"""*RCL 读取存储槽配置，自动清除保护"""
if 0 <= slot <= 120:
self.write(f"*RCL {slot}")
log_print(self.MSG_INF, "状态恢复", f"存储槽{slot}", self.logger)
time.sleep(0.1)
self.clear_protect()

# ==================== 底层统一SCPI读写核心 ====================
def write(self, cmd: str = "", retry=1):
"""自动区分读/写指令，失败重试，统一日志"""
if not isinstance(cmd, str) or len(cmd.strip()) == 0:
log_print(self.MSG_WRN, "指令非法", "SCPI指令不能为空", self.logger)
return None
cmd = cmd.strip()
attempt = 0
while attempt <= retry:
try:
if cmd.endswith('?'):
rsp = self.inst.query(cmd)
log_print(self.MSG_INF, f"发送查询", f"{cmd} -> {rsp.strip()}", self.logger)
return rsp.strip()
else:
self.inst.write(cmd)
log_print(self.MSG_INF, "发送指令", cmd, self.logger)
return None
except Exception as ex:
attempt += 1
log_print(self.MSG_WRN, f"指令发送重试{attempt}", str(ex), self.logger)
time.sleep(0.2)
if attempt > retry:
self.exceptionhandler(ex)
return None

# ==================== 负载模式切换 ====================
def mode_subsystem(self, mode_idx: int):
"""数字索引切换模式"""
if 0 <= mode_idx < len(self.mode):
self.write(f':MODE {self.mode[mode_idx]};')
else:
log_print(self.MSG_ERR, "模式索引越界", str(mode_idx), self.logger)
def Set_mode_subsystem(self, mode: str):
"""字符串直接指定工作模式"""
mode = mode.upper()
if mode not in self.mode:
log_print(self.MSG_ERR, "模式不支持", f"{mode}，可选{self.mode}", self.logger)
return
self.write(f':MODE {mode};')
log_print(self.MSG_INF, "切换负载模式", f"当前模式 = {mode}", self.logger)

# ==================== CC / CCD 恒流/动态恒流 ====================
def current_subsystem(self, curr_subsystem=0, max_iload1=0.0, min_iload2=0.0,
rise_slewrate=1.0, fall_slewrate=1.0,
dynamic_duration1="10ms", dynamic_duration2="10ms", repeat=0):
if max_iload1 > MAX_CURR or min_iload2 > MAX_CURR:
log_print(self.MSG_ERR, "电流超限", f"最大{MAX_CURR}A", self.logger)
return
sub_mode = self.curr_subsystem[curr_subsystem]
cmd = (f":CURRent:{sub_mode}:L1 {max_iload1};:CURRent:{sub_mode}:L2 {min_iload2};"
f":CURRent:{sub_mode}:RISE {rise_slewrate};:CURRent:{sub_mode}:FALL {fall_slewrate};")
self.write(cmd)
if curr_subsystem == 1:
self.write(f":CURRent:{sub_mode}:T1 {dynamic_duration1.lower()};")
self.write(f":CURRent:{sub_mode}:T2 {dynamic_duration2.lower()};")
self.write(f":CURRent:{sub_mode}:REP {repeat}")
def current_query(self, param="L1"):
return self.write(f"CURR:STAT:{param}?")

# ==================== CR / CRD 恒阻/动态恒阻 ====================
def resistance_subsystem(self, static=True, r1=10.0, r2=0.0,
rise=0.5, fall=0.5, t1="10ms", t2="10ms", repeat=0):
"""恒阻/动态恒阻配置"""
if r1 > MAX_RES_HIGH:
log_print(self.MSG_WRN, "电阻超限", f"最大{MAX_RES_HIGH}Ω", self.logger)

if static:
# 确保在静态模式
self.Set_mode_subsystem("CRH") # 或 CRL/CRM
time.sleep(0.05)
self.write(f":RES:STAT:L1 {r1};:RES:STAT:RISE {rise};:RES:STAT:FALL {fall};")
else:
# 先切换到动态模式
self.Set_mode_subsystem("CRDH") # 或 CRDL/CRDM
time.sleep(0.05) # 等待模式切换完成

# 配置动态参数
cmd = (f":RES:DYN:L1 {r1};"
f":RES:DYN:L2 {r2};"
f":RES:DYN:T1 {t1.lower()};"
f":RES:DYN:T2 {t2.lower()};"
f":RES:DYN:RISE {rise};"
f":RES:DYN:FALL {fall};"
f":RES:DYN:REP {repeat}")
self.write(cmd)

# 验证模式
mode_verify = self.write("MODE?")
log_print(self.MSG_INF, "模式验证", f"当前模式: {mode_verify}", self.logger)
# ==================== CV 恒压模式 ====================
def voltage_subsystem(self, v_set=12.0, i_limit=5.0, resp="SLOW"):
resp = resp.upper()
if resp not in VOLT_RESP:
log_print(self.MSG_ERR, "响应速度非法", resp, self.logger)
return
if v_set > MAX_VOLT:
log_print(self.MSG_ERR, "电压越限", f"最大{MAX_VOLT}V", self.logger)
return

# 使用完整命令，避免任何缩写歧义
# self.write(f"VOLT:STAT:L1 {v_set};VOLT:STAT:ILIM {i_limit};VOLT:STAT:RES {resp}")
cmd = (f":VOLTage:STATic:L1 {v_set};"
f":VOLTage:STATic:ILIM {i_limit};"
f":VOLTage:STATic:RES {resp}")
self.write(cmd)
log_print(self.MSG_INF, "CV模式配置",
f"V={v_set}V, ILIM={i_limit}A, RES={resp}", self.logger)

def set_von(self, voltage=0.35):
self.write(f"CONF:VOLT:ON {voltage}V")
def set_voff(self, voltage=0.0):
self.write(f"CONF:VOLT:OFF {voltage}")
def set_von_latch(self, latch="OFF"):
self.write(f"CONF:VOLT:LATCH {latch}")
def query_von(self):
return self._safe_float(self.write("CONF:VOLT:ON?"))
def query_voff(self):
return self._safe_float(self.write("CONF:VOLT:OFF?"))

# ==================== CP 恒功率 ====================
def power_subsystem(self, p_set=100.0, rise=1.0, fall=1.0):
if p_set > MAX_POWER:
log_print(self.MSG_ERR, "功率超限", f"最大{MAX_POWER}W", self.logger)
return
self.write(f":POW:STAT:L1 {p_set};:POW:STAT:RISE {rise};:POW:STAT:FALL {fall};")

# ==================== 负载总开关 & 模拟短路 ====================
def load_subsystem(self, active=1):
"""数字0=OFF,1=ON"""
self.write(f":LOAD {self.state[int(active)]};")
def Set_load_subsystem(self, active: str = "OFF"):
"""字符串 ON/OFF"""
active = active.upper()
if active in ["ON", "OFF"]:
self.write(f":LOAD {active};")
log_print(self.MSG_INF, "负载输出", f"状态={active}", self.logger)
def short_on(self):
self.write("LOAD:SHOR ON")
log_print(self.MSG_WRN, "短路警告", "已开启模拟短路！", self.logger)
def short_off(self):
self.write("LOAD:SHOR OFF")
def set_short_current(self, current=0):
self.write(f"LOAD:SHOR:LIM {current}")

# ==================== 量测配置与实时读数 ====================
def measure_subsystem(self, input_port=0, avg_window=0.02):
self.write(f":MEASure:INPut {self.input_port[int(input_port)]};")
if 0.02 <= avg_window <= 2.0:
self.write(f"CONF:WIND {avg_window}")
def measure_V(self):
"""实时采样电压"""
return self._safe_float(self.write("MEASure:VOLTage?"))
def measure_I(self):
"""实时采样电流"""
return self._safe_float(self.write("MEASure:CURRent?"))
def measure_P(self):
"""实时采样功率"""
return self._safe_float(self.write("MEASure:POW?"))
def fetch_V(self):
"""读取缓存电压，不重新采样"""
return self._safe_float(self.write("FETCh:VOLTage?"))
def fetch_I(self):
return self._safe_float(self.write("FETCh:CURRent?"))
def fetch_P(self):
return self._safe_float(self.write("FETCh:POW?"))
def get_load_status(self):
return self.write("LOAD?")
def fetch_status(self):
raw = self.write("FETCh:STATus?")
return int(raw) if raw and raw.isdigit() else 0
def fetch_time(self):
return self._safe_float(self.write("FETCh:TIME?"))

# ==================== BATT 电池放电模式 ====================
def batt_config(self, mode="CC", value=5.0, end_v=10.5, timeout=3600, rise=1.0, fall=1.0):
mode_map = {"CC": 0, "CR": 1, "CP": 2}
if mode.upper() not in mode_map:
log_print(self.MSG_ERR, "BATT模式错误", mode, self.logger)
return
self.write(f"ADV:BATT:MODE {mode_map[mode.upper()]}")
self.write(f"ADV:BATT:VAL {value}")
self.write(f"ADV:BATT:ENDV {end_v}")
self.write(f"ADV:BATT:TOUT {timeout}")
self.write(f"ADV:BATT:RISE {rise};ADV:BATT:FALL {fall}")
log_print(self.MSG_INF, "BATT配置", f"模式={mode},截止={end_v}V", self.logger)
def batt_read_ah(self):
return self._safe_float(self.write("MEAS:AH?"))
def batt_read_wh(self):
return self._safe_float(self.write("MEAS:WH?"))
def batt_read_time(self):
return self._safe_float(self.write("MEAS:TIME?"))

# ==================== 规格判定 ====================
def spec_set_volt(self, center, low_pct, high_pct, mode="PERCENT"):
self.write(f"SPEC:VOLT:CEN {center};SPEC:VOLT:LOW {low_pct};SPEC:VOLT:HIGH {high_pct}")
self.write(f"SPEC:UNIT {1 if mode.upper() == 'VALUE' else 0}")
def spec_set_curr(self, center, low_pct, high_pct):
self.write(f"SPEC:CURR:CEN {center};SPEC:CURR:LOW {low_pct};SPEC:CURR:HIGH {high_pct}")
def spec_set_power(self, center, low_pct, high_pct):
self.write(f"SPEC:POW:CEN {center};SPEC:POW:LOW {low_pct};SPEC:POW:HIGH {high_pct}")
def spec_enable(self, enable=True):
self.write(f"SPEC:TEST {'ON' if enable else 'OFF'}")
def spec_get_result(self):
return self.write("SPEC?")

# ==================== 系统全局配置 ====================
def configure_subsystem(self, key_lock=0, sound=1, brightness=3, factory_reset=False):
self.write(f":CONFigure:KEY {key_lock};:CONFigure:SOUND {sound};:CONF:BRIG {brightness}")
if factory_reset:
self.write("CONF:FACT")
def set_gpib_address(self, address):
if 1 <= address <= 30:
self.write(f"COMM:ADDR:GPIB {address}")
def set_digital_io(self, pin, mode):
self.write(f"CONF:DIO:{pin} {mode}")

# ==================== DIG 数字化波形采集 ====================
def digitizing_config(self, points=1024, sample_time="20us", trig_source="LOADON"):
if points > SAMPLE_POINT_MAX:
log_print(self.MSG_ERR, "采样点超限", f"最大{SAMPLE_POINT_MAX}", self.logger)
return
self.write(f"DIG:SAMP:POIN {points};DIG:SAMP:TIME {sample_time};DIG:TRIG:SOUR {trig_source}")
def digitizing_start(self):
self.write("DIG:INIT")
def digitizing_abort(self):
self.write("DIG:ABOR")
def digitizing_read(self, data_type="V"):
return self.write(f"DIG:WAV:DATA?")
def digitizing_capture(self):
return self.write("DIG:WAV:CAP?")