pypck.connection

Connection classes for pypck.

class pypck.connection.PchkConnectionManager(host: str, port: int, username: str, password: str, settings: dict[str, Any] | None = None, connection_id: str = 'PCHK')

Connection to LCN-PCHK.

async async_close() None

Close the active connection.

async async_connect(timeout: float = 30) None

Establish a connection to PCHK at the given socket.

async async_process_input(inp: Input) None

Process an input command.

async cancel_requests() None

Cancel all TimeoutRetryHandlers.

dump_modules() dict[str, dict[str, dict[str, Any]]]

Dump all modules and information about them in a JSON serializable dict.

event_callback(event: LcnEvent) None

Handle events from PchkConnection.

fire_event(event: LcnEvent) None

Fire event.

get_address_conn(addr: LcnAddr, request_serials: bool = True) AbstractConnection

Create and/or return an AbstractConnection to the given module or group.

get_group_conn(addr: LcnAddr) GroupConnection

Create and return the GroupConnection for the given group.

get_module_conn(addr: LcnAddr, request_serials: bool = True) ModuleConnection

Create and/or return the given LCN module.

is_ready() bool

Retrieve the overall connection state.

async lcn_connection_status_changed(is_lcn_connected: bool) None

Set the current connection state to the LCN bus.

async on_auth(success: bool) None

Is called after successful authentication.

async on_license_error() None

Is called if a license error occurs during connection.

async on_successful_login() None

Is called after connection to LCN bus system is established.

async open_connection() None

Connect to PCHK server (no authentication or license error check).

physical_to_logical(addr: LcnAddr) LcnAddr

Convert the physical segment id of an address to the logical one.

async ping() None

Send pings.

async ping_received(count: int | None) None

Ping was received.

async process_message(message: str) None

Is called when a new text message is received from the PCHK server.

async read_data_loop() None

Processes incoming data.

register_for_events(callback: Callable[[LcnEvent], None]) Callable[[...], None]

Register a function for callback on LCN events.

Return a function to unregister the callback.

register_for_inputs(callback: Callable[[Input], None]) Callable[[...], None]

Register a function for callback on PCK message received.

Returns a function to unregister the callback.

async scan_modules(num_tries: int = 3, timeout: float = 3) None

Scan for modules on the bus.

This is a convenience coroutine which handles all the logic when scanning modules on the bus. Because of heavy bus traffic, not all modules might respond to a scan command immediately. The coroutine will make ‘num_tries’ attempts to send a scan command and waits ‘timeout’ after the last module response before proceeding to the next try.

async scan_segment_couplers(num_tries: int = 3, timeout: float = 1.5) None

Scan for segment couplers on the bus.

This is a convenience coroutine which handles all the logic when scanning segment couplers on the bus. Because of heavy bus traffic, not all segment couplers might respond to a scan command immediately. The coroutine will make ‘num_tries’ attempts to send a scan command and waits ‘timeout’ after the last segment coupler response before proceeding to the next try.

async send_command(pck: bytes | str, to_host: bool = False, **kwargs: Any) bool

Send a PCK command to the PCHK server.

set_local_seg_id(local_seg_id: int) None

Set the local segment id.

async wait_closed() None

Wait until connection to PCHK server is closed.

async write_data_loop() None

Processes queue and writes data.