11434250339

Committed 21 Oct 2024 06:10AM UTC coverage: 84.152% (-15.8%) from 100.0%

Build # 11434250339

Build Type

Pull #179

github

Committed by

web-flow

Commit Message

Merge f1e11576e into c30598d1c

Pull Request Pull Request #179: add time_constant function

Run Details

5 of 105 new or added lines in 1 file covered. (4.76%)

70 existing lines in 1 file now uncovered.

531 of 631 relevant lines covered (84.15%)

0.84 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

59.51

/src/sorunlib/wiregrid.py

import time

import sorunlib as run
from sorunlib._internal import check_response, check_running

EL_DIFF_THRESHOLD = 0.5  # deg diff from target that its ok to run calibration
BORESIGHT_DIFF_THRESHOLD = 0.5  # deg
AGENT_TIMEDIFF_THRESHOLD = 5  # sec
OP_TIMEOUT = 60


# Internal Helper Functions
def _check_process_data(process, last_timestamp):
    """Check the latest timestamp from a process' session.data is recent enough.

    Args:
        process (str): Descriptive name, used in the error message.
        last_timestamp (float): Latest timestamp from the session.data.

    Raises:
        RuntimeError: When the last timestamp was more than
            AGENT_TIMEDIFF_THRESHOLD old.

    """
    try:
        assert ((time.time() - last_timestamp) < AGENT_TIMEDIFF_THRESHOLD)
    except AssertionError:
        error = f"{process} has no updated data. Cannot proceed with " + \
            "wiregrid calibration. Aborting."
        raise RuntimeError(error)


def _verify_temp_response(response, sensor, min_temp):
    """Check temperature above minimum temperation for safe operation of wiregrid.

    Args:
        response (ocs.ocs_client.OCSReply): Client response from the labjack
            acq process.
        sensor (str): Sensor name in the labjack session data, i.e. 'AIN0'.
        min_temp (int, float): Minimum temperature for safe operation in C.

    Raises:
        RuntimeError: When the temperature of sensor below ``min_temp``.

    """
    temp = response.session['data']['data'][sensor]
    try:
        assert (temp > min_temp)
    except AssertionError:
        error = f"Sensor {sensor} reads {temp} C, which is below minimum " + \
            f"temperature of {min_temp} C. Cannot proceed with wiregrid " + \
            "calibration. Aborting"
        raise RuntimeError(error)


def _check_zenith():
    """Checks to see if the telescope is currently pointing at zenith.

    Returns:
        bool: True if the telescope is at zenith, otherwise returns False.

    """
    acu = run.CLIENTS['acu']
    resp = acu.monitor.status()
    el = resp.session['data']['StatusDetailed']['Elevation current position']

    zenith = False
    if (abs(el - 90) < EL_DIFF_THRESHOLD):
        zenith = True

    return zenith


# Calibration Functions
def _check_telescope_position(elevation_check=True, boresight_check=True):
    # Get current telescope position
    acu = run.CLIENTS['acu']
    resp = acu.monitor.status()
    az = resp.session['data']['StatusDetailed']['Azimuth current position']
    el = resp.session['data']['StatusDetailed']['Elevation current position']
    boresight = resp.session['data']['StatusDetailed']['Boresight current position']

    # Check appropriate elevation
    if elevation_check:
        try:
            assert (el > 50 - EL_DIFF_THRESHOLD)
        except AssertionError:
            error = "Telescope not at > 50 deg elevation. Cannot proceed with " + \
                    f"wiregrid calibration in current position ({az}, {el}). " + \
                    "Aborting."
            raise RuntimeError(error)

    # Check boresight angle
    if boresight_check:
        try:
            assert (abs(boresight - 0) < BORESIGHT_DIFF_THRESHOLD)
        except AssertionError:
            error = "Telescope not at 0 deg boresight. Cannot proceed with " + \
                    f"wiregrid calibration in current position ({boresight}). " + \
                    "Aborting."
            raise RuntimeError(error)


def _configure_power(continuous):
    """Configure the KIKUSUI power supply for rotation based on rotation type.

    Args:
        continuous (bool): Configure for continuous rotation or not.

    """
    kikusui = run.CLIENTS['wiregrid']['kikusui']

    resp = kikusui.set_v(volt=12)
    check_response(kikusui, resp)

    cfg = run.config.load_config()
    current = cfg.get('wiregrid_motor_current', 3.0)

    resp = kikusui.set_c(current=current)
    check_response(kikusui, resp)


def _rotate_continuously(duration):
    kikusui = run.CLIENTS['wiregrid']['kikusui']

    # Start rotation
    resp = kikusui.set_on()
    check_response(kikusui, resp)

    # Wait
    time.sleep(duration)

    # Stop rotation
    resp = kikusui.set_off()
    check_response(kikusui, resp)


def _check_motor_on():
    actuator = run.CLIENTS['wiregrid']['actuator']

    # Check motor is on
    resp = actuator.acq.status()
    if resp.session['data']['fields']['motor'] == 1:
        print("Wiregrid motor already on.")
    # Turn on motor if needed
    elif resp.session['data']['fields']['motor'] == 0:
        resp = actuator.motor_on()
        check_response(actuator, resp)
    else:
        raise RuntimeError("Motor state unknown. Aborting...")


