From 7ad408b6e291480778d6833c1a8eceec4a7de370 Mon Sep 17 00:00:00 2001 From: RanDarshan Date: Mon, 23 Jun 2025 20:30:23 -0400 Subject: [PATCH 1/5] Create directories for coldload agent and primecam bias system agent --- pcs/agents/coldload_scpipsu/__init__.py | 0 pcs/agents/coldload_scpipsu/agent.py | 222 ++++++++++++++++++++++++ pcs/agents/primecam_bias/__init__.py | 0 pcs/agents/primecam_bias/agent.py | 0 pcs/drivers/coldload.py | 12 ++ requirements.txt | 1 + 6 files changed, 235 insertions(+) create mode 100644 pcs/agents/coldload_scpipsu/__init__.py create mode 100644 pcs/agents/coldload_scpipsu/agent.py create mode 100644 pcs/agents/primecam_bias/__init__.py create mode 100644 pcs/agents/primecam_bias/agent.py create mode 100644 pcs/drivers/coldload.py diff --git a/pcs/agents/coldload_scpipsu/__init__.py b/pcs/agents/coldload_scpipsu/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/pcs/agents/coldload_scpipsu/agent.py b/pcs/agents/coldload_scpipsu/agent.py new file mode 100644 index 0000000..add12db --- /dev/null +++ b/pcs/agents/coldload_scpipsu/agent.py @@ -0,0 +1,222 @@ +import argparse +import time +import numpy as np +import os +from collections import deque + +import txaio +from ocs import ocs_agent, site_config +from ocs.ocs_twisted import TimeoutLock +from socs.agents.scpi_psu.agent import ScpiPsuAgent + + +class ColdloadAgent_ScpiPsu(ScpiPsuAgent): + def __init__(self, agent, ip_address, gpib_slot=None, port=None, ls240=None, channel=None, max_current=None): + super().__init__(agent, ip_address, gpib_slot=gpib_slot, port=port) # Initialize ScipPsuAgent + self.max_current = max_current + + self.channel = channel + self.ls240_channel = ls240[1] + try: + self.ls240 = OCSClient(ls240[0]) # Create LS240 client for grabbing coldload temperature data + except Exception as e: + self.log.error(f'Could not connect to LS240 for temperature monitoring: {e}') + + self.temp_control = False + self.agent.register_feed('pid_output', + record=True, + agg_params={'frame_length':10*60}, + buffer_time=5) + + + #==================# + # Coldload Methods # + #==================# + + def get_temp(self, session, params): + '''get_temp() + + **Task** - Get the current coldload temperature. + + ''' + with self.lock.acquire_timeout(timeout=5, job='get_temp') as acquired: + if not acquired: + self.log.error(f'Lock could not be acquired because it is held by {self.lock.job}.') + return False, 'Could not acquire lock.' + acq_status = self.ls240.acq.status().session + if acq_status['op_code'] == 3: + try: + temp = acq_status['data']['fields'][self.ls240_channel]['T'] + except KeyError as e: + self.log.error(f'Specified LS240 channel is not valid: {e}') + return False, f'Failed to get coldload temperature.' + + session.data['temperature'] = temp + else: + self.log.error('LS240 temperature monitoring is not running.') + return False, f'Failed to get coldload temperature.' + return True, temp + + @ocs_agent.param('temp', type=float, check=lambda x: 60 <= x <= 120) + @ocs_agent.param('sample_interval', type=float, default = 0.2) + @ocs_agent.param('average_interval', type=float, default = 60) + @ocs_agent.param('lock_interval', type=float, default=0.1) + @ocs_agent.param('timeout', type=float, default=120) + @ocs_agent.param('max_current', type=float, default=None) + @ocs_agent.param('prop', type=float, default = 0) + @ocs_agent.param('int', type=float, default=0) + @ocs_agent.param('der', type=float, default=0) + def set_temp(self, session, params): + ''' + Set the coldload temperature using a PID controller. + + ''' + target = params['temp'] + max_current = self.max_current if params['max_current'] is None else params['max_current'] + p = params['prop'] + i = params['int'] + d = params['der'] + + num_points = int(params['average_interval']/params['sample_interval']) + errs = deque(num_points*[0]) + dts = deque((num_points-1)*[1]) + integral = 0.0 + der = 0.0 + + # Calcualte I^2 so that the control variable is proportional to power (which is roughly linear with temperature) + power = self.psu.get_curr(self.channel)**2 + + start_time = time.time() + last_release = time.time() + last_sample = time.time() + self.temp_control = True + + while self.temp_control and not (time.time() - start_time > params['timeout']*60): + dt = time.time() - last_sample + if dt > params['sample_interval']: + success, temp = self.get_temp(session, params = None) + last_sample = time.time() + + with self.lock.acquire_timeout(timeout=1, job='set_temp') as acquired: + if not acquired: + self.log.error(f"Lock could not be acquired because it is held by {self.lock.job}.") + return False + + err = float(temp) - target + errs.append(err) + errs.popleft() + + dts.append(dt) + dts.popleft() + + integral += err * dt + if power == 0: integral = 0.0 + + der = np.mean([diff/t for diff, t in zip(np.diff(errs), dts)]) + + power += p*errs[-1] + i*integral + d*der + current = np.sqrt(current) + current = max(min(current, max_current), 0) + self.psu.set_curr(self.channel, current) + + pids = {'target_temp': float(target),'current': float(current), 'error': float(err), 'integral': float(integral), 'derivative': float(der)} + + data = {'timestamp': time.time(), + 'block_name': 'pid', + 'data': pids} + + self.agent.publish_to_feed('pid_output', data) + session.data = data + + # Release and reacquire the lock every ~0.1 second + if time.time() - last_release > params['lock_interval']: + last_release = time.time() + if not self.lock.release_and_acquire(timeout=120): + self.log.error(f'Could not re-acquire lock now held by {self.lock.job}.') + return False, 'Could not re-acquire lock.' + + return True, 'set_temp executed successfully.' + + def stop_set_temp(self, session, params): + if self.temp_control: + self.temp_control = False + return True, 'Stopping setting coldload temperature...' + else: + return False, 'Not currently setting coldload temperature.' + + #===============================# + # Overload ScpiPsuAgent Methods # + #===============================# + + def get_voltage(self, session, params): + params['channel'] = self.channel + return super().get_voltage(session, params=params) + + def get_current(self, session, params): + params['channel'] = self.channel + return super().get_current(session, params=params) + + @ocs_agent.param('volts', type=float, check=lambda x: 0 <= x <= 30) + def set_voltage(self, session, params): + params['channel'] = self.channel + return super().set_voltage(session, params=params) + + @ocs_agent.param('current', type=float) + def set_current(self, session, params): + params['channel'] = self.channel + params['current'] = max(min(params['current'], self.max_current), 0) + return super().set_current(session, params=params) + +def make_parser(parser=None): + """Build the argument parser for the Agent. Allows sphinx to automatically + build documentation based on this function. + + """ + if parser is None: + parser = argparse.ArgumentParser() + + # Add options specific to this agent. + pgroup = parser.add_argument_group('Agent Options') + pgroup.add_argument('--ip-address') + pgroup.add_argument('--gpib-slot') + pgroup.add_argument('--port') + pgroup.add_argument('--channel', type=int, help='The power supply channel connected to the coldload.') + pgroup.add_argument('--ls240', nargs=2, type=str, help='Instance ID of LS240 agent and thermometer channel of the coldload.', metavar = ('LS240 Agent Instance ID', 'Coldload Channel')) + pgroup.add_argument('--mode', type=str, default='acq', + choices=['init', 'acq']) + pgroup.add_argument('--max_current', type=float, default=0.6, help='Maximum current limit in amps.') + + return parser + +def main(args=None): + # Start logging + txaio.start_logging(level=os.environ.get("LOGLEVEL", "info")) + + parser = make_parser() + args = site_config.parse_args(agent_class='ColdloadAgent', + parser=parser, + args=args) + init_params = False + if args.mode == 'acq': + init_params = {'auto_acquire': True} + agent, runner = ocs_agent.init_site_agent(args) + + c = ColdloadAgent(agent, args.ip_address, gpib_slot=args.gpib_slot, port=args.port, channel = args.channel, ls240 = args.ls240, max_current=args.max_current) + + agent.register_task('init', c.init, startup=init_params) + agent.register_task('set_voltage', c.set_voltage) + agent.register_task('set_current', c.set_current) + agent.register_task('set_output', c.set_output) + + agent.register_task('get_voltage', c.get_voltage) + agent.register_task('get_current', c.get_current) + agent.register_task('get_temp', c.get_temp) + agent.register_task('get_output', c.get_output) + + agent.register_process('monitor_output', c.monitor_output, c.stop_monitoring) + agent.register_process('set_temp', c.set_temp, c.stop_set_temp) + + runner.run(agent, auto_reconnect=True) + +if __name__ == '__main__': + main() diff --git a/pcs/agents/primecam_bias/__init__.py b/pcs/agents/primecam_bias/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/pcs/agents/primecam_bias/agent.py b/pcs/agents/primecam_bias/agent.py new file mode 100644 index 0000000..e69de29 diff --git a/pcs/drivers/coldload.py b/pcs/drivers/coldload.py new file mode 100644 index 0000000..f002a6a --- /dev/null +++ b/pcs/drivers/coldload.py @@ -0,0 +1,12 @@ +from ocs.ocs_client import OCSClient + +class Coldload: + + def __init__(self, ls240, channel): + pass + + def get_temp(): + pass + + def set_temp(): + pass \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index ac9194e..1daa0e7 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,6 +1,7 @@ # core dependencies autobahn[serialization] ocs +socs==0.5.2 sqlalchemy>=1.4 twisted From 372ce18250c1bfb4a86d93907bfdbf2f9a3fd50b Mon Sep 17 00:00:00 2001 From: RanDarshan Date: Wed, 25 Jun 2025 10:40:15 -0400 Subject: [PATCH 2/5] Finish writing coldload agent --- pcs/agents/beam_mapper/__init__.py | 0 pcs/agents/beam_mapper/agent.py | 0 pcs/agents/coldload_scpipsu/agent.py | 282 +++++++++++++++++---------- pcs/drivers/coldload.py | 154 ++++++++++++++- pcs/plugin.py | 5 +- 5 files changed, 326 insertions(+), 115 deletions(-) create mode 100644 pcs/agents/beam_mapper/__init__.py create mode 100644 pcs/agents/beam_mapper/agent.py diff --git a/pcs/agents/beam_mapper/__init__.py b/pcs/agents/beam_mapper/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/pcs/agents/beam_mapper/agent.py b/pcs/agents/beam_mapper/agent.py new file mode 100644 index 0000000..e69de29 diff --git a/pcs/agents/coldload_scpipsu/agent.py b/pcs/agents/coldload_scpipsu/agent.py index add12db..6968260 100644 --- a/pcs/agents/coldload_scpipsu/agent.py +++ b/pcs/agents/coldload_scpipsu/agent.py @@ -9,26 +9,29 @@ from ocs.ocs_twisted import TimeoutLock from socs.agents.scpi_psu.agent import ScpiPsuAgent +from pcs.drivers.coldload import Coldload + class ColdloadAgent_ScpiPsu(ScpiPsuAgent): - def __init__(self, agent, ip_address, gpib_slot=None, port=None, ls240=None, channel=None, max_current=None): - super().__init__(agent, ip_address, gpib_slot=gpib_slot, port=port) # Initialize ScipPsuAgent - self.max_current = max_current + def __init__(self, agent, ip_address, gpib_slot=None, port=None, lakeshore=None, psu_channel=None, max_current=None): + # Initialize ScpiPsuAgent + super().__init__(agent, ip_address, gpib_slot=gpib_slot, port=port) - self.channel = channel - self.ls240_channel = ls240[1] - try: - self.ls240 = OCSClient(ls240[0]) # Create LS240 client for grabbing coldload temperature data - except Exception as e: - self.log.error(f'Could not connect to LS240 for temperature monitoring: {e}') - + # Define additional attributes + self.psu_channel = psu_channel self.temp_control = False + self.max_current = max_current + + # Create coldload object + self.cl = Coldload(lakeshore[0], lakeshore[1], ext_log=self.log) + self.err_i = 0.0 # Store integral error of set_temp PID in case control loop is interrupted + + # Register OCS feed to log PID temperature control parameters self.agent.register_feed('pid_output', record=True, agg_params={'frame_length':10*60}, buffer_time=5) - #==================# # Coldload Methods # #==================# @@ -37,107 +40,90 @@ def get_temp(self, session, params): '''get_temp() **Task** - Get the current coldload temperature. - + ''' with self.lock.acquire_timeout(timeout=5, job='get_temp') as acquired: if not acquired: self.log.error(f'Lock could not be acquired because it is held by {self.lock.job}.') return False, 'Could not acquire lock.' - acq_status = self.ls240.acq.status().session - if acq_status['op_code'] == 3: - try: - temp = acq_status['data']['fields'][self.ls240_channel]['T'] - except KeyError as e: - self.log.error(f'Specified LS240 channel is not valid: {e}') - return False, f'Failed to get coldload temperature.' - session.data['temperature'] = temp - else: - self.log.error('LS240 temperature monitoring is not running.') - return False, f'Failed to get coldload temperature.' - return True, temp + temp = self.cl.get_temp() + data = {'timestamp': time.time(), + 'block_name': 'coldload', + 'data': {'temp': temp}} + session.data = data + + return temp is not None, temp @ocs_agent.param('temp', type=float, check=lambda x: 60 <= x <= 120) - @ocs_agent.param('sample_interval', type=float, default = 0.2) - @ocs_agent.param('average_interval', type=float, default = 60) - @ocs_agent.param('lock_interval', type=float, default=0.1) - @ocs_agent.param('timeout', type=float, default=120) + @ocs_agent.param('sample_int', type=float, default = 0.5) + @ocs_agent.param('avg_int', type=float, default = 7.5) + @ocs_agent.param('lock_int', type=float, default=0.1) + @ocs_agent.param('timeout', type=float, default=180) @ocs_agent.param('max_current', type=float, default=None) - @ocs_agent.param('prop', type=float, default = 0) - @ocs_agent.param('int', type=float, default=0) - @ocs_agent.param('der', type=float, default=0) + @ocs_agent.param('pid', type=list, default=[2.25e-3, 5.1e-7, 0.71]) def set_temp(self, session, params): - ''' - Set the coldload temperature using a PID controller. - - ''' - target = params['temp'] - max_current = self.max_current if params['max_current'] is None else params['max_current'] - p = params['prop'] - i = params['int'] - d = params['der'] - - num_points = int(params['average_interval']/params['sample_interval']) - errs = deque(num_points*[0]) - dts = deque((num_points-1)*[1]) - integral = 0.0 - der = 0.0 - - # Calcualte I^2 so that the control variable is proportional to power (which is roughly linear with temperature) - power = self.psu.get_curr(self.channel)**2 - - start_time = time.time() - last_release = time.time() - last_sample = time.time() - self.temp_control = True - - while self.temp_control and not (time.time() - start_time > params['timeout']*60): - dt = time.time() - last_sample - if dt > params['sample_interval']: - success, temp = self.get_temp(session, params = None) - last_sample = time.time() - - with self.lock.acquire_timeout(timeout=1, job='set_temp') as acquired: - if not acquired: - self.log.error(f"Lock could not be acquired because it is held by {self.lock.job}.") - return False - - err = float(temp) - target - errs.append(err) - errs.popleft() - - dts.append(dt) - dts.popleft() - - integral += err * dt - if power == 0: integral = 0.0 - - der = np.mean([diff/t for diff, t in zip(np.diff(errs), dts)]) - - power += p*errs[-1] + i*integral + d*der - current = np.sqrt(current) - current = max(min(current, max_current), 0) - self.psu.set_curr(self.channel, current) - - pids = {'target_temp': float(target),'current': float(current), 'error': float(err), 'integral': float(integral), 'derivative': float(der)} - - data = {'timestamp': time.time(), - 'block_name': 'pid', - 'data': pids} - - self.agent.publish_to_feed('pid_output', data) - session.data = data - - # Release and reacquire the lock every ~0.1 second - if time.time() - last_release > params['lock_interval']: + """ + **Process** - Set the temperature of the coldload using a proportional integral derivative (PID) controller. + The PID controller uses the coldload temperature as the process variable and the current squared as the control variable. + + Parameters: + temp (float): Temperature to set coldload to + sample_int (float): Interval at which to sample coldload temperature + avg_int (float): Interval over which to average coldload temperatures (averaged temperature used as PID process variable). Also sets timescale for PID control + lock_int (float): Interval at which to release lock + timeout (float): Time in minutes after which to exit PID loop (0 for indefinite) + max_current (float): Maximum current limit + pid (List[float]): Proportional, integral, and derivative control coefficients + """ + + temp = params.pop('temp') + + lock_int = params.pop('lock_int') + if params['max_current'] is None: params['max_current'] = self.max_current + + with self.lock.acquire_timeout(timeout=1, job='set_temp') as acquired: + if not acquired: + self.log.error(f"Lock could not be acquired because it is held by {self.lock.job}.") + return False + + + last_release = time.time() + curr_args = [self.psu_channel] + params['yield_dict'] = True + params['err_i'] = self.err_i + pid_control = self.cl.set_temp(temp, self.psu.get_curr, self.psu.set_curr, *curr_args, **params) + self.temp_control = True + while self.temp_control: + # Perform PID control loop and get PID error values + try: + pids = next(pid_control) + # Create data dictionary and publish to pid_output feed and session.data + if pids is not None: + data = {'timestamp': time.time(), + 'block_name': 'coldload', + 'data': pids} + self.agent.publish_to_feed('pid_output', data) + session.data = data + self.err_i = pids['err_i'] + + # Release and reacquire the lock + if time.time() - last_release > lock_int: last_release = time.time() if not self.lock.release_and_acquire(timeout=120): self.log.error(f'Could not re-acquire lock now held by {self.lock.job}.') return False, 'Could not re-acquire lock.' - + # Catch exception raised if set_temp timeout is reached + except StopIteration: + self.temp_control = False return True, 'set_temp executed successfully.' def stop_set_temp(self, session, params): + """stop_set_temp() + + **Process** - Stop the process setting the coldload temperature. Called when running set_temp.stop() + + """ if self.temp_control: self.temp_control = False return True, 'Stopping setting coldload temperature...' @@ -149,29 +135,106 @@ def stop_set_temp(self, session, params): #===============================# def get_voltage(self, session, params): - params['channel'] = self.channel + """get_voltage() + + **Task** - Get the voltaeg of the coldload. + + """ + params['channel'] = self.psu_channel return super().get_voltage(session, params=params) def get_current(self, session, params): - params['channel'] = self.channel + """get_current() + + **Task** - Get the current of the coldload. + + """ + params['channel'] = self.psu_channel return super().get_current(session, params=params) @ocs_agent.param('volts', type=float, check=lambda x: 0 <= x <= 30) def set_voltage(self, session, params): - params['channel'] = self.channel + """set_voltage(volts) + + **Task** - Set the voltage of the coldload. + + Parameters: + volts (float): Voltage to set. + """ + + params['channel'] = self.psu_channel return super().set_voltage(session, params=params) @ocs_agent.param('current', type=float) def set_current(self, session, params): - params['channel'] = self.channel + """set_current(current) + + **Task** - Set the current of the coldload. + + Parameters: + current (float): Current to set. + """ + + # Override set_current method to use power supply channel connected to coldload and to limit the max current. + params['channel'] = self.psu_channel params['current'] = max(min(params['current'], self.max_current), 0) return super().set_current(session, params=params) + #============================================# + # Read/Write Serial Commands to Power Supply # + #============================================# + + def read(self, session, params): + """read() + + **Task** - Read message from power supply + + """ + + with self.lock.acquire_timeout(timeout=5, job='read') as acquired: + if not acquired: + self.log.error(f'Lock could not be acquired because it is held by {self.lock.job}.') + return False, 'Could not acquire lock.' + + resp = self.psu.read() + data = {'timestamp': time.time(), + 'block_name': 'power_supply', + 'data': {'read': resp}} + session.data = data + return True, resp + + @ocs_agent.param('msg', type=str, default='') + def write(self, session, params): + """write(msg) + + **Task** - Write serial command to power supply. + + Parameters: + msg (str): Serial command + """ + with self.lock.acquire_timeout(timeout=5, job='write') as acquired: + if not acquired: + self.log.error(f'Lock could not be acquired because it is held by {self.lock.job}.') + return False, 'Could not acquire lock.' + + msg = params['msg'] + if not msg: + return False, f"Invalid message: {msg}" + else: + self.psu.write(msg) + return True, f"Wrote message to power supply." + +#===========# +# Functions # +#===========# + def make_parser(parser=None): """Build the argument parser for the Agent. Allows sphinx to automatically build documentation based on this function. """ + # From simonsobs/socs/socs/agents/scpi_psu/agent.py with additions + if parser is None: parser = argparse.ArgumentParser() @@ -180,15 +243,16 @@ def make_parser(parser=None): pgroup.add_argument('--ip-address') pgroup.add_argument('--gpib-slot') pgroup.add_argument('--port') - pgroup.add_argument('--channel', type=int, help='The power supply channel connected to the coldload.') - pgroup.add_argument('--ls240', nargs=2, type=str, help='Instance ID of LS240 agent and thermometer channel of the coldload.', metavar = ('LS240 Agent Instance ID', 'Coldload Channel')) - pgroup.add_argument('--mode', type=str, default='acq', + pgroup.add_argument('--psu-channel', type=int, help='The power supply channel connected to the coldload.') + pgroup.add_argument('--lakeshore', nargs=2, type=str, help='Instance ID of lakeshore agent and thermometer channel of the coldload.', metavar = ('Lakeshore Agent Instance ID', 'Lakeshore Thermometer Channel of Coldload')) + pgroup.add_argument('--mode', type=str, default='init', choices=['init', 'acq']) - pgroup.add_argument('--max_current', type=float, default=0.6, help='Maximum current limit in amps.') - + pgroup.add_argument('--max-current', type=float, default=1, help='Maximum current limit in Amperes.') return parser def main(args=None): + # From simonsobs/socs/socs/agents/scpi_psu/agent.py with modifications + # Start logging txaio.start_logging(level=os.environ.get("LOGLEVEL", "info")) @@ -196,12 +260,11 @@ def main(args=None): args = site_config.parse_args(agent_class='ColdloadAgent', parser=parser, args=args) - init_params = False - if args.mode == 'acq': - init_params = {'auto_acquire': True} + + init_params = {'auto_acquire': args.mode == 'acq'} agent, runner = ocs_agent.init_site_agent(args) - c = ColdloadAgent(agent, args.ip_address, gpib_slot=args.gpib_slot, port=args.port, channel = args.channel, ls240 = args.ls240, max_current=args.max_current) + c = ColdloadAgent_ScpiPsu(agent, args.ip_address, gpib_slot=args.gpib_slot, port=args.port, psu_channel = args.psu_channel, lakeshore = args.lakeshore, max_current=args.max_current) agent.register_task('init', c.init, startup=init_params) agent.register_task('set_voltage', c.set_voltage) @@ -213,6 +276,9 @@ def main(args=None): agent.register_task('get_temp', c.get_temp) agent.register_task('get_output', c.get_output) + agent.register_task('read', c.read) + agent.register_task('write', c.write) + agent.register_process('monitor_output', c.monitor_output, c.stop_monitoring) agent.register_process('set_temp', c.set_temp, c.stop_set_temp) diff --git a/pcs/drivers/coldload.py b/pcs/drivers/coldload.py index f002a6a..6ef1b64 100644 --- a/pcs/drivers/coldload.py +++ b/pcs/drivers/coldload.py @@ -1,12 +1,154 @@ from ocs.ocs_client import OCSClient +import time +import numpy as np class Coldload: - def __init__(self, ls240, channel): - pass + def __init__(self, lakeshore, ls_channel, ext_log = None): + self.ls_channel = ls_channel + self.ext_log = ext_log + + # Create Lakeshore client for grabbing coldload temperature data + try: + self.lakeshore = OCSClient(lakeshore, args=[]) + except Exception as e: + self.logger(f'Could not connect to Lakeshore agent \033[3m{lakeshore}\033[0m for temperature monitoring: {e}', 'error') - def get_temp(): - pass + def get_temp(self): + """ + Get the current temperature of the coldload. + """ + # Fetch most data from Lakeshore OCS feed + acq_status = self.lakeshore.acq.status().session - def set_temp(): - pass \ No newline at end of file + # Check to see if lakeshore is actively acquiring data + if acq_status['op_code'] == 3: + # Get coldload temperature data using specified channel + try: + temp = acq_status['data']['fields'][self.ls_channel]['T'] + except KeyError as e: + self.logger(f'Specified Lakeshore channel {self.ls_channel} is not valid: {e}', 'error') + temp = None + else: + self.logger('Lakeshore data acquisition is not running.', 'error') + temp = None + return temp + + def set_temp(self, temp, get_current, set_current, *args, **kwargs): + """ + Set the temperature of the coldload using a proportional integral derivative (PID) controller. + The PID controller uses the coldload temperature as the process variable and the current squared as the control variable. + + Parameters: + temp (float): Temperature to set coldload to + get_current: Function for getting current. Abstracted so that set_temp is compatible with different power supplies + set_current: Function for setting current. Should have "curr" argument as a keyword argument or as the last positional argument. Abstracted so that set_temp is compatible with different power supplies + args: Arguments for get_current and set_current functions + + kwargs: + sample_int (float): Interval at which to sample coldload temperature + avg_int (float): Interval over which to average coldload temperatures (averaged temperature used as PID process variable). Also sets timescale for PID control + timeout (float): Time in minutes after which to exit PID loop (0 for indefinite) + yield_dict (bool): Whether to yield error values and coldload current after each PID control loop + max_current (float): Maximum current limit + pid (List[float]): Proportional, integral, and derivative control coefficients + """ + + sample_int = 0.5 + default_avg_int = 7.5 + timeout = 180 + yield_dict = False + + max_current = 0.6 + pid = [2.25e-3, 5.1e-7, 0.71] + thresholds = [0.01, 0.1, 1, 5] + + err_p = temp - self.get_temp() + err_i = 0.0 + err_d = 0.0 + errs = [] + + for k, v in kwargs.items(): + if k == 'sample_int': + sample_int = v + elif k == 'avg_int': + default_avg_int = v + elif k == 'timeout': + timeout = v + elif k == 'yield_dict': + yield_dict = v + elif k == 'max_current': + max_current = v + elif k == 'pid': + pid = v + elif k == 'err_i': + err_i = v + + avg_int = default_avg_int + timeout *= 60 # Convert timeout to seconds + + # Get the current coldload current and use current squared as the control variable so that it is proportional to power (which is roughly linear with temperature) + curr_sq = get_current(*args)**2 + + start_time = time.time() + last_sample = start_time + last_pid = start_time + while timeout == 0 or time.time() - start_time < timeout: + curr_time = time.time() + if curr_time - last_sample > sample_int: + last_sample = curr_time + errs.append(temp - self.get_temp()) + + delta_t = curr_time - last_pid + if delta_t > avg_int: + last_pid = curr_time + + avg_err = np.mean(errs) # Average the error to reduce noise + errs = [] # Reset list of errors + + # Calculate the PID error values + err_d = (avg_err - err_p)/delta_t + err_i += avg_err * delta_t + err_p = avg_err + + # Vary avg_int depending on how small error is to reduce noise in derivative at small errors + avg_int = default_avg_int * (2 ** sum(err_p < threshold for threshold in thresholds)) + + # Set the integral error to zero if the current is already zero so that there is not a large accumulated error as the temperature decays slowly + if curr_sq == 0: err_i = 0.0 + + # Vary the current squared as specified by the PID controller + curr_sq += pid[0]*err_p + pid[1] * err_i + pid[2]*err_d + + # Convert to current and limit it to be between 0 and max_current + curr_sq = max(curr_sq, 0.0) + curr = round(min(np.sqrt(curr_sq), max_current), 3) # Round down to mA precision + + # Set the new current + set_current(*args, curr = curr) + + if yield_dict: + pids = {'target_temperature': float(temp),'current': float(curr), 'err_p': float(err_p), 'err_i': float(err_i), 'err_d': float(err_d)} + yield pids + time.sleep(0.1) # Wait to prevent wasting CPU resources + if yield_dict: yield None # Yield None on non-PID loops to prevent the method from blocking for avg_int seconds + + #=========# + # Logging # + #=========# + def logger(self, msg, level): + level = '_' + level + if not self.ext_log is None and hasattr(self, level): + method = getattr(self, level) + method(msg) + else: + print(msg) + + #==========================# + # Logging Internal Methods # + #==========================# + def _info(self, msg): + self.ext_log.info(msg) + + def _error(self, msg): + self.ext_log.error(msg) \ No newline at end of file diff --git a/pcs/plugin.py b/pcs/plugin.py index d6049ca..c09f0ef 100644 --- a/pcs/plugin.py +++ b/pcs/plugin.py @@ -3,5 +3,8 @@ 'LS325Agent': {'module': 'pcs.agents.lakeshore325.agent', 'entry_point': 'main'}, 'RaritanAgent': {'module': 'pcs.agents.raritan_pdu.agent', 'entry_point': 'main'}, 'ACUAgent': {'module': 'pcs.agents.acu_interface.agent', 'entry_point': 'main'}, - 'Bluefors_TC_Agent': {'module': 'pcs.agents.bluefors_tc.agent', 'entry_point': 'main'} + 'Bluefors_TC_Agent': {'module': 'pcs.agents.bluefors_tc.agent', 'entry_point': 'main'}, + 'ColdloadAgent_ScpiPsu': {'module': 'pcs.agents.coldload_scpipsu.agent', 'entry_point': 'main'}, + 'PrimecamBiasAgent': {'module': 'pcs.agents.primecam_bias.agent', 'entry_point': 'main'}, + 'BeamMapperAgent': {'module': 'pcs.agents.beam_mapper.agent', 'entry_point': 'main'} } From caf45a51b84b35a2e2e431cbebd5055b1788eb5b Mon Sep 17 00:00:00 2001 From: RanDarshan Date: Thu, 26 Jun 2025 15:37:42 -0400 Subject: [PATCH 3/5] Add documentation for coldload agent --- docs/agents/coldload_scpipsu.rst | 114 +++++++++++++++++++++++++++ pcs/agents/coldload_scpipsu/agent.py | 24 +++++- 2 files changed, 137 insertions(+), 1 deletion(-) create mode 100644 docs/agents/coldload_scpipsu.rst diff --git a/docs/agents/coldload_scpipsu.rst b/docs/agents/coldload_scpipsu.rst new file mode 100644 index 0000000..c1da277 --- /dev/null +++ b/docs/agents/coldload_scpipsu.rst @@ -0,0 +1,114 @@ +.. highlight:: rst + +.. _coldload_scpipsu: + +======== +Coldload +======== + +The Coldload agent controls and monitors the temperature of a coldload. The coldload therommeter is read out using a Lakeshore 240/372/etc. +through the associated Lakeshore agent. The temperature of the coldload is controlled by varying the current supplied through a Standard Commands for Programmable Instruments (SCPI) power supply unit. +Communication with the power supply unit can be done through direct Ethernet connection (if availabile) or can be mediated through GPIB (e.g, using a Prologix Interface). + +.. argparse:: + :filename: ../pcs/agents/coldload_scpipsu/agent.py + :func: add_agent_args + :prog: python3 agent.py + +Configuration File Examples +--------------------------- + +Below are configuration examples for the SO-OCS site config file +and docker-compose file for running the +Agent in a docker container. + +OCS Site Config +``````````````` + +To run the Coldload agent, a RaritanAgent block must be added +to the site config file. Here is an example configuration block with +all available arguments:: + + {'agent-class': 'ColdloadAgent_ScpiPsu', + 'instance-id': 'power-psu-coldload', + 'manage': 'docker', + 'arguments': [ + ['--ip-address', '10.10.10.50'], + ['--gpib-slot', '15'], + ['--psu-channel', 1], + ['--lakeshore', ['cryo-ls240-lsa291f', 'Channel_4']], + ['--max-current', 0.6] + ]}, + +The ``--ip-address`` argument should be changed to the IP address of the BK Precision power supply on the network. +The ``--gpib-slot`` argument should be changed to the GPIB port if using a Prologix Interface for communication. +The ``--port`` argument should be added if communicating through Ethernet directly. +The ``--psu-channel`` argument should be changed to the power supply channel connected to the coldload. +The ``--lakeshore`` argument should be a list with two elements: The instance-id of the Lakeshore agent and the coldload thermometer channel. +The ``-max-current`` argument specifies the maximum current that will be supplied by the power supply. The default is shown above. + +Docker Compose +`````````````` +The Coldload agent should be configured to run in a Docker container. An +example docker compose service configuration is shown here:: + + ocs-coldload: + image: ghcr.io/ccatobs/pcs:latest + <<: *log-options + hostname: ocs-docker + network_mode: "host" + environment: + - INSTANCE_ID=power-psu-coldload + - SITE_HUB=ws://192.168.24.55:8001/ws + - SITE_HTTP=http://192.168.24.55:8001/call + volumes: + - ${OCS_CONFIG_DIR}:/config:ro + +Description +----------- + +A "coldload" is an approximate blackbody that is used cryogenically as a calibration source. The coldload used for Mod-Cam is an aluminum plate coated with epoxy. +The temperature of the coldload is controlled by varying the current supplied by a BK Precision 9130B power supply through resistors mounted on the coldload. The +temperature is read out using a Lakeshore LS240. The Coldload agent subclasses the Simon's Observatory SOCS `ScpiPsu agent `_ for control over the power supply +but limits control to only the power supply channel connected to the coldload. Additionally, the Coldload agent subscribes to the Lakeshore agent feed monitoring the coldload temperature. +The Coldload agent's main functionality is controlling the power supply (turning the channel on/off and getting/setting the voltage/current), but subscribing to the temperature feed also allows +getting the temperature of the coldload (get_temp()) and setting the temperature of the coldload using a PID controller (set_temp()). When setting the temperature of the coldload, the PID values +(error, integral of error, and derivative of error) as well as the current are continously published to the 'pid_output' OCS feed for monitoring. Finally, the Coldload agent exposes the serial read(), write() +commands to allow for greater control over the power supply unit. + +Example Clients +--------------- + +Below is an example client to control outlets:: + + from ocs.ocs_client import OCSClient + client = OCSClient('power-psu-coldload') + + # Get channel output + client.get_output() + + # Set channel output + client.set_output(state='on') + client.set_output(state='off') + + # Get/set voltage + client.get_voltage() + client.set_voltage(volts=15) # Volts + + # Get/set current + client.get_current() + client.set_current(current=0.1) # Amps + + # Get/set temperature + client.get_temp() + client.set_temp(temp=65, max_current=1, pid=[2.25e-3, 5.1e-7, 0.71]) # Kelvin + + # Read/write serial commands + client.write(msg='*idn?') + client.read() + +Agent API +--------- + +.. autoclass:: pcs.agents.coldload_scpipsu.agent.ColdloadAgent_ScpiPsu + :members: \ No newline at end of file diff --git a/pcs/agents/coldload_scpipsu/agent.py b/pcs/agents/coldload_scpipsu/agent.py index 6968260..c3afedd 100644 --- a/pcs/agents/coldload_scpipsu/agent.py +++ b/pcs/agents/coldload_scpipsu/agent.py @@ -134,10 +134,19 @@ def stop_set_temp(self, session, params): # Overload ScpiPsuAgent Methods # #===============================# + def get_output(self, session, params): + """get_output() + + **Task** - Get whether the channel connected to the coldload is on or off. + + """ + params['channel'] = self.psu_channel + return super().get_output(session, params=params) + def get_voltage(self, session, params): """get_voltage() - **Task** - Get the voltaeg of the coldload. + **Task** - Get the voltage of the coldload. """ params['channel'] = self.psu_channel @@ -152,6 +161,19 @@ def get_current(self, session, params): params['channel'] = self.psu_channel return super().get_current(session, params=params) + @ocs_agent.param('state', type=bool) + def set_output(self, session, params): + """set_output(state) + + **Task** - Turn the channel connected to the coldload on or off. + + Parameters: + state (bool): True for on, False for off. + """ + + params['channel'] = self.psu_channel + return super().set_output(session, params=params) + @ocs_agent.param('volts', type=float, check=lambda x: 0 <= x <= 30) def set_voltage(self, session, params): """set_voltage(volts) From c0a5280d82086845f00c9f52ac134eb21a327147 Mon Sep 17 00:00:00 2001 From: RanDarshan Date: Wed, 2 Jul 2025 15:50:36 -0400 Subject: [PATCH 4/5] Add rfsoc-controller agent --- .gitignore | 4 + Dockerfile | 3 +- docker/rfsoc_controller/Dockerfile | 15 ++ pcs/agents/coldload_scpipsu/agent.py | 22 +- pcs/agents/rfsoc_controller/__init__.py | 0 pcs/agents/rfsoc_controller/agent.py | 315 ++++++++++++++++++++++++ pcs/drivers/coldload.py | 54 ++-- pcs/plugin.py | 3 +- 8 files changed, 376 insertions(+), 40 deletions(-) create mode 100644 docker/rfsoc_controller/Dockerfile create mode 100644 pcs/agents/rfsoc_controller/__init__.py create mode 100644 pcs/agents/rfsoc_controller/agent.py diff --git a/.gitignore b/.gitignore index 0f984e7..889e9b9 100644 --- a/.gitignore +++ b/.gitignore @@ -1,6 +1,10 @@ # dynamic package version pcs/_version.py +# Docker build dependencies + +docker/rfsoc_controller/ccatkidlib + # Byte-compiled / optimized / DLL files __pycache__/ *.py[cod] diff --git a/Dockerfile b/Dockerfile index e4ebbfa..2e72728 100644 --- a/Dockerfile +++ b/Dockerfile @@ -7,7 +7,8 @@ FROM simonsobs/ocs:v0.11.3-19-gd729e04 # Install addition network related packages for ACU interface agent RUN apt-get update -y && apt-get install -y iputils-ping \ - curl + curl \ + rsync # Copy in and install requirements COPY requirements.txt /app/pcs/requirements.txt diff --git a/docker/rfsoc_controller/Dockerfile b/docker/rfsoc_controller/Dockerfile new file mode 100644 index 0000000..2d052b5 --- /dev/null +++ b/docker/rfsoc_controller/Dockerfile @@ -0,0 +1,15 @@ +FROM pcs + +# Will eventually want to pip install ccatkidlib from github once made public +WORKDIR /app/pcs/ +COPY ./ccatkidlib /app/pcs/ccatkidlib + +RUN python -m pip install -e ./ccatkidlib && \ + python -m pip install -r ./ccatkidlib/requirements.txt && \ + python -m pip install numpy==2.0.2 + +WORKDIR / + +ENV OCS_CONFIG_DIR='/config' + +ENTRYPOINT ["dumb-init", "ocs-agent-cli"] \ No newline at end of file diff --git a/pcs/agents/coldload_scpipsu/agent.py b/pcs/agents/coldload_scpipsu/agent.py index c3afedd..a45f922 100644 --- a/pcs/agents/coldload_scpipsu/agent.py +++ b/pcs/agents/coldload_scpipsu/agent.py @@ -23,9 +23,10 @@ def __init__(self, agent, ip_address, gpib_slot=None, port=None, lakeshore=None, self.max_current = max_current # Create coldload object - self.cl = Coldload(lakeshore[0], lakeshore[1], ext_log=self.log) + self.cl = Coldload(lakeshore[0], lakeshore[1]) self.err_i = 0.0 # Store integral error of set_temp PID in case control loop is interrupted + self.log = agent.log # Register OCS feed to log PID temperature control parameters self.agent.register_feed('pid_output', record=True, @@ -45,7 +46,7 @@ def get_temp(self, session, params): with self.lock.acquire_timeout(timeout=5, job='get_temp') as acquired: if not acquired: self.log.error(f'Lock could not be acquired because it is held by {self.lock.job}.') - return False, 'Could not acquire lock.' + return False, f"Lock could not be acquired because it is held by {self.lock.job}." temp = self.cl.get_temp() data = {'timestamp': time.time(), @@ -58,10 +59,14 @@ def get_temp(self, session, params): @ocs_agent.param('temp', type=float, check=lambda x: 60 <= x <= 120) @ocs_agent.param('sample_int', type=float, default = 0.5) @ocs_agent.param('avg_int', type=float, default = 7.5) + @ocs_agent.param('thresholds', type=list, default=[0.01, 0.1, 1, 5]) @ocs_agent.param('lock_int', type=float, default=0.1) @ocs_agent.param('timeout', type=float, default=180) @ocs_agent.param('max_current', type=float, default=None) - @ocs_agent.param('pid', type=list, default=[2.25e-3, 5.1e-7, 0.71]) + @ocs_agent.param('pid', type=list, default=[1e-3, 1.75e-7, 0.8]) + @ocs_agent.param('int_threshold', type=float, default=0.1) + @ocs_agent.param('reset_int', type=bool, default=True) + @ocs_agent.param('reset_current', type=bool, default=False) def set_temp(self, session, params): """ **Process** - Set the temperature of the coldload using a proportional integral derivative (PID) controller. @@ -71,10 +76,12 @@ def set_temp(self, session, params): temp (float): Temperature to set coldload to sample_int (float): Interval at which to sample coldload temperature avg_int (float): Interval over which to average coldload temperatures (averaged temperature used as PID process variable). Also sets timescale for PID control + thresholds (List(float)): Error thresholds at which to modify avg_int. avg_int will be used for errors greater than the largest threshold and then multiplied by 2 for each threshold passed. lock_int (float): Interval at which to release lock timeout (float): Time in minutes after which to exit PID loop (0 for indefinite) max_current (float): Maximum current limit pid (List[float]): Proportional, integral, and derivative control coefficients + int_threshold (float): Error threshold hold after which the integral term will start contributing to the PID control. """ temp = params.pop('temp') @@ -85,13 +92,13 @@ def set_temp(self, session, params): with self.lock.acquire_timeout(timeout=1, job='set_temp') as acquired: if not acquired: self.log.error(f"Lock could not be acquired because it is held by {self.lock.job}.") - return False + return False, f"Lock could not be acquired because it is held by {self.lock.job}." last_release = time.time() curr_args = [self.psu_channel] params['yield_dict'] = True - params['err_i'] = self.err_i + params['err_i'] = 0.0 if params['reset_int'] else self.err_i pid_control = self.cl.set_temp(temp, self.psu.get_curr, self.psu.set_curr, *curr_args, **params) self.temp_control = True while self.temp_control: @@ -116,6 +123,7 @@ def set_temp(self, session, params): # Catch exception raised if set_temp timeout is reached except StopIteration: self.temp_control = False + if params['reset_current']: self.psu.set_curr(self.psu_channel, 0.0) return True, 'set_temp executed successfully.' def stop_set_temp(self, session, params): @@ -216,7 +224,7 @@ def read(self, session, params): with self.lock.acquire_timeout(timeout=5, job='read') as acquired: if not acquired: self.log.error(f'Lock could not be acquired because it is held by {self.lock.job}.') - return False, 'Could not acquire lock.' + return False, f"Lock could not be acquired because it is held by {self.lock.job}." resp = self.psu.read() data = {'timestamp': time.time(), @@ -237,7 +245,7 @@ def write(self, session, params): with self.lock.acquire_timeout(timeout=5, job='write') as acquired: if not acquired: self.log.error(f'Lock could not be acquired because it is held by {self.lock.job}.') - return False, 'Could not acquire lock.' + return False, f"Lock could not be acquired because it is held by {self.lock.job}." msg = params['msg'] if not msg: diff --git a/pcs/agents/rfsoc_controller/__init__.py b/pcs/agents/rfsoc_controller/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/pcs/agents/rfsoc_controller/agent.py b/pcs/agents/rfsoc_controller/agent.py new file mode 100644 index 0000000..478ef0e --- /dev/null +++ b/pcs/agents/rfsoc_controller/agent.py @@ -0,0 +1,315 @@ +import os +import sys +import time + +import argparse +from pathlib import Path +from functools import wraps +from ocs import ocs_agent, site_config +from ocs.ocs_twisted import TimeoutLock +import txaio + +# Import Twisted Modules for ccatkidlib python scripts +from autobahn.twisted.util import sleep as dsleep +from twisted.internet import protocol, reactor +from twisted.internet.defer import Deferred, inlineCallbacks +from twisted.python.failure import Failure +from typing import Optional + +# ccatkidlib Imports +from ccatkidlib.rfsoc.rfsoc_daq import R +import ccatkidlib.rfsoc_io as rfsoc_io + +class CCATKIDlibScriptProtocol(protocol.ProcessProtocol): + def __init__(self, script, log=None): + self.script = Path(script) + self.log = log + self.end_status: Optional[Failure] = None + + def connectionMade(self): + """Called when process is started""" + self.transport.closeStdin() + + def outReceived(self, data): + """Called whenever data is received through stdout""" + if self.log: self.log.info(f"{self.script.name} | {data.strip().decode('utf-8')}") + + def errReceived(self, data): + """Called whenever data is received through stderr""" + self.log.error(data) + + def processExited(self, status: Failure): + """Called when process has exited.""" + + exit_code = status.value.exitCode + if self.log: self.log.info(f"{self.script.name} | Process exited code {exit_code}.") + + self.deferred.callback(exit_code) + +class RFSoController: + ''' + PCS Agent for controlling RFSoCs through ccatkidlib scripts and methods. + Modelled after SOCS PysmurfController with modifications. + ''' + def __init__(self, agent, config: str = None, module: str = None): + ''' + Constructor for RfsocController. + ''' + + # Create OCS agent and get log + self.agent = agent + self.ocs_session = None + self.log = agent.log + + self.lock = TimeoutLock() # Create lock + + cfg_file = Path(os.environ['OCS_CONFIG_DIR']) / config + try: + self.control_cfg = rfsoc_io.load_config(cfg_file) + except AssertionError: + self.log.error(f'Could not find rfsoc-controller config file {cfg_file}.') + raise FileNotFoundError + + self.sys_cfg_path = self.control_cfg['modules'][module]['system_config'] + + self._new_session(init_boards=True) + + self.prot = None + + #=================# + # Control Methods # + #=================# + + @ocs_agent.param('init_boards', type = bool, default = False) + def new_session(self, session, params): + RC = self._new_session(init_boards=params['init_boards']) + return True, f'Succesfully created new session: {self.session}' + + def _new_session(self, init_boards): + RC = R(cfg_path = self.sys_cfg_path, init_boards = init_boards, init_drones = True) # Instantiate RFSoC control object with full board and drone setup + self._update_control(RC) + self.log.info(f'Succesfully created new session: {self.session}') + return RC + + @staticmethod + def _get_control(func): + @wraps(func) + def _wrapper(self, session, params): + RC = R(cfg_path = self.sys_cfg_path, initialize_boards = False, initialize_drones = False, + sess_id = self.session, measurement_name = self.measurement_name, measurement_desc = self.measurement_desc, curr_date = self.curr_date) + + RC.NCLOs = self.NCLOs + RC.drive_attens = self.drive_attens + RC.sense_attens = self.sense_attens + + RC.set_NCLO(setup=False) + RC.set_atten(setup=False) + + params['R'] = RC + + rtn = func(self, session, params) + + self._update_control(RC) + + return rtn + return _wrapper + + def _update_control(self, RC): + ''' + Get the current state of the control object. + ''' + + # Create attributes to save system state of control object across recreations + # --------------------------------------------------------------------------- + # Get the session ID, name, and description of measurement + self.session = RC.sess_id + self.curr_date = RC.curr_date + self.measurement_name = RC.measurement_name + self.measurement_desc = RC.measurement_desc + + # Get the current NCLOs and attenuations + self.NCLOs = RC.NCLOs + self.drive_attens = RC.drive_attens + self.sense_attens = RC.sense_attens + + #================# + # Script Methods # + #================# + + @inlineCallbacks + def _run_script(self, session, script, args): + """ + Runs a ccatkidlib control script using the Twisted reactor. + Modified _run_script method of SOCS PysmurfController + + Args: + script (string): + path to the script you wish to run + args (list, optional): + List of command line arguments to pass to the script. + Defaults to []. + log (string or bool, optional): + Determines if and how the process's stdout should be logged. + You can pass the path to a logfile, True to use the agent's log, + or False to not log at all. + """ + + with self.lock.acquire_timeout(5, job=script) as acquired: + if not acquired: + self.log.error(f"The requested script cannot be run because the lock is held by {self.lock.job}") + return False, f"The requested script cannot be run because lock is held by {self.lock.job}" + self.ocs_session = session + try: + self.prot = CCATKIDlibScriptProtocol(script, log=self.log) + self.prot.deferred = Deferred() + python_exec = sys.executable + + cmd = [python_exec, '-u', script] + list(map(str, args)) + + self.log.info(f"Running Script: {' '.join(cmd)}") + + reactor.spawnProcess(self.prot, python_exec, cmd, env=os.environ) + + exit_code = yield self.prot.deferred + + return exit_code == 0, f"Script has finished with exit code {exit_code}" + + finally: + # Sleep to allow any remaining messages to be put into the + # session var + yield dsleep(1.0) + self.ocs_session = None + + @inlineCallbacks + def run(self, session, params=None): + status, msg = yield self._run_script(session, params['script'], params.get('args', [])) + + # Set stored NCLO and attenuations to None since their state may have changed during script execution + self.NCLOs = None + self.drive_attens = None + self.sense_attens = None + + self._new_session(init_boards=False) + + return status, msg + + def abort(self, session, params=None): + """abort() + + **Task** - Aborts the actively running script. + + """ + self.prot.transport.signalProcess('KILL') + return True, "Aborting process" + + #==================# + # Main DAQ Methods # + #==================# + + def tune(): + return + + @_get_control + @ocs_agent.param('com_to', type=list, default=[]) + @ocs_agent.param('time', type=float) + def take_timestream(self, session, params): + return + + #===============# + # Sweep Methods # + #===============# + + @_get_control + @ocs_agent.param('R', type=R, default=None) + @ocs_agent.param('com_to', type=list, default=None) + @ocs_agent.param('write_comb', type=bool, default=None) + @ocs_agent.param('sweep_steps', type=int, default=None) + @ocs_agent.param('parallel_boards', type=int, default=None) + @ocs_agent.param('parallel_drones', type=int, default=None) + def take_vna_sweep(self, session, params): + with self.lock.acquire_timeout(5, job='vna_sweep') as acquired: + if not acquired: + self.log.error(f"Could not acquire lock because it is held by {self.lock.job}.") + return False, f"Could not acquire lock because it is held by {self.lock.job}." + params = self._filter_params(params) + RC = params.pop('R') + data = RC.take_vna_sweep(**params) + data = list(map(str, data)) + self._publish_data(data, RC, params, session) + return True, 'Successfully finished taking VNA sweep.' + + @_get_control + @ocs_agent.param('com_to', type=list, default=[]) + def take_target_sweep(self, session, params): + return + + @_get_control + @ocs_agent.param('com_to', type=list, default=[]) + def find_detectors(self, session, params): + return + + @_get_control + @ocs_agent.param('com_to', type=list, default=[]) + def find_detectors_fine(self, session, params): + return + + #=======# + # Other # + #=======# + + @ocs_agent.param('threshold', type=float, default=5) + def monitor_space(self, session, params): + '''monitor_space.start() + + **Process** - Monitor storage space of RFSoC boards and clean files as necessary. + ''' + + return + + #================# + # Helper Methods # + #================# + def _filter_params(self, params): + return {k:v for k, v in params.items() if v is not None} + + def _publish_data(self, data, RC, params, session): + com_to = params['com_to'] if 'com_to' in params else RC.drone_list + data_dict = {'name': RC.measurement_name, + 'date': RC.curr_date, + 'session': RC.sess_id, + 'com_to': com_to, + 'data': data} + session.data['data'] = data_dict + +def make_parser(parser=None): + ''' + Build ArgumentParser for passing arguments through OCS_CONFIG file + ''' + if parser is None: + parser = argparse.ArgumentParser() + + pgroup = parser.add_argument_group('Agent Options') + pgroup.add_argument('--config', type=str, default='controller_config.yaml', + help='Path to rfsoc-controller config relative to OCS_CONFIG_DIR') + pgroup.add_argument('--module', type=str, choices=['280GHz', '350GHz', '850GHz', 'EoR_Spec'], + help='Which instrument module to control with rfsoc-controller.') + + return parser + +def main(args = None): + parser = make_parser() + args = site_config.parse_args(agent_class='RfsocController', + parser = parser, + args = args) + + agent, runner = ocs_agent.init_site_agent(args) + rfsoc_controller = RFSoController(agent, config = args.config, module = args.module) + + agent.register_task('run', rfsoc_controller.run, blocking=False) + agent.register_task('abort', rfsoc_controller.abort, blocking=False) + agent.register_task('take_vna_sweep', rfsoc_controller.take_vna_sweep) + + runner.run(agent, auto_reconnect=True) + +if __name__ == '__main__': + main() \ No newline at end of file diff --git a/pcs/drivers/coldload.py b/pcs/drivers/coldload.py index 6ef1b64..6d70358 100644 --- a/pcs/drivers/coldload.py +++ b/pcs/drivers/coldload.py @@ -1,18 +1,20 @@ from ocs.ocs_client import OCSClient import time import numpy as np +import txaio +txaio.use_twisted() class Coldload: - def __init__(self, lakeshore, ls_channel, ext_log = None): + def __init__(self, lakeshore, ls_channel): self.ls_channel = ls_channel - self.ext_log = ext_log + self.log = txaio.make_logger() # Create Lakeshore client for grabbing coldload temperature data try: self.lakeshore = OCSClient(lakeshore, args=[]) except Exception as e: - self.logger(f'Could not connect to Lakeshore agent \033[3m{lakeshore}\033[0m for temperature monitoring: {e}', 'error') + self.log.error(f'Could not connect to Lakeshore agent \033[3m{lakeshore}\033[0m for temperature monitoring: {e}') def get_temp(self): """ @@ -27,14 +29,14 @@ def get_temp(self): try: temp = acq_status['data']['fields'][self.ls_channel]['T'] except KeyError as e: - self.logger(f'Specified Lakeshore channel {self.ls_channel} is not valid: {e}', 'error') + self.log.error(f'Specified Lakeshore channel {self.ls_channel} is not valid: {e}') temp = None else: - self.logger('Lakeshore data acquisition is not running.', 'error') + self.log.error('Lakeshore data acquisition is not running.') temp = None return temp - def set_temp(self, temp, get_current, set_current, *args, **kwargs): + def set_temp(self, temp: float, get_current, set_current, *args, **kwargs): """ Set the temperature of the coldload using a proportional integral derivative (PID) controller. The PID controller uses the coldload temperature as the process variable and the current squared as the control variable. @@ -48,10 +50,12 @@ def set_temp(self, temp, get_current, set_current, *args, **kwargs): kwargs: sample_int (float): Interval at which to sample coldload temperature avg_int (float): Interval over which to average coldload temperatures (averaged temperature used as PID process variable). Also sets timescale for PID control + thresholds (List(float)): Error thresholds at which to modify avg_int. avg_int will be used for errors greater than the largest threshold and then multiplied by 2 for each threshold passed. timeout (float): Time in minutes after which to exit PID loop (0 for indefinite) yield_dict (bool): Whether to yield error values and coldload current after each PID control loop max_current (float): Maximum current limit pid (List[float]): Proportional, integral, and derivative control coefficients + int_threshold (float): Error threshold hold after which the integral term will start contributing to the PID control. """ sample_int = 0.5 @@ -59,9 +63,11 @@ def set_temp(self, temp, get_current, set_current, *args, **kwargs): timeout = 180 yield_dict = False + reset_current = False max_current = 0.6 - pid = [2.25e-3, 5.1e-7, 0.71] - thresholds = [0.01, 0.1, 1, 5] + pid = [5e-4, 1e-7, 9e-2] + int_threshold = 0.125 + thresholds = [1.13e-3, 2.5e-7, 0.35] err_p = temp - self.get_temp() err_i = 0.0 @@ -73,6 +79,8 @@ def set_temp(self, temp, get_current, set_current, *args, **kwargs): sample_int = v elif k == 'avg_int': default_avg_int = v + elif k == 'thresholds': + thresholds = v elif k == 'timeout': timeout = v elif k == 'yield_dict': @@ -83,7 +91,10 @@ def set_temp(self, temp, get_current, set_current, *args, **kwargs): pid = v elif k == 'err_i': err_i = v - + elif k == 'int_threshold': + int_threshold = v + elif k == 'reset_current': + reset_current = v avg_int = default_avg_int timeout *= 60 # Convert timeout to seconds @@ -108,11 +119,11 @@ def set_temp(self, temp, get_current, set_current, *args, **kwargs): # Calculate the PID error values err_d = (avg_err - err_p)/delta_t - err_i += avg_err * delta_t + if np.abs(avg_err) <= int_threshold: err_i += avg_err * delta_t err_p = avg_err # Vary avg_int depending on how small error is to reduce noise in derivative at small errors - avg_int = default_avg_int * (2 ** sum(err_p < threshold for threshold in thresholds)) + avg_int = default_avg_int * (2 ** sum(np.abs(err_p) < threshold for threshold in thresholds)) # Set the integral error to zero if the current is already zero so that there is not a large accumulated error as the temperature decays slowly if curr_sq == 0: err_i = 0.0 @@ -132,23 +143,4 @@ def set_temp(self, temp, get_current, set_current, *args, **kwargs): yield pids time.sleep(0.1) # Wait to prevent wasting CPU resources if yield_dict: yield None # Yield None on non-PID loops to prevent the method from blocking for avg_int seconds - - #=========# - # Logging # - #=========# - def logger(self, msg, level): - level = '_' + level - if not self.ext_log is None and hasattr(self, level): - method = getattr(self, level) - method(msg) - else: - print(msg) - - #==========================# - # Logging Internal Methods # - #==========================# - def _info(self, msg): - self.ext_log.info(msg) - - def _error(self, msg): - self.ext_log.error(msg) \ No newline at end of file + if reset_current: set_current(*args, curr=0.0) \ No newline at end of file diff --git a/pcs/plugin.py b/pcs/plugin.py index c09f0ef..e4e68f1 100644 --- a/pcs/plugin.py +++ b/pcs/plugin.py @@ -6,5 +6,6 @@ 'Bluefors_TC_Agent': {'module': 'pcs.agents.bluefors_tc.agent', 'entry_point': 'main'}, 'ColdloadAgent_ScpiPsu': {'module': 'pcs.agents.coldload_scpipsu.agent', 'entry_point': 'main'}, 'PrimecamBiasAgent': {'module': 'pcs.agents.primecam_bias.agent', 'entry_point': 'main'}, - 'BeamMapperAgent': {'module': 'pcs.agents.beam_mapper.agent', 'entry_point': 'main'} + 'BeamMapperAgent': {'module': 'pcs.agents.beam_mapper.agent', 'entry_point': 'main'}, + 'RFSoController': {'module': 'pcs.agents.rfsoc_controller.agent', 'entry_point': 'main'} } From 655d283d51d5adf7c6563e1ad6bfbb942f1bc4af Mon Sep 17 00:00:00 2001 From: RanDarshan Date: Tue, 8 Jul 2025 11:10:03 -0400 Subject: [PATCH 5/5] rfsoc controller updates --- docker/rfsoc_controller/Dockerfile | 2 - pcs/agents/rfsoc_controller/agent.py | 171 ++++++++++++++++++++++----- 2 files changed, 143 insertions(+), 30 deletions(-) diff --git a/docker/rfsoc_controller/Dockerfile b/docker/rfsoc_controller/Dockerfile index 2d052b5..139d620 100644 --- a/docker/rfsoc_controller/Dockerfile +++ b/docker/rfsoc_controller/Dockerfile @@ -10,6 +10,4 @@ RUN python -m pip install -e ./ccatkidlib && \ WORKDIR / -ENV OCS_CONFIG_DIR='/config' - ENTRYPOINT ["dumb-init", "ocs-agent-cli"] \ No newline at end of file diff --git a/pcs/agents/rfsoc_controller/agent.py b/pcs/agents/rfsoc_controller/agent.py index 478ef0e..63fc1cb 100644 --- a/pcs/agents/rfsoc_controller/agent.py +++ b/pcs/agents/rfsoc_controller/agent.py @@ -48,12 +48,23 @@ def processExited(self, status: Failure): class RFSoController: ''' - PCS Agent for controlling RFSoCs through ccatkidlib scripts and methods. + PCS Agent for controlling Radio Frequency Systems on a Chip (RFSoCs) through + ccatkidlib scripts and methods. + Modelled after SOCS PysmurfController with modifications. ''' + def __init__(self, agent, config: str = None, module: str = None): ''' Constructor for RfsocController. + Initializes agent and starts new measurement session. + + Parameters: + agent (ocs.ocs_agent): OCS agent instance + config (str): Path to rfsoc-controller config relative to OCS_CONFIG_DIR + module (str): Which instrument module to control with the rfsoc-controller + Notes: + Arguments for constructor passed through OCS config file (e.g. default.yaml in OCS_CONFIG_DIR) ''' # Create OCS agent and get log @@ -82,10 +93,23 @@ def __init__(self, agent, config: str = None, module: str = None): @ocs_agent.param('init_boards', type = bool, default = False) def new_session(self, session, params): + '''new_session(init_boards=False) + + **Task** - Start a new measurement session + + Parameters: + init_boards (bool, optional): Whether to reinitialize the RFSoC boards + ''' RC = self._new_session(init_boards=params['init_boards']) return True, f'Succesfully created new session: {self.session}' def _new_session(self, init_boards): + ''' + Internal method for starting a new measurement session. + + Parameters: + init_boards (bool): Whether to reinitialize the RFSoC boards + ''' RC = R(cfg_path = self.sys_cfg_path, init_boards = init_boards, init_drones = True) # Instantiate RFSoC control object with full board and drone setup self._update_control(RC) self.log.info(f'Succesfully created new session: {self.session}') @@ -93,6 +117,14 @@ def _new_session(self, init_boards): @staticmethod def _get_control(func): + ''' + Decorator for use with OCS tasks/processes of ccatkidlib methods. + Creates the RFSoC control object with correct system state and passes it to decorated task/process. + Updates system state after task/process finishes execution. + + Parameters: + func (func): OCS task/process of ccatkidlib method to decorate + ''' @wraps(func) def _wrapper(self, session, params): RC = R(cfg_path = self.sys_cfg_path, initialize_boards = False, initialize_drones = False, @@ -116,11 +148,14 @@ def _wrapper(self, session, params): def _update_control(self, RC): ''' - Get the current state of the control object. + Internal method for updating the system state based on the state of the given RFSoC control object. + + Parameters: + RC (ccatkidlib.rfsoc.rfsoc_daq.R): RFSoC control object ''' - # Create attributes to save system state of control object across recreations - # --------------------------------------------------------------------------- + # Create/update attributes to save system state of control object across recreations + # ---------------------------------------------------------------------------------- # Get the session ID, name, and description of measurement self.session = RC.sess_id self.curr_date = RC.curr_date @@ -132,6 +167,22 @@ def _update_control(self, RC): self.drive_attens = RC.drive_attens self.sense_attens = RC.sense_attens + #===============# + # Setup Methods # + #===============# + + @_get_control + @ocs_agent.param('com_to', type=(str, list[str]), default=None) + @ocs_agent.param('drive', type=(int, list[int]), default=None) + @ocs_agent.param('sense', type=(int, list[int]), default=None) + def set_atten(self, session, params): + return + + @_get_control + @ocs_agent.param('com_to', type=list, default=[]) + def set_NCLO(self, session, params): + return + #================# # Script Methods # #================# @@ -139,19 +190,13 @@ def _update_control(self, RC): @inlineCallbacks def _run_script(self, session, script, args): """ - Runs a ccatkidlib control script using the Twisted reactor. - Modified _run_script method of SOCS PysmurfController - - Args: - script (string): - path to the script you wish to run - args (list, optional): - List of command line arguments to pass to the script. - Defaults to []. - log (string or bool, optional): - Determines if and how the process's stdout should be logged. - You can pass the path to a logfile, True to use the agent's log, - or False to not log at all. + Internal method for running a ccatkidlib RFSoC control script using the Twisted reactor. + Modelled after _run_script method of SOCS PysmurfController + + Parameters: + session (ocs.ocs_agent.OpSession): OpSession object of run task + script (str): Path of ccatkidlib python script to run + args (list[str], optional): Additional arguments to pass to script """ with self.lock.acquire_timeout(5, job=script) as acquired: @@ -182,6 +227,23 @@ def _run_script(self, session, script, args): @inlineCallbacks def run(self, session, params=None): + '''run(script, args=None) + + **Task** - Run a ccatkidlib RFSoC control script + + Parameters: + script (str): Path of ccatkidlib python script to run + args (list[str], optional): Additional arguments to pass to script + + Examples: + Example for running a test script with a client:: + client.run(script='/app/pcs/ccatkidlib/scripts/controller/test.py', args=[]) + Notes: + Script path must be that within the docker container. + For example, if ccatkidlib is mounted to /app/pcs/ccatkidlib within the container, + the path to run a script in the scripts directory would be /app/pcs/ccatkidlib/scripts/.py + + ''' status, msg = yield self._run_script(session, params['script'], params.get('args', [])) # Set stored NCLO and attenuations to None since their state may have changed during script execution @@ -210,7 +272,7 @@ def tune(): return @_get_control - @ocs_agent.param('com_to', type=list, default=[]) + @ocs_agent.param('com_to', type=list, default=None) @ocs_agent.param('time', type=float) def take_timestream(self, session, params): return @@ -220,13 +282,33 @@ def take_timestream(self, session, params): #===============# @_get_control - @ocs_agent.param('R', type=R, default=None) + @ocs_agent.param('R', type=R) @ocs_agent.param('com_to', type=list, default=None) - @ocs_agent.param('write_comb', type=bool, default=None) - @ocs_agent.param('sweep_steps', type=int, default=None) + @ocs_agent.param('write_comb', type=bool, default=True) + @ocs_agent.param('sweep_steps', type=int, default=None, check=lambda x: x > 0) @ocs_agent.param('parallel_boards', type=int, default=None) - @ocs_agent.param('parallel_drones', type=int, default=None) + @ocs_agent.param('parallel_drones', type=int, default=None, check=lambda x: 4 >= x >= 1) def take_vna_sweep(self, session, params): + '''take_vna_sweep(com_to=None, write_comb=True, sweep_steps=None, parallel_boards=None, parallel_drones=None) + + **Task** - Take a VNA sweep + + Parameters: + com_to (list[str], optional): List of drones to take VNA sweep + write_comb (bool, optional): Whether to write a new VNA comb (default: True) + sweep_steps (int, optional): Number of points each tone should sweep (default: sweep_steps in drone_config) + parallel_boards (int, optional): Number of boards to run in parallel (default: parallel_boards in system_config) + parallel_drones (int, optional): Number of drones to run in parallel (default: parallel_drones in system_config) + + Examples: + Take VNA sweep with all drones of board 1 and drone 1 of board 2 in parallel:: + client.take_vna_sweep(com_to=['1', '2.1'], sweep_steps=500, parallel_boards=2, parallel_drones=4) + + Notes: + Example session data: + >>> response.session['data'] + PUT EXAMPLE HERE + ''' with self.lock.acquire_timeout(5, job='vna_sweep') as acquired: if not acquired: self.log.error(f"Could not acquire lock because it is held by {self.lock.job}.") @@ -270,20 +352,44 @@ def monitor_space(self, session, params): # Helper Methods # #================# def _filter_params(self, params): + ''' + Internal function for filtering out keys with None value from params dictionary + so that ccatkidlib defaults are used + + Parameters: + params (dict[any]): params dictionary to filter + ''' return {k:v for k, v in params.items() if v is not None} def _publish_data(self, data, RC, params, session): + ''' + Internal method for publishing data returned by ccatkidlib OCS task/process + to the OCS OpSession.data dictionary. + + Parameters: + data (any): Data returned by ccatkidlib method that was run + RC (ccatkidlib.rfsoc.rfsoc_daq.R): RFSoC control object used to run ccatkidlib method + params (dict[any]): Parameters used to run ccatkidlib method + session (ocs.ocs_agent.OpSession): OpSession of ccatkidlib OCS task/process + ''' + + # Check if ccatkidlib method was run with a different set of drones than in system_config file com_to = params['com_to'] if 'com_to' in params else RC.drone_list + + # Create data dictionary with returned data, drones used, and measurement info data_dict = {'name': RC.measurement_name, - 'date': RC.curr_date, - 'session': RC.sess_id, - 'com_to': com_to, - 'data': data} - session.data['data'] = data_dict + 'date': RC.curr_date, + 'session': RC.sess_id, + 'timestamp': RC.timestamp, + 'com_to': com_to, + 'data': data} + + # Pass data dictionary to OpSession.data + session.data = data_dict def make_parser(parser=None): ''' - Build ArgumentParser for passing arguments through OCS_CONFIG file + Build ArgumentParser for passing arguments through OCS config file (e.g. default.yaml in OCS_CONFIG_DIR) ''' if parser is None: parser = argparse.ArgumentParser() @@ -297,18 +403,27 @@ def make_parser(parser=None): return parser def main(args = None): + # Parse arguments passed in OCS config file + # ----------------------------------------- parser = make_parser() args = site_config.parse_args(agent_class='RfsocController', parser = parser, args = args) + # Create RFSoController agent + # --------------------------- agent, runner = ocs_agent.init_site_agent(args) rfsoc_controller = RFSoController(agent, config = args.config, module = args.module) + + # Register agent tasks and processes + # ---------------------------------- agent.register_task('run', rfsoc_controller.run, blocking=False) agent.register_task('abort', rfsoc_controller.abort, blocking=False) agent.register_task('take_vna_sweep', rfsoc_controller.take_vna_sweep) + # Run agent + # --------- runner.run(agent, auto_reconnect=True) if __name__ == '__main__':