Skip to content

vsenv

Functions:

Attributes:

PRELOADED_MODULES module-attribute 

PRELOADED_MODULES = set(values())

T module-attribute 

T = TypeVar('T')

current_log module-attribute 

current_log: LogHandle | None = None

environment module-attribute 

environment: ManagedEnvironment | None = None

orig_runpy_run_code module-attribute 

orig_runpy_run_code = _run_code

policy module-attribute 

policy: FlagsPolicy = FlagsPolicy(GlobalStore())

FlagsPolicy

Bases: Policy

Methods:

new_environment 

new_environment() -> ManagedEnvironment
Source code 
36
37
38
39
40
def new_environment(self) -> ManagedEnvironment:
    data = self.api.create_environment(CoreCreationFlags.ENABLE_GRAPH_INSPECTION)
    env = self.api.wrap_environment(data)
    logger.debug("Created new environment")
    return ManagedEnvironment(env, data, self)

PyQTLoop

Bases: QObject, EventLoop

Methods:

Attributes:

move class-attribute instance-attribute 

move = pyqtSignal(int)

attach 

attach() -> None
Source code 
69
70
71
72
73
74
75
def attach(self) -> None:
    self._signal_lock = Lock()
    self._signal_counter = 0
    self._signallers = dict[int, Callable[[], None]]()

    self._slot = self._receive_task
    self.move.connect(self._slot)

detach 

detach() -> None
Source code 
77
78
79
def detach(self) -> None:
    self.move.disconnect(self._slot)
    self._signallers = {}

from_thread 

from_thread(func: Callable[..., T], *args: Any, **kwargs: Any) -> Future[T]
Source code 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def from_thread(self, func: Callable[..., T], *args: Any, **kwargs: Any) -> Future[T]:
    fut = Future[T]()

    with self._signal_lock:
        my_counter = self._signal_counter

        self._signal_counter += 1

    def wrapper() -> None:
        nonlocal my_counter

        if not fut.set_running_or_notify_cancel():
            return

        del self._signallers[my_counter]

        try:
            result = func(*args, **kwargs)
        except BaseException as e:
            fut.set_exception(e)
        else:
            fut.set_result(result)

    self._signallers[my_counter] = wrapper
    self.move.emit(my_counter)

    return fut

to_thread 

to_thread(func: Callable[..., T], *args: Any, **kwargs: Any) -> Future[T]
Source code 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
def to_thread(self, func: Callable[..., T], *args: Any, **kwargs: Any) -> Future[T]:
    fut = Future[T]()

    def wrapper() -> None:
        if not fut.set_running_or_notify_cancel():
            return

        try:
            result = func(*args, **kwargs)
        except BaseException as e:
            fut.set_exception(e)
        else:
            fut.set_result(result)

    QThreadPool.globalInstance().start(Runner(wrapper))

    return fut

Runner 

Runner(wrapper: Callable[[], None])

Bases: QRunnable

Methods:

Attributes:

Source code 
58
59
60
def __init__(self, wrapper: Callable[[], None]) -> None:
    super().__init__()
    self.wrapper = wrapper

wrapper instance-attribute 

wrapper = wrapper

run 

run() -> None
Source code 
62
63
def run(self) -> None:
    self.wrapper()

dispose_environment 

dispose_environment(env: ManagedEnvironment) -> None
Source code 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
def dispose_environment(env: ManagedEnvironment) -> None:
    import logging

    from vstools.utils.vs_proxy import _finalize_core

    if current_log:
        env.core.remove_log_handler(current_log)

    _finalize_core(env._environment.env_id, id(env.core), True)

    if logging.getLogger().level <= logging.DEBUG:
        from gc import get_referrers

        # There has to be the environment referencing it
        ref_count = len(get_referrers(env.core)) - 1

        if ref_count:
            logging.debug(f'Core {id(env.core)} is being held reference by {ref_count} objects!')

    env.dispose()

get_current_environment 

get_current_environment() -> ManagedEnvironment
Source code 
180
181
182
def get_current_environment() -> ManagedEnvironment:
    assert environment
    return environment

make_environment 

make_environment() -> None
Source code 
138
139
140
141
142
143
144
145
146
147
def make_environment() -> None:
    global environment, current_log
    assert policy is not None

    if environment and current_log:
        environment.core.remove_log_handler(current_log)

    environment = policy.new_environment()
    current_log = environment.core.add_log_handler(get_vs_logger())
    environment.switch()

set_vsengine_loop 

set_vsengine_loop() -> None
Source code 
131
132
def set_vsengine_loop() -> None:
    set_loop(PyQTLoop())