Accessing the scheduler using `sipyco`

Hi,

We have an issue where our experiment will occasionally hang indefinitely without moving to the next experiment shot. No error is thrown when this happens. I believe it is actually an issue to do with the Kasli overheating, as it happens more frequently when the Kasli's ventilation is reduced.

To fix this problem I am planning on using the scheduler to delete the RID in question--the experiment runs smoothly when a hanging RID is deleted via the ARTIQ dashboard. To implement this fix I need to access the scheduler using sipyco. Are there any code examples I can reference to accomplish this?

For those curious, the code I am using the read the dashboard status is below:

import asyncio
from time import time_ns

class ArtiqState():
    

    def __init__(self, master_ip=ARTIQ_MASTER_IP, notify_port=ARTIQ_NOTIFY_PORT, sleep_interval=1):
        # Store IP and port
        self.master_ip = master_ip
        self.notify_port = notify_port
        

        # Store sleep interval
        self.sleep_interval = sleep_interval
        

        # Create dictionaries to track state
        self.artiq_schedule = {}
        self.artiq_dataset = {}
        

    async def connect(self, dataset_callback=None, schedule_callback=None):
        # Create subscribers
        dset_sub = Subscriber('datasets', lambda x: x, lambda mod: ArtiqState._update_state(self.artiq_dataset, dataset_callback, mod))
        sched_sub = Subscriber('schedule', lambda x: x, lambda mod: ArtiqState._update_state(self.artiq_schedule, schedule_callback, mod))
        

        # Connect to ARTIQ master
        await dset_sub.connect(self.master_ip, self.notify_port)
        await sched_sub.connect(self.master_ip, self.notify_port)
        

        # Enter endless loop to keep connections alive
        while True:
            await asyncio.sleep(self.sleep_interval)
        

    def _update_state(target, callback, mod):
        # Target initialization
        if mod['action'] == 'init':
            target.update(mod['struct'])
        # Item modified perform update
        elif mod['action'] == 'setitem':
            ArtiqState._update_at_path(target, mod['path'], mod['key'], mod['value'])
        # Item deleted
        elif mod['action'] == 'delitem':
            del target[mod['key']]
        # If callback is not None, call
        if callback is not None:
            callback(target)
    

    def _update_at_path(target, path, key, value):
        """Updates a dictionary containing sub dictionarys or lists specified by the path. This is used to update values when the ARTIQ datasets and schedule send out setitem updates."""
        if len(path) == 0:
            target[key] = value
        else:
            ArtiqState._update_at_path(target[path[0]], path[1:], key, value)

if __name__ == '__main__':
    

    ar = ArtiqState()
        

    async def mon():
        while True:
            await asyncio.sleep(1)
            if 'misc.scan_idx' in ar.artiq_dataset and 'misc.scan_max' in ar.artiq_dataset and 'timestamp' in ar.artiq_dataset:
                idx = ar.artiq_dataset['misc.scan_idx'][1][0]
                max_idx = ar.artiq_dataset['misc.scan_max'][1]
                print(f'{idx+1}/{max_idx}')
                

                lt = ar.artiq_dataset['timestamp']
                print(f'last timestamp: {lt[1][idx]}')
                

                if lt[1][idx] != -1:
                

                    current_rid = -1
                    for rid in list(ar.artiq_schedule.keys()):
                        if ar.artiq_schedule[rid]['status'] == 'running':
                            current_rid = rid
                            break
                                        

                            

                    time_delta_s = (time_ns()-lt[1][idx])/1e9
                    print(f'time diff={time_delta_s}')
                    

                    if time_delta_s > 60:
                        print('No shots for 1 min, deleting current RID to unfreeze ARTIQ.')
        

    async def spawn():
        await asyncio.gather(ar.connect(), mon())
        

    asyncio.run(spawn())