Spaces:
Running
Running
File size: 7,163 Bytes
31226fd | 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 | """Configuration-driven composition for the controlled experience loop."""
from __future__ import annotations
from collections.abc import Mapping
from dataclasses import dataclass
from pathlib import Path
from typing import Any, cast
from src.embodiment import (
ActionCommand,
ActuatorResult,
ConnectionDescriptor,
ConnectionKind,
ConnectionStatus,
ControlledEmbodimentAgent,
DeterministicTargetEnvironment,
RelationshipClass,
SystemSensorAdapter,
host_system_readings,
)
from src.experience.engine import ExperienceEngine
from src.learning.learning_engine import LearningEngine
from src.memory import MemoryStore, MemoryWorldModel, TransitionWorldModel
from src.profiles import BehaviorProfile
@dataclass(slots=True)
class DeterministicActuator:
"""Actuator boundary for the fully controlled digital environment."""
actuator_id: str = "target-actuator"
active: bool = True
def apply(self, command: ActionCommand) -> ActuatorResult:
return ActuatorResult(True, command.action)
def _numeric(value: Any) -> float:
return float(value) if isinstance(value, (int, float)) else 0.0
def _system_encoder(frame: Any) -> Mapping[int, float]:
payload = frame.payload
if not isinstance(payload, dict):
return {}
values = (
_numeric(payload.get("cpu_percent")),
_numeric(payload.get("memory_percent")),
_numeric(payload.get("temperature_c")),
1.0 if payload.get("network_up") is True else 0.0,
)
return {index: value / 100.0 for index, value in enumerate(values)}
def _controlled_decoder(result: Any, frame: Any) -> ActionCommand | None:
spikes = getattr(result, "output_spike_ids", ())
if not spikes:
return None
return ActionCommand("target-actuator", frame.tick, "right")
def build_experience_subsystem(
config: Mapping[str, Any],
network: Any,
learning: LearningEngine | None,
) -> ExperienceEngine | None:
"""Build the configured experience subsystem, or return ``None`` disabled."""
raw = config.get("experience", {})
if not isinstance(raw, Mapping):
raise TypeError("experience config must be a mapping")
if not bool(raw.get("enabled", False)):
return None
sensor_config = raw.get("sensor", {})
encoder_config = raw.get("encoder", {})
decoder_config = raw.get("decoder", {})
environment_config = raw.get("environment", {})
for name, value in (
("experience.sensor", sensor_config),
("experience.encoder", encoder_config),
("experience.decoder", decoder_config),
("experience.environment", environment_config),
):
if not isinstance(value, Mapping):
raise TypeError(f"{name} config must be a mapping")
if sensor_config.get("type", "system") != "system":
raise ValueError("unsupported experience sensor type")
provider_name = sensor_config.get("provider", "host")
if provider_name == "host":
provider = host_system_readings
elif provider_name == "deterministic_trace":
trace = sensor_config.get("trace")
if not isinstance(trace, list):
raise ValueError("deterministic_trace requires a list trace")
def provider(tick: int) -> Mapping[str, Any]:
if tick >= len(trace) or not isinstance(trace[tick], Mapping):
raise ValueError("deterministic_trace has no mapping for this tick")
return cast(Mapping[str, Any], trace[tick])
else:
raise ValueError("unknown experience sensor provider")
if encoder_config.get("type", "system_v1") != "system_v1":
raise ValueError("unknown experience encoder type")
if decoder_config.get("type", "controlled_v1") != "controlled_v1":
raise ValueError("unknown experience decoder type")
if environment_config.get("type", "deterministic_target") != "deterministic_target":
raise ValueError("unknown experience environment type")
if learning is None:
raise RuntimeError("enabled experience requires the learning engine")
descriptor = ConnectionDescriptor(
connection_id="target-actuator",
name="Deterministic target actuator",
kind=ConnectionKind.ACTUATOR,
relationship=RelationshipClass.CONTROLLABLE,
status=ConnectionStatus.CONNECTED,
capabilities=("right",),
available=True,
authorized=True,
active=True,
)
embodiment = ControlledEmbodimentAgent(
environment=DeterministicTargetEnvironment(),
actuator=DeterministicActuator(),
descriptor=descriptor,
)
embodiment.reset(seed=int(config.get("seed", 42)))
memory = None
memory_config = raw.get("memory", {})
if isinstance(memory_config, Mapping) and bool(memory_config.get("enabled", False)):
persistence_value = memory_config.get("persistence_path")
persistence_path = (
None if persistence_value is None else Path(str(persistence_value))
)
store = MemoryStore(
run_id=str(memory_config.get("run_id", "experience-run")),
root=persistence_path,
episode_capacity=int(memory_config.get("episode_capacity", 128)),
working_capacity=int(memory_config.get("working_capacity", 16)),
prediction_capacity=int(memory_config.get("prediction_capacity", 128)),
retention_ticks=int(memory_config.get("retention_ticks", 1024)),
read_enabled=bool(memory_config.get("read_enabled", True)),
write_enabled=bool(memory_config.get("write_enabled", True)),
)
memory = MemoryWorldModel(
store,
TransitionWorldModel(
max_contexts=int(memory_config.get("max_contexts", 128))
),
store.run_id,
persistence_path=persistence_path,
prediction_enabled=memory_config.get("prediction_enabled", True),
learning_enabled=memory_config.get("learning_enabled", True),
)
behavior_profile = None
behavior_value = raw.get("behavior", config.get("behavior", {}))
if not isinstance(behavior_value, Mapping):
raise TypeError("experience.behavior config must be a mapping")
if bool(behavior_value.get("enabled", False)):
initial_value = behavior_value.get("initial", {})
if not isinstance(initial_value, Mapping):
raise TypeError("experience.behavior.initial config must be a mapping")
behavior_profile = BehaviorProfile(
profile_id=str(behavior_value.get("profile_id", "WESEN-0001")),
initial={str(key): float(value) for key, value in initial_value.items()},
update_rate=float(behavior_value.get("update_rate", 0.05)),
)
return ExperienceEngine(
sensor=SystemSensorAdapter(provider),
network=network,
encoder=_system_encoder,
decoder=_controlled_decoder,
embodiment=embodiment,
learning=learning,
memory=memory,
behavior_profile=behavior_profile,
)
__all__ = ["DeterministicActuator", "build_experience_subsystem"]
|