"""Main loop functions for running MeaVis measurements."""
import collections
import importlib
import logging
import sys
import threading
import meavis._detail.debug
import meavis._detail.default
import meavis._detail.inject
import meavis._detail.utils
import meavis.config
import meavis.engines.completer
import meavis.markup
import meavis.measurements
import meavis.parameters
self_module = sys.modules[__name__]
[docs]def clear(module_name):
"""Clear injected names by users."""
module = importlib.import_module("meavis.{}".format(module_name))
for injected_name in [
key
for key, value in module.__dict__.items()
if hasattr(value, "_meavis_injected") and value._meavis_injected
]:
delattr(module, injected_name)
logging.getLogger("meavis").info(
"Unregister injected class named {}.".format(injected_name)
)
[docs]def inject(instruments):
"""Inject instruments."""
meavis._detail.debug.parameter_isinstance(
"instruments", collections.abc.Mapping
)
modules = {
instrument_name: instrument["module"]
for instrument_name, instrument in meavis.markup.visit_instruments(
instruments, None, "*"
).items()
if "module" in instrument
}
all_tags = ["parameters", "measurements"]
for meavis_name, constructor_map in meavis.markup.visit_instruments(
instruments, "constructor", "*.~"
).items():
meavis_name_split = meavis_name.split(".")
intrument_cls = meavis._detail.inject.inject_namespace(
meavis_name_split[0], self_module
)
if not constructor_map:
constructor_map["class"] = meavis._detail.default.Constructor
logging.getLogger("meavis").debug(
"Add {} default constructor.".format(meavis_name_split[0])
)
if "module" not in constructor_map and meavis_name_split[0] in modules:
constructor_map["module"] = modules[meavis_name_split[0]]
logging.getLogger("meavis").info(
"Inject instrument [meavis.instruments].{}.".format(
intrument_cls._meavis_name
)
)
constructor_cls = meavis._detail.inject.inject_cls_by_dict(
constructor_map, intrument_cls, "constructor", meavis_name,
)
constructor_cls._meavis_map = {}
initialisers = {
meavis_name.split(".")[0]: initialiser_map
for meavis_name, initialiser_map in meavis.markup.visit_instruments(
instruments, "initialiser", "*.~"
).items()
}
for meavis_name, initialiser in initialisers.items():
if "module" not in initialiser and meavis_name in modules:
initialiser["module"] = modules[meavis_name]
all_usages = {}
for meavis_name, initialiser_map in meavis.markup.visit_instruments(
instruments, "initialiser", "*.*"
).items():
meavis_name_split = meavis_name.split(".")
usage_cls = meavis._detail.inject.inject_namespace(
".".join(meavis_name_split[:2]), self_module
)
meavis._detail.inject.inject_cls_by_dict(
meavis._detail.utils.merge_mapping(
initialiser_map, initialisers[meavis_name_split[0]]
),
usage_cls,
"initialiser",
meavis_name,
)
for tag_name in all_tags:
meavis._detail.inject.inject_namespace(tag_name, usage_cls)
if meavis_name_split[0] not in all_usages:
all_usages[meavis_name_split[0]] = set()
all_usages[meavis_name_split[0]].add(".".join(meavis_name_split[:2]))
root_items = {}
usage_items = {}
for tag_name in all_tags:
root_items[tag_name] = meavis.markup.visit_instruments(
instruments, tag_name, "*.~.*"
)
usage_items[tag_name] = meavis.markup.visit_instruments(
instruments, tag_name, "*.*.*"
)
for meavis_name, item_map in usage_items[tag_name].items():
meavis_name_split = meavis_name.split(".")
tag_cls = meavis._detail.inject.inject_namespace(
".".join(meavis_name_split[:2] + [tag_name]), self_module
)
root_name = ".".join(meavis_name_split[::2])
if root_name in root_items[tag_name]:
item_map = meavis._detail.utils.merge_mapping(
item_map, root_items[tag_name][root_name]
)
if "module" not in item_map and meavis_name_split[0] in modules:
item_map["module"] = modules[meavis_name_split[0]]
meavis._detail.inject.inject_cls_by_dict(
item_map, tag_cls, meavis_name_split[2], meavis_name_split[2]
)
for tag_name in all_tags:
for meavis_name, item_map in root_items[tag_name].items():
meavis_name_split = meavis_name.split(".")
for usage_name in all_usages[meavis_name_split[0]]:
tag_cls = meavis._detail.inject.inject_namespace(
"{}.{}".format(usage_name, tag_name), self_module
)
if (
"{}.{}".format(usage_name, meavis_name_split[1])
in usage_items[tag_name]
):
continue
if (
"module" not in item_map
and meavis_name_split[0] in modules
):
item_map["module"] = modules[meavis_name_split[0]]
meavis._detail.inject.inject_cls_by_dict(
item_map,
tag_cls,
meavis_name_split[1],
meavis_name_split[1],
)
[docs]def register(instances):
"""Inject instances."""
meavis._detail.debug.parameter_isinstance(
"instances", collections.abc.Mapping
)
for instance_name, instance in instances.items():
meavis._detail.debug.name_isnotinjected(
instance_name, meavis.parameters
)
meavis._detail.debug.name_isnotinjected(
instance_name, meavis.measurements
)
instrument_cls = meavis._detail.inject.inject_namespace(
instance["instrument"], self_module
)
constructor = (
instrument_cls.constructor(**(instance["kwargs"]))
if "kwargs" in instance
else instrument_cls.constructor()
)
constructor._meavis_hash = meavis._detail.debug.hash_setpair(
frozenset(
(key, "{}".format(value))
for key, value in constructor._meavis_kwargs.items()
)
)
if constructor._meavis_hash not in constructor._meavis_map:
constructor._meavis_map[constructor._meavis_hash] = constructor
constructor._meavis_handler = None
constructor._meavis_channels = set()
constructor._meavis_lock = threading.Lock()
logging.getLogger("meavis").info(
"Register {} constructor [{}] {{{}}}.".format(
instance["instrument"],
constructor._meavis_hash if __debug__ else None,
", ".join(
"{}: {}".format(key, value)
for key, value in constructor._meavis_kwargs.items()
),
)
)
usage_cls = meavis._detail.inject.inject_namespace(
instance["usage"], instrument_cls
)
initialiser = usage_cls.initialiser()
initialiser._meavis_constructor = constructor._meavis_map[
constructor._meavis_hash
]
initialiser._meavis_handler = None
initialiser._meavis_lock = initialiser._meavis_constructor._meavis_lock
logging.getLogger("meavis").info(
"Register {} initialiser {{{}}} for {}.".format(
instance["usage"],
", ".join(
"{}: {}".format(key, value)
for key, value in initialiser._meavis_kwargs.items()
),
instance_name,
)
)
all_tags = ["parameters", "measurements"]
all_instance_cls = [
meavis._detail.inject.inject_namespace(
instance_name, getattr(meavis, tag_name)
)
for tag_name in all_tags
]
if "attributes" not in instance:
instance["attributes"] = {}
for key, value in meavis.config.default_map["instances"].items():
if key in instance["attributes"]:
continue
instance["attributes"][key] = value
for key, value in instance["attributes"].items():
meavis_key = "_meavis_{}".format(key.lower().strip())
setattr(initialiser, meavis_key, value)
for instance_cls in all_instance_cls:
setattr(instance_cls, meavis_key, value)
logging.getLogger("meavis").debug(
"\tAdd instance attribute {} = {} to {}.".format(
meavis_key, value, instance_name
)
)
for tag_name in all_tags:
for item in [
value
for value in getattr(usage_cls, tag_name).__dict__.values()
if hasattr(value, "_meavis_injected")
and value._meavis_injected
]:
injected_item = getattr(meavis, tag_name).inject(
item, "{}.{}".format(instance_name, item._meavis_name),
)
injected_item._meavis_initialiser = initialiser
inject_default(injected_item, tag_name)
meavis.engines.completer.CompleterEngine.inject_instances(
(
meavis._detail.inject.inject_namespace(
instance_name, meavis.parameters
),
meavis._detail.inject.inject_namespace(
instance_name, meavis.measurements
),
)
for instance_name in instances
)
[docs]def inject_default(injected_item, tag_name):
"""Inject default MeaVis attributes in an item."""
for key, value in meavis.config.default_map[tag_name].items():
meavis_key = "_meavis_{}".format(key.lower().strip())
if not hasattr(injected_item, meavis_key):
setattr(injected_item, meavis_key, value)
logging.getLogger("meavis").debug(
"\tAdd default attribute {} = {}{}.".format(
meavis_key,
value,
" to {}".format(injected_item._meavis_name)
if hasattr(injected_item, "_meavis_name")
else "",
)
)