sgr_commhandler.api

Contains the CommHandler API, which is intended to be utilized by the user.

Submodules

Classes

`ConfigurationParameter`	Implements an EID configuration parameter.
`DynamicParameter`	Implements a dynamic parameter of data points.
`DataPoint`	Implements a data point of a generic data type.
`DataPointProtocol`	Implements a base class for data point protocols.
`DataPointValidator`	Implements a base class for data point validators.
`DeviceInformation`	Implements a device information container.
`SGrBaseInterface`	Defines an abstract base class for all SGr device interfaces.
`FunctionalProfile`	Implements a functional profile.

Package Contents

class sgr_commhandler.api.ConfigurationParameter(parameter)

Implements an EID configuration parameter.

Parameters:: parameter (sgr_specification.v0.product.ConfigurationListElement)

label

name

type

description

default_value

__str__()

Converts to string.

Returns:: A string
Return type:: str

__repr__()

Gets an object description as string.

Returns:: A string
Return type:: str

class sgr_commhandler.api.DynamicParameter(parameter)

Implements a dynamic parameter of data points.

Parameters:: parameter (sgr_specification.v0.generic.base_types.DynamicParameterDescriptionListElement)

label

name

type

description

default_value

__str__()

Converts to string.

Returns:: A string
Return type:: str

__repr__()

Gets an object description as string.

Returns:: A string
Return type:: str

class sgr_commhandler.api.DataPoint(protocol, validator)

Bases: Generic[TFpSpec, TDpSpec]

Implements a data point of a generic data type.

Parameters:

protocol (DataPointProtocol[TFpSpec, TDpSpec])
validator (DataPointValidator)

_protocol

_validator

name()

Gets the data point name.

Returns:: the functional profile and data point name
Return type:: tuple[str, str]

async get_value_async(parameters=None, skip_cache=False)

Gets the data point value asynchronously.

Returns:

the data point value

Return type:

Any

Raises:

Exception – when read value is not compatible with data type

Parameters:

parameters (Optional[dict[str, str]])
skip_cache (bool)

async set_value_async(value)

Sets the data point value asynchronously.

Parameters:: value (Any) – the data point value

subscribe(fn)

Subscribes to data point value changes.

Parameters:: fn (Callable[[DataPointProtocol, Any], NoReturn]) – the handler method

unsubscribe(): Unsubscribes from data point value changes.

direction()

Gets the data point direction.

Returns:: the data point direction
Return type:: DataDirectionProduct

data_type()

Gets the data point data type.

Returns:: the data point data type
Return type:: DataTypes

unit()

Gets the unit of measurement of the data point.

Returns:: the unit
Return type:: Units

describe()

Describes the data point.

Returns:: the data point information
Return type:: tuple[tuple[str, str], DataDirectionProduct, DataTypes, Units]

options()

Describes the data point options.

Returns:: the data point options
Return type:: list[Any]

dynamic_parameters()

Gets the dynamic parameters of the data point.

Returns:: the dynamic parameters
Return type:: list[DynamicParameter]

get_specification()

Gets the data point specification.

Returns:: the data point specification
Return type:: TDpSpec

class sgr_commhandler.api.DataPointProtocol(fp_spec, dp_spec)

Bases: abc.ABC, Generic[TFpSpec, TDpSpec]

Implements a base class for data point protocols.

Parameters:

fp_spec (TFpSpec)
dp_spec (TDpSpec)

_fp_spec: TFpSpec

_dp_spec: TDpSpec

_fp_name: str

_dp_name: str

_dynamic_parameters: list[sgr_commhandler.api.dynamic_parameter.DynamicParameter]

get_specification()

Gets the data point specification.

Returns:: the interface-specific specification
Return type:: TDpSpec

abstract set_val(value)

Async:
Parameters:: value (Any)

Writes the data point value.

Parameters:: value (Any) – the data point value to write

abstract get_val(parameters=None, skip_cache=False)

Async:

Parameters:

parameters (Optional[dict[str, str]])
skip_cache (bool)

Return type:

Any

Reads the data point value.

Parameters:

parameters (Optional[dict[str, str]]) – optional dynamic parameters of the request
skip_cache (bool) – does not use cache if true

Returns:

the data point value

Return type:

Any

name()

Gets the functional profile and data point names.

Returns:: the functional profile and data point names as tuple
Return type:: tuple[str, str]

direction()

Gets the data direction of the data point.

Returns:: the data point direction
Return type:: DataDirectionProduct

unit()

Gets the unit of measurement of the data point.

Returns:: the unit
Return type:: Units

dynamic_parameters()

Gets the dynamic parameters of the data point.

Returns:: the dynamic parameters
Return type:: list[DynamicParameter]

