sgr_commhandler.api.data_point_api

Provides the data-point-level API.

Attributes

`TFpSpec`	Defines a generic functional profile data type.
`TDpSpec`	Defines a generic data point data type.

Classes

`DataPointValidator`	Implements a base class for data point validators.
`DataPointProtocol`	Implements a base class for data point protocols.
`DataPoint`	Implements a data point of a generic data type.

Module Contents

sgr_commhandler.api.data_point_api.TFpSpec: Defines a generic functional profile data type.

sgr_commhandler.api.data_point_api.TDpSpec: Defines a generic data point data type.

class sgr_commhandler.api.data_point_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.data_point_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.data_point_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