model_connector

connector

class ConnectorStreamHandler(connector: 'ModelConnector', stream: 'Stream[Message]')

Bases: object

connector: ModelConnector

handle_message(msg: Message)

handle_message(msg: NewTimeMessage)

handle_message(msg: RemapMessage)

handle_message(msg: UpdateMessage)

handle_message(msg: UpdateSeriesMessage)

handle_message(msg: QuitMessage)

initialize()

stream: Stream[Message]

class ModelConnector(model: 'ModelAdapterBase', updates: 'UpdateDataClientBase[T]', init_data: 'InitDataHandler', serialization: 'InternalSerializationStrategy[T]', name: 't.Optional[str]' = None)

Bases: Generic[T]

close(message: QuitMessage)

data_mask: DataMask

init_data: InitDataHandler

initialize() → RegistrationMessage

model: ModelAdapterBase

name: str | None = None

new_time(message: NewTimeMessage)

remap(message: RemapMessage) → None

Process a REMAP command from the orchestrator. The adapter decides what middleware to install; the connector stores the rename dictionaries and rewrites its sub mask so subsequent GET requests ask for the variant keys. See issue #127.

remapper: RemapMiddleware | None = None

serialization: InternalSerializationStrategy[T]

update(update: UpdateMessage) → ResultMessage

update_series(update: UpdateSeriesMessage) → ResultMessage

updates: UpdateDataClientBase[T]

class RemapMiddleware(remap: 'RemapMessage', auto_remap: 'AutoRemap')

Bases: object

auto_remap: AutoRemap

remap: RemapMessage

rename_incoming_update(update: dict)

rename_outgoing_update(update: dict)

class UpdateDataClient(name: str, home_address: str, sockets: Sockets | None = None)

Bases: UpdateDataClientBase[bytes]

clear()

close()

counter: Iterator[str]

get(address: str, key: str, mask: dict | None) → bytes

home_address: str

put(data: bytes) → Tuple[str, str]

reset_counter()

init_data

class DirectoryInitDataClient(root: 'pathlib.Path')

Bases: InitDataClient

get(name: str) → Tuple[FileType | None, DatasetPath | None]

get_type_and_path(path) → Tuple[FileType, DatasetPath]

root: Path

class InitDataClient(*args, **kwargs)

Bases: InitDataHandler

ensure_ftype(name: str, ftype: FileType)

get(name: str) → Tuple[FileType | None, DatasetPath | None]

get_type_and_path(path) → Tuple[FileType, DatasetPath]

class ServicedInitDataClient(name: 'str', server: 'str')

Bases: InitDataClient

client: RequestClient

close()

get(name: str, mask: dict | None = None) → Tuple[FileType | None, DatasetPath | None]

name: str

server: str