can_subscribe()

Defines if subscribe() is allowed.

Returns:: True if allowed, False otherwise
Return type:: bool

subscribe(fn)

Subscribes to changes of the data point value.

Parameters:: fn (Callable[[DataPointProtocol, Any], NoReturn]) – the callback method

unsubscribe(): Unsubscribes from changes of the data point value.

class sgr_commhandler.api.DataPointValidator(data_type)

Bases: object

Implements a base class for data point validators.

Parameters:: data_type (sgr_commhandler.api.data_types.DataTypes)

_data_type: sgr_commhandler.api.data_types.DataTypes

validate(value)

Validates the compatibility of a value.

Parameters:: value (Any) – the value to validate
Returns:: true if value is compatible, false otherwise
Return type:: bool

data_type()

Gets the validator’s data type.

Returns:: the data type enumeration
Return type:: DataTypes

options()

Gets the validator’s options.

Returns:: the options
Return type:: list[Any]

class sgr_commhandler.api.DeviceInformation

Implements a device information container.

name

Defines the device name

Type:: str

manufacturer

Defines the manufacturer’s name

Type:: str

software_revision

Defines the device software version

Type:: str

hardware_revision

Defines the device hardware version

Type:: str

device_category

Defines the device category

Type:: DeviceCategory

is_local

Defines if the device is controlled locally or via cloud

Type:: bool

name: str

manufacturer: str

software_revision: str

hardware_revision: str

device_category: sgr_specification.v0.generic.DeviceCategory

is_local: bool

class sgr_commhandler.api.SGrBaseInterface(frame)

Bases: abc.ABC

Defines an abstract base class for all SGr device interfaces.

Parameters:: frame (sgr_specification.v0.product.product.DeviceFrame)

device_frame

the device specification

Type:: DeviceFrame

configuration_parameters

the configuration parameters with default values

Type:: list[ConfigurationParameter]

device_information

the device information

Type:: DeviceInformation

functional_profiles

the configured functional profiles

Type:: dict[str, FunctionalProfile]

device_frame: sgr_specification.v0.product.product.DeviceFrame

configuration_parameters: list[sgr_commhandler.api.configuration_parameter.ConfigurationParameter]

device_information: DeviceInformation

functional_profiles: dict[str, sgr_commhandler.api.functional_profile_api.FunctionalProfile]

abstract connect_async()

Async:

Connects the device asynchronously.

abstract disconnect_async()

Async:

Disconnects the device asynchronously.

abstract is_connected()

Gets the connection state.

Returns:: the connection state
Return type:: bool

get_functional_profiles()

Gets all functional profiles.

Returns:: all functional profiles
Return type:: dict[str, FunctionalProfile]

get_functional_profile(functional_profile_name)

Gets a functional profile.

Parameters:: functional_profile_name (str) – the functional profile name
Returns:: a functional profile
Return type:: FunctionalProfile

get_data_point(dp)

Gets a data point.

Parameters:: dp (tuple[str, str]) – the functional profile and data point name
Returns:: a data point
Return type:: DataPoint

get_data_points()

Gets all data points.

Returns:: all data points
Return type:: dict[tuple[str, str], DataPoint]

async get_values_async(parameters=None)

Gets all data point values asynchronously.

Returns:: all data point values
Return type:: dict[tuple[str, str], Any]
Parameters:: parameters (Optional[dict[str, str]])

describe()

Gets the device description and all data.

Returns:: a tuple of device name and all data point values
Return type:: tuple[str, dict[str, dict[str, tuple[DataDirectionProduct, DataTypes]]]]

get_specification()

Gets the complete SGr specification of the device.

Returns:: the device specification
Return type:: DeviceFrame

class sgr_commhandler.api.FunctionalProfile(fp_spec)

Bases: Generic[TFpSpec]

Implements a functional profile.

Parameters:: fp_spec (TFpSpec)

_fp_spec: TFpSpec

name()

Gets the functional profile name.

Returns:: the functional profile name
Return type:: str

get_data_points()

Gets all data points.

Returns:: all data points
Return type:: dict[tuple[str, str], DataPoint]

get_data_point(dp_name)

Gets a data point.

Parameters:: dp_name (str) – the data point name
Returns:: a data point
Return type:: DataPoint

async get_values_async(parameters=None)

Gets all data point values of the functional profile asynchronously.

Returns:: all data point values by name
Return type:: dict[str, Any]
Parameters:: parameters (Optional[dict[str, str]])

describe()

Describes the functional profile.

Returns:: the functional profile information
Return type:: tuple[str, dict[str, tuple[DataDirectionProduct, DataTypes]]]

get_specification()

Gets the functional profile specification.

Returns:: the functional profile specification
Return type:: TFpSpec