def _check_agents_online():
    """Check OCS agents related to the wiregrid are running."""
    actuator = run.CLIENTS['wiregrid']['actuator']
    kikusui = run.CLIENTS['wiregrid']['kikusui']
    encoder = run.CLIENTS['wiregrid']['encoder']
    labjack = run.CLIENTS['wiregrid']['labjack']

    # wiregrid_actuator
    resp = actuator.acq.status()
    last_timestamp = resp.session['data']['timestamp']
    check_running(actuator, resp)
    _check_process_data("Actuator agent", last_timestamp)

    # wiregrid_kikusui
    resp = kikusui.IV_acq.status()
    last_timestamp = resp.session['data']['timestamp']
    check_running(kikusui, resp)
    _check_process_data("Kikusui agent", last_timestamp)

    # wiregrid_encoder
    resp = encoder.acq.status()
    last_timestamp = resp.session['data']['timestamp']
    check_running(encoder, resp)
    _check_process_data("Encoder agent", last_timestamp)

    # labjack
    resp = labjack.acq.status()
    last_timestamp = resp.session['data']['timestamp']
    check_running(labjack, resp)
    _check_process_data("Labjack agent", last_timestamp)

    # encoder data stream
    resp = encoder.acq.status()
    last_timestamp = resp.session['data']['fields']['encoder_data']['last_updated']
    _check_process_data("Encoder BBB", last_timestamp)


def _check_temperature_sensors():
    """Check temperatures within safe range."""
    labjack = run.CLIENTS['wiregrid']['labjack']
    resp = labjack.acq.status()
    # AIN0 > -10 C && AIN1 > -10 C (On the wiregrid)
    _verify_temp_response(resp, 'AIN0C', -10)
    _verify_temp_response(resp, 'AIN1C', -10)
    # AIN2 > 0 C (Inside the electronics enclosure for the gridloader)
    _verify_temp_response(resp, 'AIN2C', 0)


def _check_wiregrid_position():
    """Check the wiregrid position."""
    actuator = run.CLIENTS['wiregrid']['actuator']
    resp = actuator.acq.status()
    ls_status = resp.session['data']['fields']['limitswitch']
    if (ls_status['LSR1'] == 1 or ls_status['LSL1'] == 1) and not (ls_status['LSR2'] == 1 and ls_status['LSL2'] == 1):
        position = 'outside'
    elif (ls_status['LSR2'] == 1 or ls_status['LSL2'] == 1) and not (ls_status['LSR1'] == 1 or ls_status['LSL1'] == 1):
        position = 'inside'
    else:
        raise RuntimeError("The wiregrid position is unknown. The limitswitch response is like this:\n" +
                           "LSR1: {}\n".format('ON' if ls_status['LSR1'] == 1 else 'OFF') +
                           "LSL1: {}\n".format('ON' if ls_status['LSL1'] == 1 else 'OFF') +
                           "LSR2: {}\n".format('ON' if ls_status['LSR2'] == 1 else 'OFF') +
                           "LSL2: {}\n".format('ON' if ls_status['LSL2'] == 1 else 'OFF') +
                           "Aborting...")
    return position


def _stop_hwp_with_wiregrid():
    """Stop the HWP after comfirming the wiregrid is inserted."""
    # Check the wiregrid position
    position = _check_wiregrid_position()
    if position == 'outside':
        raise RuntimeError("The wiregrid is not inserted. Aborting...")
    elif position == 'inside':
        # Stop the HWP
        print("Starting to stop the HWP.")
        run.hwp.stop(active=True)
        print("The HWP stopped successfully.")
    else:
        raise RuntimeError("Unknown wiregrid position. Aborting...")


def _spin_hwp_with_wiregrid(target_hwp_direction):
    """Spin the HWP to the target direction at 2Hz after comfirming the wiregrid is inserted.
    
    Args:
        target_hwp_direction (str): Target HWP direction, 'forward' or 'backward'.
    """
    # Check the wiregrid position
    position = _check_wiregrid_position()
    if position == 'outside':
        raise RuntimeError("The wiregrid is not inserted. Aborting...")
    elif position == 'inside':
        if target_hwp_direction == 'forward':
            print("Starting to spin up the HWP in forward.")
            run.hwp.set_freq(freq=2.0)
            print("The HWP is spinning at 2Hz in forward successfully.")
        elif target_hwp_direction == 'backward':
            print("Starting to spin up the HWP in backward.")
            run.hwp.set_freq(freq=-2.0)
            print("The HWP is spinning at 2Hz in backward successfully.")
        else:
            raise RuntimeError("Unknown HWP target direction. Aborting...")
    else:
        raise RuntimeError("Unknown wiregrid position. Aborting...")


