Commit 3644401a authored by Benoit Formet's avatar Benoit Formet
Browse files

user_script: use namedtuple as namespace

parent 58ed0aba
Pipeline #29132 failed with stages
in 162 minutes and 11 seconds
......@@ -12,7 +12,7 @@ import collections
import functools
from treelib import Tree
from types import ModuleType, SimpleNamespace
from types import ModuleType
from bliss import setup_globals, global_map, is_bliss_shell
from bliss.config import static
......@@ -20,6 +20,7 @@ from bliss.config.settings import SimpleSetting
from bliss.config.conductor.client import get_text_file, get_python_modules, get_file
from bliss.common.proxy import Proxy
from bliss.common.logtools import log_warning
from bliss.common.utils import UserNamespace
from bliss.scanning import scan_saving
......@@ -574,12 +575,17 @@ class Session:
if k not in globals_dict:
env_dict[k] = globals_dict[k]
ns = SimpleNamespace(**env_dict)
ns = UserNamespace(env_dict)
if isinstance(export_global, str):
if isinstance(self.env_dict.get(export_global), SimpleNamespace):
if (
getattr(self.env_dict.get(export_global), "__module__", None)
== "bliss.common.utils.namespace"
# case: export and merge to existing namespace in env dict
d = self.env_dict[export_global]._asdict()
self.env_dict[export_global] = UserNamespace(d)
# case: export to given (non existing) namespace in env dict
safe_save_to_env_dict(self.env_dict, export_global, ns)
......@@ -16,6 +16,7 @@ import itertools
import functools
import numpy
import importlib.util
from import MutableMapping, MutableSequence
import socket
import fnmatch
......@@ -365,9 +366,9 @@ def with_custom_members(klass):
attr_info = custom_attrs.get(name)
if attr_info:
orig_type_info, access_mode = attr_info
if fget and not "r" in access_mode:
if fget and "r" not in access_mode:
access_mode = "rw"
if fset and not "w" in access_mode:
if fset and "w" not in access_mode:
access_mode = "rw"
assert type_info == orig_type_info, "%s get/set types mismatch" % name
......@@ -563,6 +564,36 @@ class autocomplete_property(property):
# create a copy of module collections to have a copy of namedtuple
__SPEC = importlib.util.find_spec("collections")
mycollections = importlib.util.module_from_spec(__SPEC)
sys.modules["mycollections"] = mycollections
from mycollections import namedtuple as UserNamedtuple # noqa E402
# patch property to trigger jedi signature hint
UserNamedtuple.__globals__["property"] = autocomplete_property
def UserNamespace(env_dict={}):
klass = UserNamedtuple("namespace", env_dict, module=__name__ + ".namespace")
def namespace_dir(self):
__dir__ = super(self.__class__, self).__dir__()
to_remove = []
if "count" not in env_dict:
if "index" not in env_dict:
return [i for i in __dir__ if i not in to_remove]
# patch dir function to hide "count" & "index" built-in tuples functions from jedi completion
klass.__dir__ = namespace_dir
ns = klass(**env_dict)
return ns
def deep_update(d, u):
"""Do a deep merge of one dict into another.
......@@ -145,13 +145,13 @@ def test_user_script(session4, capsys):
# test that user_script_load can return a namespace
expected_symbols = ["ascan", "time", "test1", "a"]
ns = user_script_load("sessions/subdir/scripts/simple_script", export_global=False)
assert list(ns.__dict__) == expected_symbols
assert list(ns._fields) == expected_symbols
session4.env_dict["user"] = 42
# test that user_script_load can export to "user" namespace
assert list(session4.env_dict["user"].__dict__) == expected_symbols
assert list(session4.env_dict["user"]._fields) == expected_symbols
# test backup of pre existing user
assert session4.env_dict["user_bak"] == 42
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment