language,master,run: support raw access to DDB from experiments. Closes #123

This commit is contained in:
Sebastien Bourdeauducq 2015-10-04 18:29:39 +08:00
parent 512bc79484
commit b3584bc190
8 changed files with 31 additions and 10 deletions

View File

@ -74,11 +74,12 @@ def main():
repo_backend = GitBackend(args.repository) repo_backend = GitBackend(args.repository)
else: else:
repo_backend = FilesystemBackend(args.repository) repo_backend = FilesystemBackend(args.repository)
repository = Repository(repo_backend, log.log) repository = Repository(repo_backend, ddb.get_ddb, log.log)
atexit.register(repository.close) atexit.register(repository.close)
repository.scan_async() repository.scan_async()
worker_handlers = { worker_handlers = {
"get_ddb": ddb.get_ddb,
"get_device": ddb.get, "get_device": ddb.get,
"get_parameter": pdb.get, "get_parameter": pdb.get,
"set_parameter": pdb.set, "set_parameter": pdb.set,

View File

@ -13,6 +13,7 @@ import h5py
from artiq.language.environment import EnvExperiment from artiq.language.environment import EnvExperiment
from artiq.protocols.file_db import FlatFileDB from artiq.protocols.file_db import FlatFileDB
from artiq.master.ddb import DDB
from artiq.master.worker_db import DeviceManager, ResultDB from artiq.master.worker_db import DeviceManager, ResultDB
from artiq.tools import * from artiq.tools import *
@ -111,7 +112,7 @@ def run(with_file=False):
args = get_argparser(with_file).parse_args() args = get_argparser(with_file).parse_args()
init_logger(args) init_logger(args)
dmgr = DeviceManager(FlatFileDB(args.ddb), dmgr = DeviceManager(DDB(args.ddb),
virtual_devices={"scheduler": DummyScheduler()}) virtual_devices={"scheduler": DummyScheduler()})
pdb = FlatFileDB(args.pdb) pdb = FlatFileDB(args.pdb)
pdb.hooks.append(SimpleParamLogger()) pdb.hooks.append(SimpleParamLogger())

View File

@ -182,6 +182,12 @@ class HasEnvironment:
attribute are the same.""" attribute are the same."""
setattr(self, key, self.get_argument(key, processor, group)) setattr(self, key, self.get_argument(key, processor, group))
def get_ddb(self):
"""Returns the full contents of the device database."""
if self.__parent is not None:
return self.__parent.get_ddb()
return self.__dmgr.get_ddb()
def get_device(self, key): def get_device(self, key):
"""Creates and returns a device driver.""" """Creates and returns a device driver."""
if self.__parent is not None: if self.__parent is not None:

View File

@ -17,5 +17,8 @@ class DDB:
if k not in self.data.read or self.data.read[k] != new_data[k]: if k not in self.data.read or self.data.read[k] != new_data[k]:
self.data[k] = new_data[k] self.data[k] = new_data[k]
def get_ddb(self):
return self.data.read
def get(self, key): def get(self, key):
return self.data.read[key] return self.data.read[key]

View File

@ -12,12 +12,15 @@ from artiq.tools import exc_to_warning
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
async def _scan_experiments(wd, log): async def _scan_experiments(wd, get_ddb, log):
r = dict() r = dict()
for f in os.listdir(wd): for f in os.listdir(wd):
if f.endswith(".py"): if f.endswith(".py"):
try: try:
worker = Worker({"log": lambda message: log("scan", message)}) worker = Worker({
"get_ddb": get_ddb,
"log": lambda message: log("scan", message)
})
try: try:
description = await worker.examine(os.path.join(wd, f)) description = await worker.examine(os.path.join(wd, f))
finally: finally:
@ -53,8 +56,9 @@ def _sync_explist(target, source):
class Repository: class Repository:
def __init__(self, backend, log_fn): def __init__(self, backend, get_ddb_fn, log_fn):
self.backend = backend self.backend = backend
self.get_ddb_fn = get_ddb_fn
self.log_fn = log_fn self.log_fn = log_fn
self.cur_rev = self.backend.get_head_rev() self.cur_rev = self.backend.get_head_rev()
@ -77,7 +81,7 @@ class Repository:
wd, _ = self.backend.request_rev(new_cur_rev) wd, _ = self.backend.request_rev(new_cur_rev)
self.backend.release_rev(self.cur_rev) self.backend.release_rev(self.cur_rev)
self.cur_rev = new_cur_rev self.cur_rev = new_cur_rev
new_explist = await _scan_experiments(wd, self.log_fn) new_explist = await _scan_experiments(wd, self.get_ddb_fn, self.log_fn)
_sync_explist(self.explist, new_explist) _sync_explist(self.explist, new_explist)
finally: finally:

View File

@ -244,7 +244,7 @@ class Worker:
def register(class_name, name, arguments): def register(class_name, name, arguments):
r[class_name] = {"name": name, "arguments": arguments} r[class_name] = {"name": name, "arguments": arguments}
self.register_experiment = register self.register_experiment = register
await self._worker_action({"action": "examine", await self._worker_action({"action": "examine", "file": file},
"file": file}, timeout) timeout)
del self.register_experiment del self.register_experiment
return r return r

View File

@ -136,6 +136,10 @@ class DeviceManager:
self.virtual_devices = virtual_devices self.virtual_devices = virtual_devices
self.active_devices = OrderedDict() self.active_devices = OrderedDict()
def get_ddb(self):
"""Returns the full contents of the device database."""
return self.ddb.get_ddb()
def get(self, name): def get(self, name):
"""Get the device driver or controller client corresponding to a """Get the device driver or controller client corresponding to a
device database entry.""" device database entry."""

View File

@ -122,7 +122,9 @@ register_experiment = make_parent_action("register_experiment",
"class_name name arguments") "class_name name arguments")
class DummyDMGR: class ExamineDMGR:
get_ddb = make_parent_action("get_ddb", "")
def get(self, name): def get(self, name):
return None return None
@ -208,7 +210,7 @@ def main():
f.close() f.close()
put_object({"action": "completed"}) put_object({"action": "completed"})
elif action == "examine": elif action == "examine":
examine(DummyDMGR(), DummyPDB(), ResultDB(), obj["file"]) examine(ExamineDMGR(), DummyPDB(), ResultDB(), obj["file"])
put_object({"action": "completed"}) put_object({"action": "completed"})
elif action == "terminate": elif action == "terminate":
break break