def _reverse_hwp_with_wiregrid(initial_hwp_direction, streaming=False, stepwise_before=False, stepwise_after=False):
    """Change the HWP direction after comfirming the wiregrid is inserted.
    
    Args:
        initial_hwp_direction (str): Initial HWP direction, 'forward' or 'backward'.
        streaming (bool): Do SMuRF streaming during the HWP direction chaging or not . Default is False.
        stepwise_before (bool): Do stepwise rotation before changing HWP direction or not. Default is False.
        stepwise_after (bool): Do stepwise rotation after changing HWP direction or not. Default is False.
    """
    if initial_hwp_direction not in ['forward', 'backward']: 
        raise RuntimeError("Initial HWP direction should be either 'forward' or 'backward'. Aborting...")

    current_hwp_direction = initial_hwp_direction
    if current_hwp_direction == 'forward':
        target_hwp_direction = 'backward'
    elif current_hwp_direction == 'backward':
        target_hwp_direction = 'forward'

    # Stop and spin up reversely the HWP
    try:
        # Enable SMuRF streams
        if streaming:
            stream_tag = f'wiregrid, wg_time_constant, hwp_change_to_{target_hwp_direction}'
            if stepwise_before or stepwise_after:
                stream_tag += ', wg_stepwise'
            if _check_zenith():
                stream_tag += ', wg_el90'
            run.smurf.stream('on', subtype='cal', tag=stream_tag)
        # Stepwise rotation before stopping the HWP
        if stepwise_before:
            rotate(False)
        # Stop the HWP
        _stop_hwp_with_wiregrid()
        # Spin up the HWP in the opposite direction
        _spin_hwp_with_wiregrid(target_hwp_direction)
        current_hwp_direction = target_hwp_direction
        # Stepwise rotation after spinning up the HWP 
        if stepwise_after:
            rotate(False)
    finally:
        # Stop SMuRF streams
        if streaming:
            run.smurf.stream('off')

    return current_hwp_direction


# Public API
def insert():
    """Insert the wiregrid."""
    actuator = run.CLIENTS['wiregrid']['actuator']
    resp = actuator.insert()
    try:
        check_response(actuator, resp)
    except RuntimeError as e:
        error = "Wiregrid insertion failed. Please inspect wiregrid before " + \
                "continuing observations.\n" + str(e)
        raise RuntimeError(error)


def eject():
    """Eject the wiregrid."""
    actuator = run.CLIENTS['wiregrid']['actuator']
    resp = actuator.eject()
    try:
        check_response(actuator, resp)
    except RuntimeError as e:
        error = "Wiregrid eject failed. Please inspect wiregrid before " + \
                "continuing observations.\n" + str(e)
        raise RuntimeError(error)


def rotate(continuous, duration=30, num_laps=1, stopped_time=10.):
    """Rotate the wiregrid.

    Rotation is either continuously for ``duration`` seconds or stepwise for
    ``num_laps``, stopping every 22.5 degrees for ``stopped_time`` seconds.

    This function sets the Kikusui power settings appropriately based on the
    rotation type.

    Args:
        continuous (bool): Rotate the grid continuously if True or not if
            False.
        duration (int, float): Amount of time in seconds to rotate if rotating
            continuously. Defaults to 30 seconds.
        num_laps (int): Number of revolutions if rotating stepwise (i.e.
            ``continuous = False``). Defaults to 1.
        stopped_time (float): Duration of each 22.5 deg step in seconds if
            rotating stepwise. Defaults to 10 seconds.

    """
    kikusui = run.CLIENTS['wiregrid']['kikusui']

    # Configure power settings based on rotation type
    _configure_power(continuous)

    if continuous:
        _rotate_continuously(duration)
    else:
        resp = kikusui.stepwise_rotation(num_laps=num_laps,
                                         stopped_time=stopped_time)
        check_response(kikusui, resp)


def calibrate(continuous=False, elevation_check=True, boresight_check=True,
              temperature_check=True):
    """Run a wiregrid calibration.

    Args:
        continuous (bool): Calibration by continuous rotation or not.
            Default is False, in which the wiregrid rotates step-wisely.
        elevation_check (bool): Check the elevation angle is in an appropriate
            range before the calibration or not. Default is True.
        boresight_check (bool): Check the boresight angle is in an appropriate
            range before the calibration or not. Default is True.
        temperature_check (bool): Check the temperature of various components
            are within operational limits before the calibration or not.
            Default is True.

    """
    try:
        _check_telescope_position(elevation_check=elevation_check,
                                  boresight_check=boresight_check)
        _check_agents_online()
        if temperature_check:
            _check_temperature_sensors()
        _check_motor_on()

        # Rotate for reference before insertion
        rotate(continuous=True, duration=10)

        # Enable SMuRF streams
        if continuous:
            rotation = 'wg_continuous'
        else:
            rotation = 'wg_stepwise'
        if _check_zenith():
            el_tag = ', wg_el90'
        else:
            el_tag = ''
        run.smurf.stream('on', tag=f'wiregrid, {rotation}{el_tag}', subtype='cal')

        # Insert the wiregrid
        insert()

        # Rotate the wiregrid
        rotate(continuous)

        # Eject the wiregrid
        eject()
    finally:
        # Stop SMuRF streams
        run.smurf.stream('off')


def time_constant(initial_hwp_direction, stepwise_first=True, stepwise_last=True, stepwise_mid=False, repeat=1):
    """
    Run a wiregrid time constant measurement.
    
    Args:
        initial_hwp_direction (str): Initial HWP direction, 'forward' or 'backward'.
        stepwise_first (bool): Do stepwise rotation or not before the first HWP speed change. Default is True.
        stepwise_last (bool): Do stepwise rotation or not after the last HWP speed change. Default is True.
        stepwise_mid (bool): Do stepwise rotation between each HWP speed change. Default is False.
        repeat (int): Number of repeats. Default is 1.
            If this is odd, the HWP direction will be changed to the opposite of the initial direction.
            If this is even, the HWP direction will be the same as the initial direction.
    """
    
    # Check the initial HWP direction
    if initial_hwp_direction not in ['forward', 'backward']:
        raise RuntimeError("Initial HWP direction should be either 'forward' or 'backward'. Aborting...")
    current_hwp_direction = initial_hwp_direction

    # Check the repeat
    if repeat < 1 or not isinstance(repeat, int):
        raise RuntimeError("The repeat should be int and larger than 0. Aborting...")

    _check_agents_online()
    _check_motor_on()
    _check_telescope_position(elevation_check=True, boresight_check=False)
    
    if _check_zenith:
        el_tag = ', wg_el90'
    else:
        el_tag = ''

    # Rotate for reference before insertion
    rotate(continuous=True, duration=10)

    # Bias step (the wire grid is off the window)
    bs_tag = f'biasstep, wg_time_constant, wg_ejected, hwp_2hz_{current_hwp_direction}' + el_tag
    run.smurf.bias_step(tag=bs_tag, concurrent=True)

    # Insert the wiregrid with streaming
    time.sleep(5)
    try:
        # Enable SMuRF streams
        stream_tag = f'wg_time_constant, wg_inserting, hwp_2hz_{current_hwp_direction}' + el_tag
        run.smurf.stream('on', tag=stream_tag, subtype='cal')
        # Insert the wiregrid
        insert()
    finally:
        # Stop SMuRF streams
        run.smurf.stream('off')
    time.sleep(5)

    for i in range(repeat):
        # Bias step (the wire grid is on the window)
        bs_tag = f'biasstep, wg_time_constant, wg_inserted, hwp_2hz_{current_hwp_direction}' + el_tag
        run.smurf.bias_step(tag=bs_tag, concurrent=True)

        stepwise_before = False
        stepwise_after = False
        if stepwise_first and i == 0:
            # Stepwise rotation before the first HWP speed change
            stepwise_before = True
        if stepwise_mid and i != 0:
            # Stepwise rotation between changing HWP speed
            stepwise_before = True
        if stepwise_last and i == repeat-1:
            # Stepwise rotation after the last HWP speed change
            stepwise_after = True

        # Spin the HWP in the opposite direction of the initial direction with streaming and stepwise rotation
        current_hwp_direction = _reverse_hwp_with_wiregrid(current_hwp_direction,
                                                           streaming=True,
                                                           stepwise_before=stepwise_before,
                                                           stepwise_after=stepwise_after)

    # Bias step (the wire grid is on the window)
    bs_tag = f'biasstep, wg_time_constant, wg_inserted, hwp_2hz_{current_hwp_direction}' + el_tag
    run.smurf.bias_step(tag=bs_tag, concurrent=True)

    # Eject the wiregrid with streaming
    time.sleep(5)
    try:
        # Enable SMuRF streams
        stream_tag = f'wg_time_constant, wg_ejecting, hwp_2hz_{current_hwp_direction}' + el_tag
        run.smurf.stream('on', tag=stream_tag, subtype='cal')
        # Eject the wiregrid
        eject()
    finally:
        # Stop SMuRF streams
        run.smurf.stream('off')
    time.sleep(5)

    # Bias step (the wire grid is off the window)
    bs_tag = f'biasstep, wg_time_constant, wg_ejected, hwp_2hz_{current_hwp_direction}' + el_tag
    run.smurf.bias_step(tag=bs_tag, concurrent=True)

