Electronic Load Molecule

"""SDL1000X SCPI adapter. Hardware qualification pending; simulation is explicit."""
import math,time,threading

class Load:
 def __init__(self,resource=None,serial=None,simulate=False):
  self.lock=threading.RLock();self.simulate=simulate;self.io=None;self.connected=False
  self.serial=serial;self.idn=None;self.current=0.;self.enabled=False;self.error=None;self.samples=[];self.deadline=0
  if simulate:self.connected=True;self.idn='SIMULATION,SDL1020X-E,DEMO,0'
  elif resource:
   if not serial:raise ValueError('Pin the instrument serial with --serial before connecting')
   import pyvisa
   self.rm=pyvisa.ResourceManager('@py');self.io=self.rm.open_resource(resource)
   self.io.timeout=2000;self.io.write_termination='\n';self.io.read_termination='\n'
   try:
    self.idn=self.io.query('*IDN?').strip();p=[s.strip() for s in self.idn.split(',')]
    if len(p)<3 or 'SIGLENT' not in p[0].upper() or p[1]!='SDL1020X-E' or p[2]!=serial:raise ValueError('Instrument model/serial mismatch: refusing control')
    self.connected=True;self.off();self.configure()
   except Exception:
    self.io.close();self.io=None;raise
 def write(self,c):
  if not self.connected:raise RuntimeError('No instrument connected. Configure a VISA resource and serial.')
  if self.io:self.io.write(c)
 def query(self,c):return self.io.query(c).strip()
 def configure(self):
  for c in [':SOURCE:SHORT:STATE OFF',':SOURCE:EXT:INPUT:STATE OFF',':SOURCE:FUNCTION CURRENT',':SOURCE:CURRENT:IRANGE 5',':SOURCE:CURRENT:LEVEL:IMMEDIATE 0','SYSTEM:SENSE:STATE OFF',':SOURCE:CURRENT:PROTECTION:LEVEL 1.1',':SOURCE:CURRENT:PROTECTION:STATE ON',':SOURCE:POWER:PROTECTION:LEVEL 6',':SOURCE:POWER:PROTECTION:STATE ON']:self.write(c)
  if self.io:
   if self.query(':SOURCE:FUNCTION?').upper() not in ['CURRENT','CURR']:raise RuntimeError('CC mode readback failed')
   if abs(float(self.query(':SOURCE:CURRENT:LEVEL:IMMEDIATE?')))>0.002:raise RuntimeError('Zero-current readback failed')
 def set_current(self,value):
  with self.lock:
   if isinstance(value,bool) or not isinstance(value,(float,int)) or not math.isfinite(value) or not 0<=value<=1:raise ValueError('Buck profile permits 0 to 1.000 A only')
   self.write(f':SOURCE:CURRENT:LEVEL:IMMEDIATE {value:.6f}')
   if self.io and abs(float(self.query(':SOURCE:CURRENT:LEVEL:IMMEDIATE?'))-value)>0.002:raise RuntimeError('Current setpoint readback mismatch')
   self.current=value
 def on(self):
  with self.lock:
   if self.error:raise RuntimeError('Reconnect after resolving the fault: '+self.error)
   self.write(':SOURCE:INPUT:STATE ON')
   if self.io and self.query(':SOURCE:INPUT:STATE?')!='1':raise RuntimeError('Input ON readback failed')
   self.enabled=True;self.deadline=time.monotonic()+10
 def heartbeat(self):
  with self.lock:
   if self.enabled:self.deadline=time.monotonic()+10
 def off(self):
  with self.lock:
   self.write(':SOURCE:INPUT:STATE OFF')
   if self.io and self.query(':SOURCE:INPUT:STATE?')!='0':raise RuntimeError('Input OFF not confirmed')
   self.enabled=False;self.deadline=0
 def sample(self):
  with self.lock:
   if not self.connected:return None
   try:
    if self.enabled and time.monotonic()>self.deadline:self.off();self.error='Control lease expired; input disabled'
    if self.simulate:
     i=self.current if self.enabled else 0.;v=4.98-0.025*i;p=v*i
    else:
     v=float(self.query('MEASURE:VOLTAGE:DC?'));i=float(self.query('MEASURE:CURRENT:DC?'));p=float(self.query('MEASURE:POWER:DC?'))
     self.enabled=self.query(':SOURCE:INPUT:STATE?')=='1'
    if not all(math.isfinite(x) for x in [v,i,p]):raise RuntimeError('Invalid instrument measurement')
    if self.enabled and (v>5.5 or i>1.1 or p>6):self.off();self.error='Buck measurement limit exceeded; input disabled'
    row={'time':time.time(),'voltage_v':v,'current_a':i,'power_w':p,'setpoint_a':self.current,'input_on':self.enabled,'simulated':self.simulate}
    self.samples.append(row);self.samples=self.samples[-3600:];return row
   except Exception as e:
    self.error=str(e)
    try:self.off()
    except Exception:self.error+='; INPUT STATE UNKNOWN: disconnect DUT power locally'
    self.connected=False;return None
 def state(self):
  with self.lock:return {'connected':self.connected,'simulated':self.simulate,'identity':self.idn,'current_a':self.current,'input_on':self.enabled,'error':self.error,'measurement':self.samples[-1] if self.samples and self.connected else None,'limits':{'current_a':1,'voltage_v':5.5,'power_w':6},'cad_status':'Planning reconstruction; fit unverified'}