engine_registry.py
4.89 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
"""Static runtime metadata registry for managed engines."""
from __future__ import annotations
from typing import Any, Iterable
EngineEntry = dict[str, Any]
EngineCatalog = dict[str, EngineEntry]
def build_default_engine_catalog() -> EngineCatalog:
"""Build the default engine metadata catalog for the web API runtime."""
return {
"insight": {
"name": "insight",
"available": True,
"launch_mode": "streamlit",
"script_path": "apps/engine_console/insight_engine_streamlit_app.py",
"local_runner": "insight",
},
"media": {
"name": "media",
"available": True,
"launch_mode": "streamlit",
"script_path": "apps/engine_console/media_engine_streamlit_app.py",
"local_runner": "media",
},
"query": {
"name": "query",
"available": True,
"launch_mode": "streamlit",
"script_path": "apps/engine_console/query_engine_streamlit_app.py",
"local_runner": "query",
},
"forum": {
"name": "forum",
"available": True,
"launch_mode": "callable",
"script_path": None,
"local_runner": None,
},
}
def _clone_entry(entry: EngineEntry) -> EngineEntry:
cloned: EngineEntry = {}
for key, value in entry.items():
if isinstance(value, list):
cloned[key] = list(value)
elif isinstance(value, dict):
cloned[key] = dict(value)
else:
cloned[key] = value
return cloned
class EngineRuntimeRegistry:
"""Explicit wrapper around the static engine runtime metadata catalog."""
def __init__(self, catalog: EngineCatalog | None = None) -> None:
default_catalog = catalog if catalog is not None else build_default_engine_catalog()
self._default_catalog = {name: _clone_entry(entry) for name, entry in default_catalog.items()}
self._catalog = {name: _clone_entry(entry) for name, entry in self._default_catalog.items()}
@property
def catalog(self) -> EngineCatalog:
return self._catalog
def items(self):
return self._catalog.items()
def contains(self, engine_name: str) -> bool:
return engine_name in self._catalog
def get_entry(self, engine_name: str) -> EngineEntry:
return self._catalog[engine_name]
def get_local_runner(self, engine_name: str) -> str | None:
runner = self.get_entry(engine_name).get("local_runner")
return str(runner) if runner else None
def get_script_path(self, engine_name: str) -> str | None:
script_path = self.get_entry(engine_name).get("script_path")
return str(script_path) if script_path else None
def is_available(self, engine_name: str) -> bool:
return bool(self.get_entry(engine_name).get("available", False))
def set_available(self, engine_name: str, available: bool) -> None:
self.get_entry(engine_name)["available"] = bool(available)
def reset(self) -> None:
self._catalog = {
name: _clone_entry(entry)
for name, entry in self._default_catalog.items()
}
def snapshot(self, engine_name: str) -> EngineEntry:
return _clone_entry(self.get_entry(engine_name))
def snapshot_all(self) -> EngineCatalog:
return {name: self.snapshot(name) for name in self._catalog}
def names(self) -> list[str]:
return list(self._catalog.keys())
def engines_for_launch_mode(self, launch_mode: str) -> list[str]:
return [
engine_name
for engine_name, entry in self._catalog.items()
if entry.get("launch_mode") == launch_mode
]
def streamlit_script_snapshot(self) -> dict[str, str]:
return {
engine_name: str(entry["script_path"])
for engine_name, entry in self._catalog.items()
if entry.get("launch_mode") == "streamlit" and entry.get("script_path")
}
def streamlit_scripts(self) -> dict[str, str]:
return self.streamlit_script_snapshot()
def local_runner_snapshot(self, engine_names: Iterable[str] | None = None) -> dict[str, str]:
names = self.names() if engine_names is None else [str(name) for name in engine_names]
runners: dict[str, str] = {}
for engine_name in names:
if not self.contains(engine_name):
continue
runner = self.get_local_runner(engine_name)
if runner:
runners[engine_name] = runner
return runners
def local_runners(self, engine_names: Iterable[str] | None = None) -> dict[str, str]:
return self.local_runner_snapshot(engine_names)
ENGINE_RUNTIME_REGISTRY = EngineRuntimeRegistry()
__all__ = [
"ENGINE_RUNTIME_REGISTRY",
"EngineCatalog",
"EngineEntry",
"EngineRuntimeRegistry",
"build_default_engine_catalog",
]