jmri_core.hal.internal

InternalConnection — in-memory loopback for tests and the default "internal" system.

JMRI ships a built-in connection with no hardware behind it: turnouts and sensors created against the internal prefix change state purely in memory. This is invaluable for tests, the JSON server's contract tests, and any "layout-less" scripting.

The internal prefix is "I" to match JMRI's convention.

 1"""``InternalConnection`` — in-memory loopback for tests and the default
 2"internal" system.
 3
 4JMRI ships a built-in connection with no hardware behind it: turnouts and
 5sensors created against the internal prefix change state purely in memory.
 6This is invaluable for tests, the JSON server's contract tests, and any
 7"layout-less" scripting.
 8
 9The internal prefix is ``"I"`` to match JMRI's convention.
10"""
11from __future__ import annotations
12
13from typing import ClassVar
14
15from jmri_core.hal.connection import ConnectionSettings, SystemConnection
16
17
18class InternalConnection(SystemConnection):
19    """Pure in-memory system connection. No I/O. No background tasks."""
20
21    name: ClassVar[str] = "Internal"
22    prefix: ClassVar[str] = "I"
23
24    async def open(self, settings: ConnectionSettings | None = None) -> None:
25        await super().open(settings)
26
27    async def close(self) -> None:
28        await super().close()
class InternalConnection(jmri_core.hal.connection.SystemConnection):
19class InternalConnection(SystemConnection):
20    """Pure in-memory system connection. No I/O. No background tasks."""
21
22    name: ClassVar[str] = "Internal"
23    prefix: ClassVar[str] = "I"
24
25    async def open(self, settings: ConnectionSettings | None = None) -> None:
26        await super().open(settings)
27
28    async def close(self) -> None:
29        await super().close()

Pure in-memory system connection. No I/O. No background tasks.

name: ClassVar[str] = 'Internal'
prefix: ClassVar[str] = 'I'
async def open(self, settings: dict[str, typing.Any] | None = None) -> None:
25    async def open(self, settings: ConnectionSettings | None = None) -> None:
26        await super().open(settings)

Connect to hardware and start any IO loops.

Concrete drivers acquire serial/TCP ports here, install protocol handlers, and start any required background tasks. The base implementation should be called via await super().open(settings) to mark the connection opened.

async def close(self) -> None:
28    async def close(self) -> None:
29        await super().close()

Disconnect from hardware. Idempotent. Base impl flips the flag.