Repository navigation
Expand file tree
/
Copy pathruntime.py
More file actions
373 lines (308 loc) · 12.8 KB
/
Copy pathruntime.py
File metadata and controls
373 lines (308 loc) · 12.8 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
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
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
"""Shared per-session CloudScope runtime for multi-page NiceGUI clients."""
from __future__ import annotations
import threading
from collections.abc import Callable
from dataclasses import dataclass
from pathlib import Path
from cloudscope.app_config import AppConfig
from cloudscope.controllers.analysis_controller import AnalysisController
from cloudscope.controllers.event_analysis_controller import EventAnalysisController
from cloudscope.controllers.home_page_controller import HomePageController, HomePageState
from cloudscope.controllers.acq_image_data_controller import AcqImageDataController
from cloudscope.controllers.load_save_controller import LoadSaveController
from cloudscope.controllers.roi_controller import RoiController
from cloudscope.controllers.sum_intensity_pool_controller import SumIntensityPoolController
from cloudscope.controllers.velocity_pool_controller import VelocityPoolController
from cloudscope.event_bus import EventBus
from cloudscope.events.files import LoadPathIntent, LoadPathKind
from cloudscope.raster_display_cache import RasterDisplayCache, resolve_raster_display_cache_max_entries
from cloudscope.session_state import HomePageSessionSnapshot
from cloudscope.task_runner import TaskRunner
from cloudscope.user_context import (
UserContext,
get_or_create_demo_session_id,
resolve_user_context_from_env,
)
def infer_load_kind(path: str) -> LoadPathKind:
"""Infer load kind from a filesystem path string.
Args:
path: Path to a file, folder, or CSV list.
Returns:
Matching ``LoadPathKind`` for ``LoadPathIntent``.
"""
if path.lower().endswith('.csv'):
return LoadPathKind.CSV
return LoadPathKind.FOLDER if Path(path).expanduser().is_dir() else LoadPathKind.FILE
@dataclass(slots=True)
class CloudScopeRuntime:
"""Shared application runtime for one user or browser session.
Owns canonical in-memory scientific state, controllers, and the event bus.
NiceGUI pages create UI views only and read state from this object.
Args:
user_context: Resolved workspace/user context.
app_config: Application configuration for this context.
event_bus: Shared typed event bus.
home_page_controller: Controller owning ``HomePageState``.
load_save_controller: Load and save workflow controller.
analysis_controller: Analysis workflow controller.
roi_controller: ROI mutation controller.
event_analysis_controller: Event-analysis controller.
velocity_pool_controller: Velocity pool synchronization controller.
sum_intensity_pool_controller: Sum-intensity pool synchronization controller.
acq_image_data_controller: Explicit lazy AcqImage data load/unload controller.
task_runner: Background task runner for long-running work.
raster_display_cache: Shared LRU cache of raster planes and pyramids.
initialized: Whether one-time bootstrap has completed.
controllers_bound: Whether controller ``bind()`` has run.
session_snapshot: Last captured reconnect session snapshot, if any.
reconnect_build_in_progress: True while ``HomePage.build(reconnect=True)``
is running before the reconnect hydrate event is published.
"""
user_context: UserContext
app_config: AppConfig
event_bus: EventBus
home_page_controller: HomePageController
load_save_controller: LoadSaveController
analysis_controller: AnalysisController
roi_controller: RoiController
event_analysis_controller: EventAnalysisController
velocity_pool_controller: VelocityPoolController
sum_intensity_pool_controller: SumIntensityPoolController
acq_image_data_controller: AcqImageDataController
task_runner: TaskRunner
raster_display_cache: RasterDisplayCache
initialized: bool = False
controllers_bound: bool = False
# Latest in-memory home-page snapshot for rebuilding this runtime. It is
# scoped to the surviving Python process and is not persistent across a
# server restart. If multiple clients share a runtime, the most recently
# captured snapshot replaces the previous snapshot.
session_snapshot: HomePageSessionSnapshot | None = None
reconnect_build_in_progress: bool = False
@property
def app_state(self) -> HomePageState:
"""Return shared home-page application state.
Returns:
Current ``HomePageState`` owned by ``home_page_controller``.
"""
return self.home_page_controller.state
def ensure_controllers_bound(self) -> None:
"""Bind runtime controllers to the shared event bus once.
Returns:
None.
"""
if self.controllers_bound:
return
self.home_page_controller.bind()
self.load_save_controller.bind()
self.analysis_controller.bind()
self.roi_controller.bind()
self.event_analysis_controller.bind()
self.velocity_pool_controller.bind()
self.sum_intensity_pool_controller.bind()
self.controllers_bound = True
def initialize_once(self) -> None:
"""Run one-time bootstrap for this runtime instance.
Idempotent: does not reload ``AcqImageList`` or repeat bootstrap when
the runtime already holds loaded state or has been initialized.
Returns:
None.
"""
self.ensure_controllers_bound()
if self.initialized:
return
if self.home_page_controller.state.acq_image_list is not None:
self.initialized = True
return
self.home_page_controller.load_demo_files([])
last_path = self.app_config.get_last_path().strip()
if last_path:
self.event_bus.publish(
LoadPathIntent(path=last_path, kind=infer_load_kind(last_path))
)
self.initialized = True
class CloudScopeRuntimeRegistry:
"""Thread-safe registry of per-session ``CloudScopeRuntime`` instances."""
def __init__(self) -> None:
"""Initialize an empty runtime registry."""
self._lock = threading.RLock()
self._runtimes: dict[str, CloudScopeRuntime] = {}
def get_or_create(self, key: str, factory: Callable[[], CloudScopeRuntime]) -> CloudScopeRuntime:
"""Return an existing runtime or create one with ``factory``.
Args:
key: Stable per-user or per-session runtime key.
factory: Callable returning a new ``CloudScopeRuntime``.
Returns:
Shared runtime for ``key``.
"""
with self._lock:
runtime = self._runtimes.get(key)
if runtime is None:
runtime = factory()
self._runtimes[key] = runtime
return runtime
def get(self, key: str) -> CloudScopeRuntime | None:
"""Return a runtime for ``key`` when present.
Args:
key: Stable per-user or per-session runtime key.
Returns:
Registered runtime, or ``None`` when absent.
"""
with self._lock:
return self._runtimes.get(key)
def clear(self) -> None:
"""Remove all registered runtimes.
Returns:
None.
"""
with self._lock:
self._runtimes.clear()
_registry: CloudScopeRuntimeRegistry | None = None
_process_app_config: AppConfig | None = None
_process_user_context: UserContext | None = None
def set_process_app_config(
app_config: AppConfig,
*,
user_context: UserContext | None = None,
) -> None:
"""Register the single desktop ``AppConfig`` for Option C launcher mode.
Option C loads configuration in the launcher thread before the NiceGUI
server starts. Page handlers must reuse that same in-memory instance so
geometry and preference saves are not split across duplicate objects.
Args:
app_config: Shared application configuration for this process.
user_context: Optional resolved user context for the same process.
Returns:
None.
"""
global _process_app_config, _process_user_context
_process_app_config = app_config
if user_context is not None:
_process_user_context = user_context
def clear_process_app_config() -> None:
"""Clear process-wide desktop config registration.
Returns:
None.
"""
global _process_app_config, _process_user_context
_process_app_config = None
_process_user_context = None
def get_process_app_config() -> AppConfig | None:
"""Return the registered desktop ``AppConfig`` when present.
Returns:
Process-wide app config, or ``None`` outside Option C desktop mode.
"""
return _process_app_config
def get_registry() -> CloudScopeRuntimeRegistry:
"""Return the process-wide runtime registry.
Returns:
Shared ``CloudScopeRuntimeRegistry`` instance.
"""
global _registry
if _registry is None:
_registry = CloudScopeRuntimeRegistry()
return _registry
def reset_runtime_registry_for_tests() -> None:
"""Clear the process-wide runtime registry.
For unit tests only.
Returns:
None.
"""
global _registry
if _registry is not None:
_registry.clear()
_registry = None
clear_process_app_config()
def runtime_key_from_user_context(user_context: UserContext) -> str:
"""Return the registry key for a resolved user context.
Args:
user_context: Resolved CloudScope user context.
Returns:
Stable runtime key, currently ``user_context.user_id``.
"""
return user_context.user_id
def resolve_runtime_context() -> tuple[UserContext, AppConfig]:
"""Resolve user context and app config for the current NiceGUI client.
Returns:
Tuple of ``(UserContext, AppConfig)`` for the active session.
"""
if _process_app_config is not None:
if _process_user_context is not None:
return _process_user_context, _process_app_config
demo_session_id = get_or_create_demo_session_id()
user_context = resolve_user_context_from_env(demo_session_id=demo_session_id)
return user_context, _process_app_config
demo_session_id = get_or_create_demo_session_id()
user_context = resolve_user_context_from_env(demo_session_id=demo_session_id)
app_config = user_context.load_app_config(create_if_missing=False)
return user_context, app_config
def _build_runtime(user_context: UserContext, app_config: AppConfig) -> CloudScopeRuntime:
"""Construct a new runtime for one user/session.
Args:
user_context: Resolved user context.
app_config: Application configuration.
Returns:
New, uninitialized ``CloudScopeRuntime``.
"""
event_bus = EventBus()
raster_display_cache = RasterDisplayCache(
max_entries=resolve_raster_display_cache_max_entries(),
)
acq_image_data_controller = AcqImageDataController(
event_bus=event_bus,
raster_display_cache=raster_display_cache,
)
home_page_controller = HomePageController(
event_bus=event_bus,
acq_image_data_controller=acq_image_data_controller,
)
task_runner = TaskRunner(event_bus=event_bus)
load_save_controller = LoadSaveController(
event_bus=event_bus,
home_controller=home_page_controller,
app_config=app_config,
user_context=user_context,
task_runner=task_runner,
)
return CloudScopeRuntime(
user_context=user_context,
app_config=app_config,
event_bus=event_bus,
home_page_controller=home_page_controller,
load_save_controller=load_save_controller,
analysis_controller=AnalysisController(
event_bus=event_bus,
home_controller=home_page_controller,
task_runner=task_runner,
),
roi_controller=RoiController(
event_bus=event_bus,
home_page_controller=home_page_controller,
),
event_analysis_controller=EventAnalysisController(
event_bus=event_bus,
home_controller=home_page_controller,
),
velocity_pool_controller=VelocityPoolController(
event_bus=event_bus,
home_controller=home_page_controller,
),
sum_intensity_pool_controller=SumIntensityPoolController(
event_bus=event_bus,
home_controller=home_page_controller,
),
acq_image_data_controller=acq_image_data_controller,
task_runner=task_runner,
raster_display_cache=raster_display_cache,
)
def get_current_runtime() -> CloudScopeRuntime:
"""Return the shared runtime for the current user/session.
Returns:
Shared ``CloudScopeRuntime`` for the active NiceGUI client context.
"""
user_context, app_config = resolve_runtime_context()
key = runtime_key_from_user_context(user_context)
return get_registry().get_or_create(
key,
lambda: _build_runtime(user_context, app_config),
)