sgr_commhandler.driver.messaging.messaging_interface_async

Attributes

logger

Classes

MessagingDataPoint

Implements a data point of a messaging interface.

MessagingFunctionalProfile

Implements a functional profile of a messaging interface.

SGrMessagingInterface

SmartGridready External Interface Class for Messaging Protocols

Functions

build_messaging_data_point(data_point, ...)

Builds a data point of a messaging interface.

get_messaging_client(name, desc)

Builds a messaging client from the interface specification.

Module Contents

sgr_commhandler.driver.messaging.messaging_interface_async.logger
sgr_commhandler.driver.messaging.messaging_interface_async.build_messaging_data_point(data_point, functional_profile, interface)

Builds a data point of a messaging interface.

Parameters:

  • data_point (sgr_specification.v0.product.MessagingDataPoint)

  • functional_profile (sgr_specification.v0.product.MessagingFunctionalProfile)

  • interface (SGrMessagingInterface)

Return type:

sgr_commhandler.api.data_point_api.DataPoint

sgr_commhandler.driver.messaging.messaging_interface_async.get_messaging_client(name, desc)

Builds a messaging client from the interface specification.

Parameters:

  • name (str)

  • desc (sgr_specification.v0.product.MessagingInterfaceDescription)

Return type:

sgr_commhandler.driver.messaging.messaging_client_async.SGrMessagingClient

class sgr_commhandler.driver.messaging.messaging_interface_async.MessagingDataPoint(dp_spec, fp_spec, interface)

Bases: sgr_commhandler.api.data_point_api.DataPointProtocol[sgr_specification.v0.product.MessagingFunctionalProfile, sgr_specification.v0.product.MessagingDataPoint]

Implements a data point of a messaging interface.

Parameters:

  • dp_spec (sgr_specification.v0.product.MessagingDataPoint)

  • fp_spec (sgr_specification.v0.product.MessagingFunctionalProfile)

  • interface (SGrMessagingInterface)

_in_cmd: sgr_specification.v0.product.InMessage | None = None
_read_cmd: sgr_specification.v0.product.OutMessage | None = None
_write_cmd: sgr_specification.v0.product.OutMessage | None = None
_in_filter: sgr_commhandler.driver.messaging.messaging_filter.MessagingFilter | None = None
_read_response_event: asyncio.Event | None = None
_cached_value: Any = None
_handle_message: collections.abc.Callable[[sgr_commhandler.api.data_point_api.DataPointProtocol, Any], NoReturn] | None = None
_interface
async get_val(parameters=None, skip_cache=False)
Parameters:

  • parameters (Optional[dict[str, str]])

  • skip_cache (bool)

Return type:

Any

async set_val(value)
Parameters:

value (Any)

can_subscribe()
Return type:

bool

subscribe(fn)
Parameters:

fn (collections.abc.Callable[[sgr_commhandler.api.data_point_api.DataPointProtocol, Any], NoReturn])

unsubscribe()
_on_message(payload)
Parameters:

payload (Any)

class sgr_commhandler.driver.messaging.messaging_interface_async.MessagingFunctionalProfile(fp_spec, interface)

Bases: sgr_commhandler.api.functional_profile_api.FunctionalProfile[sgr_specification.v0.product.MessagingFunctionalProfile]

Implements a functional profile of a messaging interface.

Parameters:
_interface
_data_points
get_data_points()
Return type:

dict[tuple[str, str], sgr_commhandler.api.data_point_api.DataPoint]

class sgr_commhandler.driver.messaging.messaging_interface_async.SGrMessagingInterface(frame)

Bases: sgr_commhandler.api.device_api.SGrBaseInterface

SmartGridready External Interface Class for Messaging Protocols

Parameters:

frame (sgr_specification.v0.product.DeviceFrame)

_client
_subscribed_topics: set[str]
_data_point_handlers: list[tuple[str, str, str, collections.abc.Callable[[Any], NoReturn], sgr_commhandler.driver.messaging.messaging_filter.MessagingFilter | None]] = []
functional_profiles
is_connected()
async disconnect_async()
async connect_async()
async read_message(out_topic, in_topic, payload)
Parameters:

  • out_topic (str)

  • in_topic (str)

  • payload (Any)

Return type:

Any

async write_message(topic, payload)
Parameters:

  • topic (str)

  • payload (Any)

data_point_handler(data_point)
Parameters:

data_point (tuple[str, str, str, collections.abc.Callable[[Any], NoReturn], Optional[sgr_commhandler.driver.messaging.messaging_filter.MessagingFilter]])

handle_client_message(topic, payload)
Parameters:

  • topic (str)

  • payload (Any)