1	import time	1✔
2
3	import sorunlib as run	1✔
4	from sorunlib._internal import check_response, check_running	1✔
5
6	EL_DIFF_THRESHOLD = 0.5 # deg diff from target that its ok to run calibration	1✔
7	BORESIGHT_DIFF_THRESHOLD = 0.5 # deg	1✔
8	AGENT_TIMEDIFF_THRESHOLD = 5 # sec	1✔
9	OP_TIMEOUT = 60	1✔
10
11
12	# Internal Helper Functions
13	def _check_process_data(process, last_timestamp):	1✔
14	"""Check the latest timestamp from a process' session.data is recent enough.
15
16	Args:
17	process (str): Descriptive name, used in the error message.
18	last_timestamp (float): Latest timestamp from the session.data.
19
20	Raises:
21	RuntimeError: When the last timestamp was more than
22	AGENT_TIMEDIFF_THRESHOLD old.
23
24	"""
25	try:	1✔
26	assert ((time.time() - last_timestamp) < AGENT_TIMEDIFF_THRESHOLD)	1✔
27	except AssertionError:	1✔
28	error = f"{process} has no updated data. Cannot proceed with " + \	1✔
29	"wiregrid calibration. Aborting."
30	raise RuntimeError(error)	1✔
31
32
33	def _verify_temp_response(response, sensor, min_temp):	1✔
34	"""Check temperature above minimum temperation for safe operation of wiregrid.
35
36	Args:
37	response (ocs.ocs_client.OCSReply): Client response from the labjack
38	acq process.
39	sensor (str): Sensor name in the labjack session data, i.e. 'AIN0'.
40	min_temp (int, float): Minimum temperature for safe operation in C.
41
42	Raises:
43	RuntimeError: When the temperature of sensor below ``min_temp``.
44
45	"""
46	temp = response.session['data']['data'][sensor]	1✔
47	try:	1✔
48	assert (temp > min_temp)	1✔
49	except AssertionError:	1✔
50	error = f"Sensor {sensor} reads {temp} C, which is below minimum " + \	1✔
51	f"temperature of {min_temp} C. Cannot proceed with wiregrid " + \
52	"calibration. Aborting"
53	raise RuntimeError(error)	1✔
54
55
56	def _check_zenith():	1✔
57	"""Checks to see if the telescope is currently pointing at zenith.
58
59	Returns:
60	bool: True if the telescope is at zenith, otherwise returns False.
61
62	"""
63	acu = run.CLIENTS['acu']	1✔
64	resp = acu.monitor.status()	1✔
65	el = resp.session['data']['StatusDetailed']['Elevation current position']	1✔
66
67	zenith = False	1✔
68	if (abs(el - 90) < EL_DIFF_THRESHOLD):	1✔
69	zenith = True	1✔
70
71	return zenith	1✔
72
73
74	# Calibration Functions
75	def _check_telescope_position(elevation_check=True, boresight_check=True):	1✔
76	# Get current telescope position
77	acu = run.CLIENTS['acu']	1✔
78	resp = acu.monitor.status()	1✔
79	az = resp.session['data']['StatusDetailed']['Azimuth current position']	1✔
80	el = resp.session['data']['StatusDetailed']['Elevation current position']	1✔
81	boresight = resp.session['data']['StatusDetailed']['Boresight current position']	1✔
82
83	# Check appropriate elevation
84	if elevation_check:	1✔
85	try:	1✔
86	assert (el > 50 - EL_DIFF_THRESHOLD)	1✔
87	except AssertionError:	1✔
88	error = "Telescope not at > 50 deg elevation. Cannot proceed with " + \	1✔
89	f"wiregrid calibration in current position ({az}, {el}). " + \
90	"Aborting."
91	raise RuntimeError(error)	1✔
92
93	# Check boresight angle
94	if boresight_check:	1✔
95	try:	1✔
96	assert (abs(boresight - 0) < BORESIGHT_DIFF_THRESHOLD)	1✔
97	except AssertionError:	1✔
98	error = "Telescope not at 0 deg boresight. Cannot proceed with " + \	1✔
99	f"wiregrid calibration in current position ({boresight}). " + \
100	"Aborting."
101	raise RuntimeError(error)	1✔
102
103
104	def _configure_power(continuous):	1✔
105	"""Configure the KIKUSUI power supply for rotation based on rotation type.
106
107	Args:
108	continuous (bool): Configure for continuous rotation or not.
109
110	"""
111	kikusui = run.CLIENTS['wiregrid']['kikusui']	1✔
112
113	resp = kikusui.set_v(volt=12)	1✔
114	check_response(kikusui, resp)	1✔
115
116	cfg = run.config.load_config()	1✔
117	current = cfg.get('wiregrid_motor_current', 3.0)	1✔
118
119	resp = kikusui.set_c(current=current)	1✔
120	check_response(kikusui, resp)	1✔
121
122
123	def _rotate_continuously(duration):	1✔
124	kikusui = run.CLIENTS['wiregrid']['kikusui']	1✔
125
126	# Start rotation
127	resp = kikusui.set_on()	1✔
128	check_response(kikusui, resp)	1✔
129
130	# Wait
131	time.sleep(duration)	1✔
132
133	# Stop rotation
134	resp = kikusui.set_off()	1✔
135	check_response(kikusui, resp)	1✔
136
137
138	def _check_motor_on():	1✔
139	actuator = run.CLIENTS['wiregrid']['actuator']	1✔
140
141	# Check motor is on
142	resp = actuator.acq.status()	1✔
143	if resp.session['data']['fields']['motor'] == 1:	1✔
144	print("Wiregrid motor already on.")	1✔
145	# Turn on motor if needed
146	elif resp.session['data']['fields']['motor'] == 0:	1✔
147	resp = actuator.motor_on()	1✔
148	check_response(actuator, resp)	1✔
149	else:
150	raise RuntimeError("Motor state unknown. Aborting...")	1✔
151
152
153	def _check_agents_online():	1✔
154	"""Check OCS agents related to the wiregrid are running."""
155	actuator = run.CLIENTS['wiregrid']['actuator']	1✔
156	kikusui = run.CLIENTS['wiregrid']['kikusui']	1✔
157	encoder = run.CLIENTS['wiregrid']['encoder']	1✔
158	labjack = run.CLIENTS['wiregrid']['labjack']	1✔
159
160	# wiregrid_actuator
161	resp = actuator.acq.status()	1✔
162	last_timestamp = resp.session['data']['timestamp']	1✔
163	check_running(actuator, resp)	1✔
164	_check_process_data("Actuator agent", last_timestamp)	1✔
165
166	# wiregrid_kikusui
167	resp = kikusui.IV_acq.status()	1✔
168	last_timestamp = resp.session['data']['timestamp']	1✔
169	check_running(kikusui, resp)	1✔
170	_check_process_data("Kikusui agent", last_timestamp)	1✔
171
172	# wiregrid_encoder
173	resp = encoder.acq.status()	1✔
174	last_timestamp = resp.session['data']['timestamp']	1✔
175	check_running(encoder, resp)	1✔
176	_check_process_data("Encoder agent", last_timestamp)	1✔
177
178	# labjack
179	resp = labjack.acq.status()	1✔
180	last_timestamp = resp.session['data']['timestamp']	1✔
181	check_running(labjack, resp)	1✔
182	_check_process_data("Labjack agent", last_timestamp)	1✔
183
184	# encoder data stream
185	resp = encoder.acq.status()	1✔
186	last_timestamp = resp.session['data']['fields']['encoder_data']['last_updated']	1✔
187	_check_process_data("Encoder BBB", last_timestamp)	1✔
188
189
190	def _check_temperature_sensors():	1✔
191	"""Check temperatures within safe range."""
192	labjack = run.CLIENTS['wiregrid']['labjack']	1✔
193	resp = labjack.acq.status()	1✔
194	# AIN0 > -10 C && AIN1 > -10 C (On the wiregrid)
195	_verify_temp_response(resp, 'AIN0C', -10)	1✔
196	_verify_temp_response(resp, 'AIN1C', -10)	1✔
197	# AIN2 > 0 C (Inside the electronics enclosure for the gridloader)
198	_verify_temp_response(resp, 'AIN2C', 0)	1✔
199
200
201	def _check_wiregrid_position():	1✔
202	"""Check the wiregrid position."""
NEW 203	actuator = run.CLIENTS['wiregrid']['actuator']	×
NEW 204	resp = actuator.acq.status()	×
NEW 205	ls_status = resp.session['data']['fields']['limitswitch']	×
NEW 206	if (ls_status['LSR1'] == 1 or ls_status['LSL1'] == 1) and not (ls_status['LSR2'] == 1 and ls_status['LSL2'] == 1):	×
NEW 207	position = 'outside'	×
NEW 208	elif (ls_status['LSR2'] == 1 or ls_status['LSL2'] == 1) and not (ls_status['LSR1'] == 1 or ls_status['LSL1'] == 1):	×
NEW 209	position = 'inside'	×
210	else:
NEW 211	raise RuntimeError("The wiregrid position is unknown. The limitswitch response is like this:\n" +	×
212	"LSR1: {}\n".format('ON' if ls_status['LSR1'] == 1 else 'OFF') +
213	"LSL1: {}\n".format('ON' if ls_status['LSL1'] == 1 else 'OFF') +
214	"LSR2: {}\n".format('ON' if ls_status['LSR2'] == 1 else 'OFF') +
215	"LSL2: {}\n".format('ON' if ls_status['LSL2'] == 1 else 'OFF') +
216	"Aborting...")
NEW 217	return position	×
218
219
220	def _stop_hwp_with_wiregrid():	1✔
221	"""Stop the HWP after comfirming the wiregrid is inserted."""
222	# Check the wiregrid position
NEW UNCOV 223	position = _check_wiregrid_position()	×
NEW UNCOV 224	if position == 'outside':	×
NEW UNCOV 225	raise RuntimeError("The wiregrid is not inserted. Aborting...")	×
NEW UNCOV 226	elif position == 'inside':	×
227	# Stop the HWP
NEW UNCOV 228	print("Starting to stop the HWP.")	×
NEW UNCOV 229	run.hwp.stop(active=True)	×
NEW UNCOV 230	print("The HWP stopped successfully.")	×
231	else:
NEW UNCOV 232	raise RuntimeError("Unknown wiregrid position. Aborting...")	×
233
234
235	def _spin_hwp_with_wiregrid(target_hwp_direction):	1✔
236	"""Spin the HWP to the target direction at 2Hz after comfirming the wiregrid is inserted.
237
238	Args:
239	target_hwp_direction (str): Target HWP direction, 'forward' or 'backward'.
240	"""
241	# Check the wiregrid position
NEW UNCOV 242	position = _check_wiregrid_position()	×
NEW UNCOV 243	if position == 'outside':	×
NEW UNCOV 244	raise RuntimeError("The wiregrid is not inserted. Aborting...")	×
NEW UNCOV 245	elif position == 'inside':	×
NEW UNCOV 246	if target_hwp_direction == 'forward':	×
NEW UNCOV 247	print("Starting to spin up the HWP in forward.")	×
NEW UNCOV 248	run.hwp.set_freq(freq=2.0)	×
NEW UNCOV 249	print("The HWP is spinning at 2Hz in forward successfully.")	×
NEW UNCOV 250	elif target_hwp_direction == 'backward':	×
NEW UNCOV 251	print("Starting to spin up the HWP in backward.")	×
NEW UNCOV 252	run.hwp.set_freq(freq=-2.0)	×
NEW UNCOV 253	print("The HWP is spinning at 2Hz in backward successfully.")	×
254	else:
NEW UNCOV 255	raise RuntimeError("Unknown HWP target direction. Aborting...")	×
256	else:
NEW UNCOV 257	raise RuntimeError("Unknown wiregrid position. Aborting...")	×
258
259
260	def _reverse_hwp_with_wiregrid(initial_hwp_direction, streaming=False, stepwise_before=False, stepwise_after=False):	1✔
261	"""Change the HWP direction after comfirming the wiregrid is inserted.
262
263	Args:
264	initial_hwp_direction (str): Initial HWP direction, 'forward' or 'backward'.
265	streaming (bool): Do SMuRF streaming during the HWP direction chaging or not . Default is False.
266	stepwise_before (bool): Do stepwise rotation before changing HWP direction or not. Default is False.
267	stepwise_after (bool): Do stepwise rotation after changing HWP direction or not. Default is False.
268	"""
NEW UNCOV 269	if initial_hwp_direction not in ['forward', 'backward']:	×
NEW UNCOV 270	raise RuntimeError("Initial HWP direction should be either 'forward' or 'backward'. Aborting...")	×
271
NEW UNCOV 272	current_hwp_direction = initial_hwp_direction	×
NEW UNCOV 273	if current_hwp_direction == 'forward':	×
NEW UNCOV 274	target_hwp_direction = 'backward'	×
NEW UNCOV 275	elif current_hwp_direction == 'backward':	×
NEW UNCOV 276	target_hwp_direction = 'forward'	×
277
278	# Stop and spin up reversely the HWP
NEW UNCOV 279	try:	×
280	# Enable SMuRF streams
NEW UNCOV 281	if streaming:	×
NEW UNCOV 282	stream_tag = f'wiregrid, wg_time_constant, hwp_change_to_{target_hwp_direction}'	×
NEW UNCOV 283	if stepwise_before or stepwise_after:	×
NEW UNCOV 284	stream_tag += ', wg_stepwise'	×
NEW UNCOV 285	if _check_zenith():	×
NEW UNCOV 286	stream_tag += ', wg_el90'	×
NEW UNCOV 287	run.smurf.stream('on', subtype='cal', tag=stream_tag)	×
288	# Stepwise rotation before stopping the HWP
NEW UNCOV 289	if stepwise_before:	×
NEW UNCOV 290	rotate(False)	×
291	# Stop the HWP
NEW UNCOV 292	_stop_hwp_with_wiregrid()	×
293	# Spin up the HWP in the opposite direction
NEW UNCOV 294	_spin_hwp_with_wiregrid(target_hwp_direction)	×
NEW UNCOV 295	current_hwp_direction = target_hwp_direction	×
296	# Stepwise rotation after spinning up the HWP
NEW UNCOV 297	if stepwise_after:	×
NEW UNCOV 298	rotate(False)	×
299	finally:
300	# Stop SMuRF streams
NEW UNCOV 301	if streaming:	×
NEW UNCOV 302	run.smurf.stream('off')	×
303
NEW UNCOV 304	return current_hwp_direction	×
305
306
307	# Public API
308	def insert():	1✔
309	"""Insert the wiregrid."""
310	actuator = run.CLIENTS['wiregrid']['actuator']	1✔
311	resp = actuator.insert()	1✔
312	try:	1✔
313	check_response(actuator, resp)	1✔
314	except RuntimeError as e:	1✔
315	error = "Wiregrid insertion failed. Please inspect wiregrid before " + \	1✔
316	"continuing observations.\n" + str(e)
317	raise RuntimeError(error)	1✔
318
319
320	def eject():	1✔
321	"""Eject the wiregrid."""
322	actuator = run.CLIENTS['wiregrid']['actuator']	1✔
323	resp = actuator.eject()	1✔
324	try:	1✔
325	check_response(actuator, resp)	1✔
326	except RuntimeError as e:	1✔
327	error = "Wiregrid eject failed. Please inspect wiregrid before " + \	1✔
328	"continuing observations.\n" + str(e)
329	raise RuntimeError(error)	1✔
330
331
332	def rotate(continuous, duration=30, num_laps=1, stopped_time=10.):	1✔
333	"""Rotate the wiregrid.
334
335	Rotation is either continuously for ``duration`` seconds or stepwise for
336	``num_laps``, stopping every 22.5 degrees for ``stopped_time`` seconds.
337
338	This function sets the Kikusui power settings appropriately based on the
339	rotation type.
340
341	Args:
342	continuous (bool): Rotate the grid continuously if True or not if
343	False.
344	duration (int, float): Amount of time in seconds to rotate if rotating
345	continuously. Defaults to 30 seconds.
346	num_laps (int): Number of revolutions if rotating stepwise (i.e.
347	``continuous = False``). Defaults to 1.
348	stopped_time (float): Duration of each 22.5 deg step in seconds if
349	rotating stepwise. Defaults to 10 seconds.
350
351	"""
352	kikusui = run.CLIENTS['wiregrid']['kikusui']	1✔
353
354	# Configure power settings based on rotation type
355	_configure_power(continuous)	1✔
356
357	if continuous:	1✔
358	_rotate_continuously(duration)	1✔
359	else:
360	resp = kikusui.stepwise_rotation(num_laps=num_laps,	1✔
361	stopped_time=stopped_time)
362	check_response(kikusui, resp)	1✔
363
364
365	def calibrate(continuous=False, elevation_check=True, boresight_check=True,	1✔
366	temperature_check=True):
367	"""Run a wiregrid calibration.
368
369	Args:
370	continuous (bool): Calibration by continuous rotation or not.
371	Default is False, in which the wiregrid rotates step-wisely.
372	elevation_check (bool): Check the elevation angle is in an appropriate
373	range before the calibration or not. Default is True.
374	boresight_check (bool): Check the boresight angle is in an appropriate
375	range before the calibration or not. Default is True.
376	temperature_check (bool): Check the temperature of various components
377	are within operational limits before the calibration or not.
378	Default is True.
379
380	"""
381	try:	1✔
382	_check_telescope_position(elevation_check=elevation_check,	1✔
383	boresight_check=boresight_check)
384	_check_agents_online()	1✔
385	if temperature_check:	1✔
386	_check_temperature_sensors()	1✔
387	_check_motor_on()	1✔
388
389	# Rotate for reference before insertion
390	rotate(continuous=True, duration=10)	1✔
391
392	# Enable SMuRF streams
393	if continuous:	1✔
394	rotation = 'wg_continuous'	1✔
395	else:
396	rotation = 'wg_stepwise'	1✔
397	if _check_zenith():	1✔
398	el_tag = ', wg_el90'	1✔
399	else:
400	el_tag = ''	1✔
401	run.smurf.stream('on', tag=f'wiregrid, {rotation}{el_tag}', subtype='cal')	1✔
402
403	# Insert the wiregrid
404	insert()	1✔
405
406	# Rotate the wiregrid
407	rotate(continuous)	1✔
408
409	# Eject the wiregrid
410	eject()	1✔
411	finally:
412	# Stop SMuRF streams
413	run.smurf.stream('off')	1✔
414
415
416	def time_constant(initial_hwp_direction, stepwise_first=True, stepwise_last=True, stepwise_mid=False, repeat=1):	1✔
417	"""
418	Run a wiregrid time constant measurement.
419
420	Args:
421	initial_hwp_direction (str): Initial HWP direction, 'forward' or 'backward'.
422	stepwise_first (bool): Do stepwise rotation or not before the first HWP speed change. Default is True.
423	stepwise_last (bool): Do stepwise rotation or not after the last HWP speed change. Default is True.
424	stepwise_mid (bool): Do stepwise rotation between each HWP speed change. Default is False.
425	repeat (int): Number of repeats. Default is 1.
426	If this is odd, the HWP direction will be changed to the opposite of the initial direction.
427	If this is even, the HWP direction will be the same as the initial direction.
428	"""
429
430	# Check the initial HWP direction
NEW 431	if initial_hwp_direction not in ['forward', 'backward']:	×
NEW 432	raise RuntimeError("Initial HWP direction should be either 'forward' or 'backward'. Aborting...")	×
NEW 433	current_hwp_direction = initial_hwp_direction	×
434
435	# Check the repeat
NEW 436	if repeat < 1 or not isinstance(repeat, int):	×
NEW 437	raise RuntimeError("The repeat should be int and larger than 0. Aborting...")	×
438
NEW 439	_check_agents_online()	×
NEW 440	_check_motor_on()	×
NEW 441	_check_telescope_position(elevation_check=True, boresight_check=False)	×
442
NEW 443	if _check_zenith:	×
NEW 444	el_tag = ', wg_el90'	×
445	else:
NEW 446	el_tag = ''	×
447
448	# Rotate for reference before insertion
NEW 449	rotate(continuous=True, duration=10)	×
450
451	# Bias step (the wire grid is off the window)
NEW 452	bs_tag = f'biasstep, wg_time_constant, wg_ejected, hwp_2hz_{current_hwp_direction}' + el_tag	×
NEW 453	run.smurf.bias_step(tag=bs_tag, concurrent=True)	×
454
455	# Insert the wiregrid with streaming
NEW 456	time.sleep(5)	×
NEW 457	try:	×
458	# Enable SMuRF streams
NEW 459	stream_tag = f'wg_time_constant, wg_inserting, hwp_2hz_{current_hwp_direction}' + el_tag	×
NEW 460	run.smurf.stream('on', tag=stream_tag, subtype='cal')	×
461	# Insert the wiregrid
NEW 462	insert()	×
463	finally:
464	# Stop SMuRF streams
NEW 465	run.smurf.stream('off')	×
NEW 466	time.sleep(5)	×
467
NEW UNCOV 468	for i in range(repeat):	×
469	# Bias step (the wire grid is on the window)
NEW UNCOV 470	bs_tag = f'biasstep, wg_time_constant, wg_inserted, hwp_2hz_{current_hwp_direction}' + el_tag	×
NEW UNCOV 471	run.smurf.bias_step(tag=bs_tag, concurrent=True)	×
472
NEW UNCOV 473	stepwise_before = False	×
NEW UNCOV 474	stepwise_after = False	×
NEW UNCOV 475	if stepwise_first and i == 0:	×
476	# Stepwise rotation before the first HWP speed change
NEW UNCOV 477	stepwise_before = True	×
NEW UNCOV 478	if stepwise_mid and i != 0:	×
479	# Stepwise rotation between changing HWP speed
NEW UNCOV 480	stepwise_before = True	×
NEW UNCOV 481	if stepwise_last and i == repeat-1:	×
482	# Stepwise rotation after the last HWP speed change
NEW UNCOV 483	stepwise_after = True	×
484
485	# Spin the HWP in the opposite direction of the initial direction with streaming and stepwise rotation
NEW UNCOV 486	current_hwp_direction = _reverse_hwp_with_wiregrid(current_hwp_direction,	×
487	streaming=True,
488	stepwise_before=stepwise_before,
489	stepwise_after=stepwise_after)
490
491	# Bias step (the wire grid is on the window)
NEW UNCOV 492	bs_tag = f'biasstep, wg_time_constant, wg_inserted, hwp_2hz_{current_hwp_direction}' + el_tag	×
NEW UNCOV 493	run.smurf.bias_step(tag=bs_tag, concurrent=True)	×
494
495	# Eject the wiregrid with streaming
NEW UNCOV 496	time.sleep(5)	×
NEW UNCOV 497	try:	×
498	# Enable SMuRF streams
NEW UNCOV 499	stream_tag = f'wg_time_constant, wg_ejecting, hwp_2hz_{current_hwp_direction}' + el_tag	×
NEW UNCOV 500	run.smurf.stream('on', tag=stream_tag, subtype='cal')	×
501	# Eject the wiregrid
NEW UNCOV 502	eject()	×
503	finally:
504	# Stop SMuRF streams
NEW UNCOV 505	run.smurf.stream('off')	×
NEW UNCOV 506	time.sleep(5)	×
507
508	# Bias step (the wire grid is off the window)
NEW UNCOV 509	bs_tag = f'biasstep, wg_time_constant, wg_ejected, hwp_2hz_{current_hwp_direction}' + el_tag	×
NEW UNCOV 510	run.smurf.bias_step(tag=bs_tag, concurrent=True)	×

simonsobs / sorunlib / 11434250339

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous