added job queueing to the tool set (needed for calibartion mangagment systems)

core_tools/job_mgnt/job_meta.py

+from core_tools.sweeps.progressbar import progress_bar
+from core_tools.sweeps.sweep_utility import KILL_EXP
+
+def run_wrapper(run_function):
+    def run(*args, **kwargs):
+        args[0].n = progress_bar(args[0].n_tot)
+        try:
+            returns = run_function(*args, **kwargs)
+        except KILL_EXP:
+            print('kill signal for the current experiment received.')
+            returns = None
+        
+        args[0].n.close()
+        args[0].n = 0
+
+        return returns
+    return run
+
+class job_meta(type):  
+    def __new__(cls,name, bases, dct):
+        if 'run' not in dct:
+            raise ValueError('Please define a run function in your job class.')
+
+        x = super().__new__(cls, name, bases, dct)
+        x.run = run_wrapper(dct['run'])
+
+        x.n_tot = 0
+        x.n = 0
+        x.KILL = False
+        return x
+        
+if __name__ == '__main__':
+    import time
+
+    class pulse_lib_sweep_virt(metaclass=job_meta):
+        def __init__(self, n_steps):
+            self.n_tot = n_steps
+
+        def run(self):
+            for i in range(self.n_tot):
+                time.sleep(0.01)
+                self.n += 1
+    
+    a = pulse_lib_sweep_virt(5)
+    a.run()
+    a.KILL = True
+
+    b = pulse_lib_sweep_virt(5)
+    b.run()
+    print(b.KILL)
\ No newline at end of file

core_tools/job_mgnt/job_mgmt.py

+from dataclasses import dataclass, field
+from typing import Any
+import time
+
+import threading, queue
+from queue import PriorityQueue
+
+@dataclass(order=True)
+class ExperimentJob:
+    priority: float
+    job: Any = field(compare=False)
+
+    def kill(self):
+        self.job.KILL = True
+
+class queue_mgr():
+    def __init__(self):
+        self.q = PriorityQueue()
+        self.job_refs = list()
+
+        def worker():
+            while True:
+                n_jobs = self.q.qsize()
+                if n_jobs != 0:
+                    print('{} items queued.'.format(n_jobs))
+                    print('Starting new job.')
+                    job_object = self.q.get()
+                    try:
+                        print(job_object.job.KILL)
+                        if job_object.job.KILL != True:
+                            job_object.job.run()
+                    except:
+                        print('an exception in the job occurred? Going to the next job.')
+                    self.q.task_done()
+                else:
+                    # 200ms sleep.
+                    time.sleep(0.2)
+
+        self.worker_thread = threading.Thread(target=worker, daemon=True).start()
+
+    def put(self, job):
+        '''
+        put a job into the measurement queue
+
+        Args:
+            job (ExperimentJob) : job object
+        '''
+        self.q.put(job)
+        self.job_refs.append(job)
+
+    def kill(self, job):
+        '''
+        kill a certain job that has been submitted to the queue
+
+        Args:
+            job (ExperimentJob) : job object
+        '''
+        job.KILL = True
+
+    def killall(self):
+        '''
+        kill all the jobs
+        '''
+        for job in self.job_refs:
+            job.kill()
+
+        self.job_refs = []
+
+    def join(self):
+        self.q.join()
+
+    @property
+    def n_jobs(self):
+        return self.q.qsize()
+
+
+#%%
+if __name__ == '__main__':
+    from core_tools.sweeps.sweeps import do1D, do2D
+    import os
+    import qcodes as qc
+    from qcodes.dataset.sqlite.database import initialise_or_create_database_at
+    from qcodes.dataset.experiment_container import load_or_create_experiment
+    from qcodes.instrument.specialized_parameters import ElapsedTimeParameter
+    class MyCounter(qc.Parameter):
+        def __init__(self, name):
+            # only name is required
+            super().__init__(name, label='Times this has been read',
+                             docstring='counts how many times get has been called '
+                                       'but can be reset to any integer >= 0 by set')
+            self._count = 0
+
+        # you must provide a get method, a set method, or both.
+        def get_raw(self):
+            self._count += 1
+            return self._count
+
+        def set_raw(self, val):
+            self._count = val
+
+    tutorial_db_path = os.path.join(os.getcwd(), 'linking_datasets_tutorial.db')
+    initialise_or_create_database_at(tutorial_db_path)
+    load_or_create_experiment('tutorial', 'no sample')
+
+    my_param = MyCounter('test_instr')
+
+    x = qc.Parameter(name='x', label='Voltage_x', unit='V',
+              set_cmd=None, get_cmd=None)
+    y = qc.Parameter(name='y', label='Voltage_y', unit='V',
+              set_cmd=None, get_cmd=None)
+    timer = ElapsedTimeParameter('time')
+
+    scan1 = do2D(x, 0, 20, 20, 0.0, y, 0, 80, 30, 0.1, my_param)
+    scan2 = do2D(x, 0, 20, 20, 0.0, timer, 0, 80, 30, .1, my_param)
+    scan3 = do1D(x, 0, 100, 50, 0.1 , my_param, reset_param=True)
+
+    q = queue_mgr()
+    job1 = ExperimentJob(1, scan1)
+    job2 = ExperimentJob(1, scan2)
+    job3 = ExperimentJob(1, scan3)
+    q.put(job1)
+    q.put(job2)
+    q.put(job3)
+    
+    q.killall()
+    scan1 = do2D(x, 0, 20, 20, 0.0, y, 0, 80, 30, 0.1, my_param)
+    scan2 = do2D(x, 0, 20, 20, 0.0, timer, 0, 80, 30, .1, my_param)
+    job1 = ExperimentJob(1, scan1)
+    job2 = ExperimentJob(1, scan2)
+    q.put(job1)
+    q.put(job2)
\ No newline at end of file