AWS Connection Module

JciHitachi.aws_connection.random() → x in the interval [0, 1).

class JciHitachi.aws_connection.AWSTokens(access_token: 'str', id_token: 'str', refresh_token: 'str', expiration: 'float')[source]: Bases: object

class JciHitachi.aws_connection.AWSIdentity(identity_id: 'str', host_identity_id: 'str', user_name: 'str', user_attributes: 'dict')[source]: Bases: object

class JciHitachi.aws_connection.JciHitachiMqttEvents(device_status: 'dict[str, JciHitachiAWSStatus]' = <factory>, device_support: 'dict[str, JciHitachiAWSStatusSupport]' = <factory>, device_control: 'dict[str, dict]' = <factory>, device_shadow: 'dict[str, dict]' = <factory>, mqtt_error: 'str' = <factory>, device_status_event: 'dict[str, threading.Event]' = <factory>, device_support_event: 'dict[str, threading.Event]' = <factory>, device_control_event: 'dict[str, threading.Event]' = <factory>, device_shadow_event: 'dict[str, threading.Event]' = <factory>, mqtt_error_event: 'threading.Event' = <factory>)[source]: Bases: object

class JciHitachi.aws_connection.JciHitachiExecutionPools(status_execution_pool: 'list' = <factory>, shadow_execution_pool: 'list' = <factory>, support_execution_pool: 'list' = <factory>, control_execution_pool: 'list' = <factory>)[source]: Bases: object

class JciHitachi.aws_connection.JciHitachiAWSHttpConnection(print_response: bool)[source]

Bases: ABC

Abstract class for AWS http connections.

class JciHitachi.aws_connection.JciHitachiAWSCognitoConnection(email: str, password: str, aws_tokens: AWSTokens | None = None, proxy: str | None = None, print_response: bool = False)[source]

Bases: JciHitachiAWSHttpConnection

Connecting to Jci-Hitachi AWS Cognito API.

Parameters:

email (str) – User email.
password (str) – User password.
aws_tokens (AWSTokens, optional) – If aws_tokens is given, it is used by request; otherwise, a login procedure is performed to obtain new aws_tokens, by default None.
proxy (str, optional) – Proxy setting. Format:”schema://IP:port”, e.g., http://127.0.0.1:8080, by default None.
print_response (bool, optional) – If set, all responses of httpx will be printed, by default False.

login(use_refresh_token: bool = False)[source]

Login API.

Parameters:: use_refresh_token (bool, optional) – Whether or not to use AWSTokens.refresh_token to login. If AWSTokens is not provided, fallback to email and password, by default False
Returns:: (status, aws tokens).
Return type:: (str, AWSTokens)

class JciHitachi.aws_connection.ChangePassword(email, password, **kwargs)[source]

Bases: JciHitachiAWSCognitoConnection

API internal endpoint. https://docs.aws.amazon.com/cognito-user-identity-pools/latest/APIReference/API_ChangePassword.html

Parameters:

email (str) – User email.
password (str) – User password.

class JciHitachi.aws_connection.GetUser(email, password, **kwargs)[source]

Bases: JciHitachiAWSCognitoConnection

API internal endpoint. https://docs.aws.amazon.com/cognito-user-identity-pools/latest/APIReference/API_GetUser.html

Parameters:

email (str) – User email.
password (str) – User password.

class JciHitachi.aws_connection.GetCredentials(email, password, **kwargs)[source]

Bases: JciHitachiAWSCognitoConnection

API internal endpoint.

Parameters:

email (str) – User email.
password (str) – User password.

class JciHitachi.aws_connection.JciHitachiAWSIoTConnection(aws_tokens: AWSTokens, proxy: str | None = None, print_response: bool = False)[source]

Bases: JciHitachiAWSHttpConnection

Connecting to Jci-Hitachi AWS IoT API.

Parameters:

aws_tokens (AWSTokens) – AWS tokens.
proxy (str, optional) – Proxy setting. Format:”schema://IP:port”, e.g., http://127.0.0.1:8080, by default None.
print_response (bool, optional) – If set, all responses of httpx will be printed, by default False.

class JciHitachi.aws_connection.GetAllDevice(aws_tokens: AWSTokens, **kwargs)[source]

