| import time |
| from typing import Dict, Any, Optional |
| from dataclasses import dataclass, asdict |
|
|
| class VRAMManager: |
| """Utility to manage GPU memory during model switching.""" |
| @staticmethod |
| def purge(): |
| import torch |
| import gc |
| if torch.cuda.is_available(): |
| torch.cuda.empty_cache() |
| gc.collect() |
|
|
| @dataclass |
| class NerveOutput: |
| """Standardized output from a Phase 2 Sensor Model (Nerve).""" |
| sensor_type: str |
| timestamp: float |
| caption: str |
| metadata: Dict[str, Any] |
| raw_data: Optional[Any] = None |
|
|
| class SensorNerve: |
| """Base class for all Phase 2 Nerve models.""" |
| def __init__(self, sensor_type: str): |
| self.sensor_type = sensor_type |
|
|
| def process(self, raw_data: Any) -> NerveOutput: |
| raise NotImplementedError |
|
|
| class VisionNerve(SensorNerve): |
| """Camera Nerve: Converts video/images to basic visual perception.""" |
| def __init__(self): |
| super().__init__("camera") |
|
|
| def process(self, caption: str, raw_data: Any = None, objects: list = None) -> NerveOutput: |
| return NerveOutput( |
| sensor_type=self.sensor_type, |
| timestamp=time.time(), |
| caption=caption, |
| raw_data=raw_data, |
| metadata={ |
| "base_objects": objects or [], |
| "scene_context": "general" |
| } |
| ) |
|
|
| class AudioNerve(SensorNerve): |
| """Microphone Nerve: Converts audio to basic acoustic perception.""" |
| def __init__(self): |
| super().__init__("microphone") |
|
|
| def process(self, caption: str, raw_data: Any = None) -> NerveOutput: |
| return NerveOutput( |
| sensor_type=self.sensor_type, |
| timestamp=time.time(), |
| caption=caption, |
| raw_data=raw_data, |
| metadata={ |
| "acoustic_summary": caption, |
| "domain": "ambient" |
| } |
| ) |
|
|
| |
| class RadarNerve(SensorNerve): |
| def __init__(self): super().__init__("radar") |
| def process(self, raw_signal: Any) -> NerveOutput: |
| return NerveOutput(self.sensor_type, time.time(), "Radar scan active", {"anomalies": []}) |
|
|
| class WiFiNerve(SensorNerve): |
| def __init__(self): super().__init__("wifi") |
| def process(self, raw_signal: Any) -> NerveOutput: |
| return NerveOutput(self.sensor_type, time.time(), "WiFi signal stable", {"snr": 0}) |
|
|
