Skip to content

Sim2Real API

bridge

DockerBridge

Bridge connecting actions to the hypervisor backend.

Source code in netforge_rl\docker_bridge\bridge.py 
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
class DockerBridge:
    """Bridge connecting actions to the hypervisor backend."""

    def __init__(self, mode: Literal['sim', 'real'] = 'sim') -> None:
        self.mode = mode
        self._driver: BaseHypervisor = self._init_driver(mode)

    def dispatch(
        self,
        action_name: str,
        target_ip: str,
        target_os: str,
    ) -> HypervisorResult:
        """Execute payload; auto-fallback to mock if real driver is down."""
        result = self._driver.dispatch(action_name, target_ip, target_os)
        logger.debug('DockerBridge: %s', result)
        return result

    def reward_delta(self, result: HypervisorResult) -> float:
        """Map a HypervisorResult to an immediate scalar reward delta."""
        if result.success:
            return _REWARD_DELTA['success']
        elif result.return_code == 2:
            # Container/infrastructure error
            return _REWARD_DELTA['failure_error']
        elif result.latency_ms >= _NOISY_LATENCY_THRESHOLD_MS:
            return _REWARD_DELTA['failure_noisy']
        else:
            return _REWARD_DELTA['failure_clean']

    def teardown_all(self) -> None:
        """Destroy all active containers/sessions — call at episode end."""
        self._driver.teardown_all()

    def is_available(self) -> bool:
        return self._driver.is_available()

    def _init_driver(self, mode: str) -> BaseHypervisor:
        if mode == 'real':
            driver = DockerHypervisor()
            if not driver.is_available():
                logger.warning(
                    'DockerBridge: real mode requested but Docker unavailable. '
                    'Falling back to mock hypervisor.'
                )
                return MockHypervisor()
            return driver

        # Default: sim / mock
        return MockHypervisor()

dispatch 

dispatch(
    action_name: str, target_ip: str, target_os: str
) -> HypervisorResult

Execute payload; auto-fallback to mock if real driver is down.

Source code in netforge_rl\docker_bridge\bridge.py 
29
30
31
32
33
34
35
36
37
38
def dispatch(
    self,
    action_name: str,
    target_ip: str,
    target_os: str,
) -> HypervisorResult:
    """Execute payload; auto-fallback to mock if real driver is down."""
    result = self._driver.dispatch(action_name, target_ip, target_os)
    logger.debug('DockerBridge: %s', result)
    return result

reward_delta 

reward_delta(result: HypervisorResult) -> float

Map a HypervisorResult to an immediate scalar reward delta.

Source code in netforge_rl\docker_bridge\bridge.py 
40
41
42
43
44
45
46
47
48
49
50
def reward_delta(self, result: HypervisorResult) -> float:
    """Map a HypervisorResult to an immediate scalar reward delta."""
    if result.success:
        return _REWARD_DELTA['success']
    elif result.return_code == 2:
        # Container/infrastructure error
        return _REWARD_DELTA['failure_error']
    elif result.latency_ms >= _NOISY_LATENCY_THRESHOLD_MS:
        return _REWARD_DELTA['failure_noisy']
    else:
        return _REWARD_DELTA['failure_clean']

teardown_all 

teardown_all() -> None

Destroy all active containers/sessions — call at episode end.

Source code in netforge_rl\docker_bridge\bridge.py 
52
53
54
def teardown_all(self) -> None:
    """Destroy all active containers/sessions — call at episode end."""
    self._driver.teardown_all()