m-labs · Apr 2, 2016 · Apr 2, 2016 · Apr 2, 2016 · Apr 2, 2016 · Apr 2, 2016
diff --git a/artiq/frontend/artiq_master.py b/artiq/frontend/artiq_master.py
@@ -63,30 +63,30 @@ def main():
     dataset_db = DatasetDB(args.dataset_db)
     dataset_db.start()
     atexit_register_coroutine(dataset_db.stop)
+    worker_handlers = dict()
 
     if args.git:
         repo_backend = GitBackend(args.repository)
     else:
         repo_backend = FilesystemBackend(args.repository)
-    experiment_db = ExperimentDB(repo_backend, device_db.get_device_db)
+    experiment_db = ExperimentDB(repo_backend, worker_handlers)
     atexit.register(experiment_db.close)
-    experiment_db.scan_repository_async()
 
-    worker_handlers = {
+    scheduler = Scheduler(RIDCounter(), worker_handlers, experiment_db)
+    scheduler.start()
+    atexit_register_coroutine(scheduler.stop)
+
+    worker_handlers.update({
         "get_device_db": device_db.get_device_db,
         "get_device": device_db.get,
         "get_dataset": dataset_db.get,
-        "update_dataset": dataset_db.update
-    }
-    scheduler = Scheduler(RIDCounter(), worker_handlers, experiment_db)
-    worker_handlers.update({
+        "update_dataset": dataset_db.update,
         "scheduler_submit": scheduler.submit,
         "scheduler_delete": scheduler.delete,
         "scheduler_request_termination": scheduler.request_termination,
-        "scheduler_get_status": scheduler.get_status
+        "scheduler_get_status": scheduler.get_status        
     })
-    scheduler.start()
-    atexit_register_coroutine(scheduler.stop)
+    experiment_db.scan_repository_async()
 
     bind = bind_address_from_args(args)
 

diff --git a/artiq/gui/experiments.py b/artiq/gui/experiments.py
@@ -113,6 +113,7 @@ async def _recompute_argument(self, name):
         except:
             logger.error("Could not recompute argument '%s' of '%s'",
                          name, self.expurl, exc_info=True)
+            return
         argument = self.manager.get_submission_arguments(self.expurl)[name]
 
         procdesc = arginfo[name][0]

diff --git a/artiq/master/experiments.py b/artiq/master/experiments.py
@@ -14,8 +14,8 @@
 
 
 async def _get_repository_entries(entry_dict,
-                                  root, filename, get_device_db):
-    worker = Worker({"get_device_db": get_device_db})
+                                  root, filename, worker_handlers):
+    worker = Worker(worker_handlers)
     try:
         description = await worker.examine("scan", os.path.join(root, filename))
     except:
@@ -45,7 +45,7 @@ async def _get_repository_entries(entry_dict,
         entry_dict[name] = entry
 
 
-async def _scan_experiments(root, get_device_db, subdir=""):
+async def _scan_experiments(root, worker_handlers, subdir=""):
     entry_dict = dict()
     for de in os.scandir(os.path.join(root, subdir)):
         if de.name.startswith("."):
@@ -54,13 +54,13 @@ async def _scan_experiments(root, get_device_db, subdir=""):
             filename = os.path.join(subdir, de.name)
             try:
                 await _get_repository_entries(
-                    entry_dict, root, filename, get_device_db)
+                    entry_dict, root, filename, worker_handlers)
             except Exception as exc:
                 logger.warning("Skipping file '%s'", filename,
                     exc_info=not isinstance(exc, WorkerInternalException))
         if de.is_dir():
             subentries = await _scan_experiments(
-                root, get_device_db,
+                root, worker_handlers,
                 os.path.join(subdir, de.name))
             entries = {de.name + "/" + k: v for k, v in subentries.items()}
             entry_dict.update(entries)
@@ -77,9 +77,9 @@ def _sync_explist(target, source):
 
 
 class ExperimentDB:
-    def __init__(self, repo_backend, get_device_db_fn):
+    def __init__(self, repo_backend, worker_handlers):
         self.repo_backend = repo_backend
-        self.get_device_db_fn = get_device_db_fn
+        self.worker_handlers = worker_handlers
 
         self.cur_rev = self.repo_backend.get_head_rev()
         self.repo_backend.request_rev(self.cur_rev)
@@ -107,7 +107,7 @@ async def scan_repository(self, new_cur_rev=None):
             self.repo_backend.release_rev(self.cur_rev)
             self.cur_rev = new_cur_rev
             self.status["cur_rev"] = new_cur_rev
-            new_explist = await _scan_experiments(wd, self.get_device_db_fn)
+            new_explist = await _scan_experiments(wd, self.worker_handlers)
 
             _sync_explist(self.explist, new_explist)
         finally:
@@ -123,7 +123,7 @@ async def examine(self, filename, use_repository=True):
             revision = self.cur_rev
             wd, _ = self.repo_backend.request_rev(revision)
             filename = os.path.join(wd, filename)
-        worker = Worker({"get_device_db": self.get_device_db_fn})
+        worker = Worker(self.worker_handlers)
         try:
             description = await worker.examine("examine", filename)
         finally:

diff --git a/artiq/master/worker_impl.py b/artiq/master/worker_impl.py
@@ -124,14 +124,6 @@ def get(name):
         return None
 
 
-class DummyDatasetMgr:
-    def set(key, value, broadcast=False, persist=False, save=True):
-        return None
-
-    def get(key):
-        pass
-
-
 def examine(device_mgr, dataset_mgr, file):
     module = file_import(file)
     for class_name, exp_class in module.__dict__.items():
@@ -244,7 +236,7 @@ def main():
                     f.close()
                 put_object({"action": "completed"})
             elif action == "examine":
-                examine(ExamineDeviceMgr, DummyDatasetMgr, obj["file"])
+                examine(ExamineDeviceMgr, ParentDatasetDB, obj["file"])
                 put_object({"action": "completed"})
             elif action == "terminate":
                 break

diff --git a/artiq/protocols/remote_exec.py b/artiq/protocols/remote_exec.py
@@ -2,33 +2,27 @@
 import inspect
 
 from artiq.protocols.pc_rpc import simple_server_loop
-from artiq.protocols.pc_rpc import Client as RPCClient
 
 
 __all__ = ["RemoteExecServer", "RemoteExecClient", "simple_rexec_server_loop"]
 
 
 class RemoteExecServer:
-    def __init__(self, target):
-        self.target = target
-        self.namespace = dict()
+    def __init__(self, initial_namespace):
+        self.namespace = dict(initial_namespace)
 
     def add_code(self, code):
         exec(code, self.namespace)
 
     def call(self, function, *args, **kwargs):
-        return self.namespace[function](self, *args, **kwargs)
-
-
-class RemoteExecClient(RPCClient):
-    def transfer_obj_source(self, obj):
-        self.add_code(self, inspect.getsource(obj))
+        return self.namespace[function](*args, **kwargs)
 
 
 def simple_rexec_server_loop(target_name, target, host, port,
                              description=None):
+    initial_namespace = {"controller_driver": target}
     targets = {
         target_name: target,
-        target_name + "_rexec": lambda: RemoteExecServer(target)
+        target_name + "_rexec": lambda: RemoteExecServer(initial_namespace)
     }
     simple_server_loop(targets, host, port, description)
diff --git a/examples/fit_image.py b/examples/fit_image.py
@@ -1,7 +1,8 @@
+"""Image processing with SciPy example"""
+
 import numpy as np
 from scipy.optimize import least_squares
 from scipy import constants
-# from numba import jit
 
 
 class Fit:
@@ -81,7 +82,6 @@ def guess(self):
         # TODO: this is usually smarter, based on self.data and self.meta
         return [1000, 100, 100, 2000, 4, 4, 2000, 20, 20]
 
-    # @jit
     def model(self, param):
         p = self.variables_dict(param)
         x, y = self.xy

diff --git a/examples/master/repository/arguments_demo.py b/examples/master/repository/arguments_demo.py
@@ -37,7 +37,10 @@ def do(self):
 
 class ArgumentsDemo(EnvExperiment):
     def build(self):
-        self.setattr_argument("pyon_value", PYONValue(None))
+        # change the "foo" dataset and click the "recompute argument"
+        # buttons.
+        self.setattr_argument("pyon_value",
+            PYONValue(self.get_dataset("foo", default=42)))
         self.setattr_argument("number", NumberValue(42e-6,
                                                     unit="us", scale=1e-6,
                                                     ndecimals=4))

diff --git a/examples/master/repository/remote_exec_demo.py b/examples/master/repository/remote_exec_demo.py
@@ -0,0 +1,72 @@
+import time
+import inspect
+
+from artiq.experiment import *
+
+import remote_exec_processing
+
+
+class RemoteExecDemo(EnvExperiment):
+    def build(self):
+        self.setattr_device("camera_sim")
+        self.setattr_device("scheduler")
+        self.setattr_argument("remote_exec", BooleanValue(False))
+        self.setattr_argument("show_picture", BooleanValue(True), "Local options")
+        self.setattr_argument("enable_fit", BooleanValue(True), "Local options")
+        if self.remote_exec:
+            self.setattr_device("camera_sim_rexec")
+
+    def prepare(self):
+        if self.remote_exec:
+            self.camera_sim_rexec.add_code(
+                inspect.getsource(remote_exec_processing))
+
+    def set_dataset(self, name, *args, **kwargs):
+        EnvExperiment.set_dataset(self, "rexec_demo." + name,
+                                  *args, **kwargs)
+
+    def transfer_parameters(self, parameters):
+        w, h, cx, cy = parameters
+        self.set_dataset("gaussian_w", w, save=False, broadcast=True)
+        self.set_dataset("gaussian_h", h, save=False, broadcast=True)
+        self.set_dataset("gaussian_cx", cx, save=False, broadcast=True)
+        self.set_dataset("gaussian_cy", cy, save=False, broadcast=True)
+
+    def fps_meter(self):
+        t = time.monotonic()
+        if hasattr(self, "last_pt_update"):
+            self.iter_count += 1
+            dt = t - self.last_pt_update
+            if dt >= 5:
+                pt = dt/self.iter_count
+                self.set_dataset("picture_pt", pt, save=False, broadcast=True)
+                self.last_pt_update = t
+                self.iter_count = 0
+        else:
+            self.last_pt_update = t
+            self.iter_count = 0
+
+    def run_local(self):
+        while True:
+            self.fps_meter()
+            data = self.camera_sim.get_picture()
+            if self.show_picture:
+                self.set_dataset("picture", data, save=False, broadcast=True)
+            if self.enable_fit:
+                self.transfer_parameters(remote_exec_processing.fit(data))
+            self.scheduler.pause()
+
+    def run_remote(self):
+        while True:
+            self.fps_meter()
+            self.transfer_parameters(self.camera_sim_rexec.call("get_and_fit"))
+            self.scheduler.pause()
+
+    def run(self):
+        try:
+            if self.remote_exec:
+                self.run_remote()
+            else:
+                self.run_local()
+        except TerminationRequested:
+            pass
diff --git a/examples/master/repository/remote_exec_processing.py b/examples/master/repository/remote_exec_processing.py
@@ -0,0 +1,30 @@
+import numpy as np
+from numba import jit
+from scipy.optimize import least_squares
+
+
+@jit(nopython=True)
+def compute_gaussian(r, img_w, img_h,
+                     gaussian_w, gaussian_h,
+                     gaussian_cx, gaussian_cy):
+    for y in range(img_h):
+        for x in range(img_w):
+            ds = ((gaussian_cx-x)/gaussian_w)**2 
+            ds += ((gaussian_cy-y)/gaussian_h)**2
+            r[x, y] = np.exp(-ds/2)
+
+
+def fit(data):
+    img_w, img_h = data.shape
+    def err(parameters):
+        r = np.empty((img_w, img_h))
+        compute_gaussian(r, img_w, img_h, *parameters)
+        r -= data
+        return r.ravel()
+    guess = [12, 15, img_w/2, img_h/2]
+    res = least_squares(err, guess)
+    return res.x
+
+
+def get_and_fit():
+    return fit(controller_driver.get_picture())
diff --git a/examples/remote_exec_controller.py b/examples/remote_exec_controller.py
@@ -7,7 +7,7 @@
 
 
 @jit(nopython=True)
-def compute_picture(r, img_h, img_w,
+def compute_picture(r, img_w, img_h,
                     gaussian_w, gaussian_h,
                     gaussian_cx, gaussian_cy,
                     noise_level):
@@ -43,7 +43,7 @@ def set_noise_level(self, noise_level):
 
     def get_picture(self):
         r = np.empty((self.img_w, self.img_h))
-        compute_picture(r, self.img_h, self.img_w,
+        compute_picture(r, self.img_w, self.img_h,
             self.gaussian_w, self.gaussian_h,
             self.gaussian_cx, self.gaussian_cy,
             self.noise_level)