mirror of
https://github.com/m-labs/artiq.git
synced 2024-12-11 12:46:37 +08:00
154 lines
4.3 KiB
Python
154 lines
4.3 KiB
Python
"""
|
|
Connection to device, parameter and result database.
|
|
|
|
"""
|
|
|
|
class _AttributeKind:
|
|
pass
|
|
|
|
|
|
class Device(_AttributeKind):
|
|
"""Represents a device for ``AutoDB`` to process.
|
|
|
|
"""
|
|
pass
|
|
|
|
|
|
class NoDefault:
|
|
"""Represents the absence of a default value for ``Parameter``
|
|
and ``Argument``.
|
|
|
|
"""
|
|
pass
|
|
|
|
|
|
class Parameter(_AttributeKind):
|
|
"""Represents a parameter (from the database) for ``AutoDB``
|
|
to process.
|
|
|
|
:param default: Default value of the parameter to be used if not found
|
|
in the database.
|
|
|
|
"""
|
|
def __init__(self, default=NoDefault):
|
|
self.default = default
|
|
|
|
|
|
class Argument(_AttributeKind):
|
|
"""Represents an argument (specifiable at instance creation) for
|
|
``AutoDB`` to process.
|
|
|
|
:param default: Default value of the argument to be used if not specified
|
|
at instance creation.
|
|
|
|
"""
|
|
def __init__(self, default=NoDefault):
|
|
self.default = default
|
|
|
|
|
|
class Result(_AttributeKind):
|
|
"""Represents a result for ``AutoDB`` to process.
|
|
|
|
"""
|
|
pass
|
|
|
|
|
|
class AutoDB:
|
|
"""Base class to automate device, parameter and result database access.
|
|
|
|
Drivers and experiments should in most cases overload this class to
|
|
obtain the parameters and devices (including the core device) that they
|
|
need, report results, and modify parameters.
|
|
|
|
:param dbh: database hub to use. If ``None``, all devices and parameters
|
|
must be supplied as keyword arguments, and reporting results and
|
|
modifying parameters is not supported.
|
|
|
|
"""
|
|
class DBKeys:
|
|
pass
|
|
|
|
@staticmethod
|
|
def realtime_results():
|
|
return dict()
|
|
|
|
def __init__(self, dbh=None, **kwargs):
|
|
self.dbh = dbh
|
|
|
|
dbkeys = self.DBKeys()
|
|
if getattr(dbkeys, "implicit_core", True):
|
|
if hasattr(dbkeys, "core"):
|
|
raise ValueError(
|
|
"Set implicit_core to False when"
|
|
" core is explicitly specified")
|
|
dbkeys.core = Device()
|
|
|
|
for k, v in kwargs.items():
|
|
object.__setattr__(self, k, v)
|
|
|
|
for k in dir(dbkeys):
|
|
if k not in self.__dict__:
|
|
ak = getattr(dbkeys, k)
|
|
if isinstance(ak, Argument):
|
|
if ak.default is NoDefault:
|
|
raise AttributeError(
|
|
"No value specified for argument '{}'".format(k))
|
|
object.__setattr__(self, k, ak.default)
|
|
elif isinstance(ak, Device):
|
|
try:
|
|
dev = self.dbh.get_device(k)
|
|
except KeyError:
|
|
raise KeyError("Device '{}' not found".format(k))
|
|
object.__setattr__(self, k, dev)
|
|
|
|
self.build()
|
|
|
|
def __getattr__(self, name):
|
|
ak = getattr(self.DBKeys, name)
|
|
if isinstance(ak, Parameter):
|
|
try:
|
|
if self.dbh is None:
|
|
raise KeyError
|
|
return self.dbh.get_parameter(name)
|
|
except KeyError:
|
|
if ak.default is not NoDefault:
|
|
return ak.default
|
|
else:
|
|
raise AttributeError("Parameter '{}' not in database"
|
|
" and without default value"
|
|
.format(name))
|
|
elif isinstance(ak, Result):
|
|
try:
|
|
return self.dbh.get_result(name)
|
|
except KeyError:
|
|
raise AttributeError("Result '{}' not found".format(name))
|
|
else:
|
|
raise ValueError
|
|
|
|
def __setattr__(self, name, value):
|
|
try:
|
|
ak = getattr(self.DBKeys, name)
|
|
except AttributeError:
|
|
object.__setattr__(self, name, value)
|
|
else:
|
|
if isinstance(ak, Parameter):
|
|
self.dbh.set_parameter(name, value)
|
|
elif isinstance(ak, Result):
|
|
self.dbh.set_result(name, value)
|
|
else:
|
|
raise ValueError
|
|
|
|
@classmethod
|
|
def get_realtime_results():
|
|
return dict()
|
|
|
|
def build(self):
|
|
"""This is called by ``__init__`` after the parameter initialization
|
|
is done.
|
|
|
|
The user may overload this method to complete the object's
|
|
initialization with all parameters available.
|
|
|
|
"""
|
|
pass
|