Bases: JciHitachiAWSIoTConnection

API internal endpoint.

Parameters:: aws_tokens (AWSTokens) – AWS tokens.

class JciHitachi.aws_connection.GetAllGroup(aws_tokens: AWSTokens, **kwargs)[source]

Bases: JciHitachiAWSIoTConnection

API internal endpoint.

Parameters:: aws_tokens (AWSTokens) – AWS tokens.

class JciHitachi.aws_connection.GetAllRegion(aws_tokens: AWSTokens, **kwargs)[source]

Bases: JciHitachiAWSIoTConnection

API internal endpoint.

Parameters:: aws_tokens (AWSTokens) – AWS tokens.

class JciHitachi.aws_connection.GetAvailableAggregationMonthlyData(aws_tokens: AWSTokens, **kwargs)[source]

Bases: JciHitachiAWSIoTConnection

API internal endpoint.

Parameters:: aws_tokens (AWSTokens) – AWS tokens.

class JciHitachi.aws_connection.GetHistoryEventByUser(aws_tokens: AWSTokens, **kwargs)[source]

Bases: JciHitachiAWSIoTConnection

API internal endpoint.

Parameters:: aws_tokens (AWSTokens) – AWS tokens.

class JciHitachi.aws_connection.ListSubUser(aws_tokens: AWSTokens, **kwargs)[source]

Bases: JciHitachiAWSIoTConnection

API internal endpoint.

Parameters:: aws_tokens (AWSTokens) – AWS tokens.

class JciHitachi.aws_connection.JciHitachiAWSMqttConnection(get_credentials_callable: Callable, print_response: bool = False)[source]

Bases: object

Connecting to Jci-Hitachi AWS MQTT to get latest events.

Parameters:

get_credentials_callable (Callable) – Callable which takes no arguments and returns AwsCredentials.
print_response (bool, optional) – If set, all responses of MQTT will be printed, by default False.

property mqtt_events: JciHitachiMqttEvents

MQTT events.

Returns:: See JciHitachiMqttEvents.
Return type:: JciHitachiMqttEvents

disconnect() → None[source]: Disconnect from the MQTT broker.

configure(identity_id) → None[source]

Configure MQTT.

Parameters:: identity_id (str) – Identity ID.

connect(host_identity_id: str, shadow_names: list[str] | str | None = None, thing_names: list[str] | str | None = None) → bool[source]

Connect to the MQTT broker and start loop.

Parameters:

host_identity_id (str) – Host identity ID.
shadow_names (str or list of str, optional) – Names to be subscribed in Shadow, by default None.
thing_names (str or list of str, optional) – Things to be subscribed in Shadow, by default None.

Returns:

A bool indicating whether the mqtt is successfully connected and subscribed.

Return type:

bool

publish(host_identity_id: str, thing_name: str, publish_type: str, timeout: float = 10.0, payload: dict | None = None) → None[source]

Put messages to be published in the execution pool. execute() should be called to start async publish.

Parameters:

host_identity_id (str) – Host identity id.
thing_name (str) – Thing name.
publish_type (str) – Publish type. There are three types available: support, status, and control.
timeout (float, optional) – Timeout for messages published, by default 10.0.
payload (dict, optional) – Payload to publish, by default None.

publish_shadow(thing_name: str, command_name: str, payload: dict = {}, shadow_name: str | None = None, timeout: float = 10.0) → None[source]

Publish message to IoT Shadow Service.

Parameters:

thing_name (str) – Thing name.
command_name (str) – Command name, which can be get or update.
payload (dict, optional) – Payload to publish, by default {}.
shadow_name (str, optional) – Shadow name, by default None.
timeout (float, optional) – Timeout for messages published, by default 10.0.

execute(control: bool = False) → list[list[str | BaseException], list[str | BaseException], list[str | BaseException], list[str | BaseException]][source]

Execute publish commands in the execution pools.

Parameters:: control (bool) – If True, commands in the control_execution_pool will be executed; otherwise, commands in other execution pools will be executed.
Returns:: Execution results of support, shadow, status, control, respectively. Each result is a list containing thing names if the execution was successful or BaseException(s) if an error occurred during execution.
Return type:: list