Spaces:
Running on Zero
Running on Zero
File size: 3,202 Bytes
3f76564 | 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 | # =====================================================
# Apckeyl Framework
# Version 1.0
# control_plane.py
# =====================================================
"""
Apckeyl Control Plane.
Version 1.0
Центральный слой управления Apckeyl Framework.
На этом этапе Control Plane:
- принимает тип задачи;
- ищет подходящий Compute Module;
- возвращает информацию о модуле.
Сетевые запросы пока НЕ выполняются.
"""
from module_registry import registry
# =====================================================
# Control Plane
# =====================================================
class ControlPlane:
def __init__(self, module_registry=None):
if module_registry is None:
module_registry = registry
self.registry = module_registry
# =================================================
# Resolve Task
# =================================================
def resolve_task(self, task_type):
if not task_type:
raise ValueError(
"task_type is required"
)
modules = self.registry.find_by_type(
self._map_task_to_module_type(
task_type
)
)
if not modules:
raise LookupError(
f"No Compute Module available "
f"for task type: {task_type}"
)
# -------------------------------------------------
# First available module
# -------------------------------------------------
for module in modules:
if module["status"] == "available":
return module
raise RuntimeError(
f"No available Compute Module "
f"for task type: {task_type}"
)
# =================================================
# Task → Module Type
# =================================================
def _map_task_to_module_type(
self,
task_type
):
task_mapping = {
"image_upscale":
"image_upscaler",
}
module_type = task_mapping.get(
task_type
)
if module_type is None:
raise LookupError(
f"Unknown task type: {task_type}"
)
return module_type
# =================================================
# Get Module
# =================================================
def get_module(
self,
module_id
):
module = self.registry.get(
module_id
)
if module is None:
raise LookupError(
f"Unknown module: {module_id}"
)
return module
# =================================================
# List Modules
# =================================================
def list_modules(self):
return self.registry.get_all()
# =====================================================
# Default Control Plane
# =====================================================
control_plane = ControlPlane() |