CMD API Client

Bases: BaseAPIClient

API client for CMD Report API.

It is used to communicate with CMD Report API that provides the data useful for Drive Report Generation process. Origin of this data is from the CMD Azure Function.

API documentation can be found [here] (https://dev.monitoring.drives.abb.com/extapi/swagger/index.html?urls.primaryName=Report%20API)

Source code in reportconnectors/api_client/cmd_api_client.py

class CmdAPIClient(BaseAPIClient):
    """
    API client for CMD Report API.

    It is used to communicate with CMD Report API that provides the data useful for Drive Report Generation process.
    Origin of this data is from the CMD Azure Function.

    API documentation can be found [here]
    (https://dev.monitoring.drives.abb.com/extapi/swagger/index.html?urls.primaryName=Report%20API)

    """

    __version__ = "0.4.0"
    _datetime_format: str = "%Y-%m-%dT%H:%M:%SZ"

    class KeyNames(BaseAPIClient.KeyNames):
        START_DATE = "startDate"
        END_DATE = "endDate"
        SERIAL_NUMBER = "serialNumber"
        TREND_NAME = "trendName"
        SCATTER_NAME = "scatterName"
        ALGORITHM = "algorithm"
        ALGORITHM_NAME = "algorithmName"
        ALGORITHM_PROPERTY = "algorithmProperty"
        METHOD = "method"
        EMAIL = "email"
        PAYLOAD = "payload"
        SUCCESS = "success"
        ERRORS = "errors"

    def __init__(self, url: str, **kwargs):
        super().__init__(url=url, **kwargs)
        self.report_api_suffix = kwargs.get("report_api_suffix", "/extapi/report/202208")
        self.cloud_api_suffix = kwargs.get("cloud_api_suffix", "/extapi/api/v2")

    def authenticate(self, client_id: str, client_secret: str, token_url: str, **kwargs) -> bool:
        """
        Authenticate to CMD API using Client ID and Client Secret parameters.
        To authenticate external identity provider is used. It is defined by `token_url` parameter

        Args:
            client_id: Client ID
            client_secret: Client secret
            token_url: Authentication token URL

        Keyword Args:
            timeout (float): Timeout for the authentication process. Default: `60`

        Returns:
            True if client is successfully authenticated and access token in obtained and set. Otherwise, False.
        """

        token_response = self._get_openid_token(
            client_id=client_id, client_secret=client_secret, token_url=token_url, **kwargs
        )

        self.auth_data[self.KeyNames.ACCESS_TOKEN] = token_response.get("access_token")
        self.auth_data[self.KeyNames.REFRESH_TOKEN] = token_response.get("refresh_token")
        _exp_timestamp = token_response.get("expires_at")
        _exp_at = (
            datetime.datetime.fromtimestamp(_exp_timestamp).replace(tzinfo=datetime.timezone.utc)
            if isinstance(_exp_timestamp, float)
            else None
        )
        self.auth_data[self.KeyNames.EXPIRATION_DATE] = _exp_at
        self.auth_data[self.KeyNames.EMAIL] = self._get_email_from_auth_token(auth_token=str(self.auth_token))
        return self.is_logged

    def set_auth_token(self, auth_token: str, **kwargs) -> bool:
        """
        Sets the authorization token without the authentication process. It might be used when the authorization
        token is obtained from external source.

        We expect this token to be Bearer token. To validate it, we use PyJWT library to decode it, and
        extract the expiration date

        To verify if the token is valid, the `validate_auth_token` method is called afterward.

        Args:
            auth_token: Authentication token to be set.

        Keyword Args:
            refresh_token (str): Refresh token to be set.

        Returns:
            True if provided token is set and valid. Otherwise, False.

        """

        try:
            decoded_token = jwt.decode(auth_token, options={"verify_signature": False})
            _exp_timestamp = float(decoded_token.get("exp"))  # type: ignore
            expiration_date = datetime.datetime.fromtimestamp(_exp_timestamp).replace(tzinfo=datetime.timezone.utc)
        except (ValueError, TypeError, jwt.exceptions.InvalidTokenError):
            log.error("Invalid token provided.")
            return False

        # Set new tokens ...
        self.auth_data[self.KeyNames.ACCESS_TOKEN] = auth_token
        self.auth_data[self.KeyNames.REFRESH_TOKEN] = kwargs.get("refresh_token")
        self.auth_data[self.KeyNames.EXPIRATION_DATE] = expiration_date
        self.auth_data[self.KeyNames.EMAIL] = self._get_email_from_auth_token(auth_token=auth_token)
        # ... and validate them
        validation_results = self.validate_auth_token()
        return validation_results

    @property
    def auth_token(self) -> Optional[str]:
        _auth_token = self.auth_data.get(self.KeyNames.ACCESS_TOKEN, None)
        return _auth_token

    @property
    def is_logged(self) -> bool:
        """
        Checks if the client is still logged in to the API.
        """

        _is_logged = (
            bool(self.auth_token)
            and isinstance(self.token_expiration_date, datetime.datetime)
            and (self.token_expiration_date > datetime.datetime.now(datetime.timezone.utc))
        )
        return _is_logged

    @classmethod
    def _get_request_params(
        cls,
        serial_number: str,
        start_date: Optional[DateType] = None,
        end_date: Optional[Union[datetime.datetime, datetime.date]] = None,
        other: Optional[Dict] = None,
    ) -> Dict:
        """
        A helper function for creating a parameter dict.

        Args:
            serial_number: Drive serial number
            start_date: Start datetime
            end_date: End datetime
            other: Other request parameters as a dict

        Returns:
            Dictionary with request parameters formatted according to CMD API specification

        """
        params = {cls.KeyNames.SERIAL_NUMBER: serial_number}
        if start_date and end_date:
            params[cls.KeyNames.START_DATE] = start_date.strftime(cls._datetime_format)
            params[cls.KeyNames.END_DATE] = end_date.strftime(cls._datetime_format)
        if other:
            params.update(other)
        return params

    def get_results(self, serial_number: str, start_date: DateType, end_date: DateType, algorithm_name: str) -> Dict:
        """
        Gets the results of given algorithm calculations from the `/results` endpoint for requested
        drive serial number and time range.

        Args:
            serial_number: Drive Serial Number
            start_date: Start Date
            end_date: End Date
            algorithm_name: Name of the algorithm, as given in type definitions.

        Returns:
            Algorithm calculation results.

        Response Example:
            ```json
            {
              "payload": {
                "serialNumber": "2UEA902000692",
                "algorithm": "PowerTrend",
                "startDate": "2022-08-01T00:00:00+00:00",
                "endDate": "2022-08-30T00:00:00+00:00",
                "results": [
                  {
                    "date": "2022-08-01T07:00:00Z",
                    "details": {
                      "PowerTrendAmin": 0.0,
                      "PowerTrendAmax": 1807.388916,
                      "PowerTrendRms": 673.2872564275,
                      "PowerTrendDelta": 1807.388916,
                      "PowerTrendStd": 622.9944096008,
                      "PowerTrendCalculate_Average": 255.3305607908
                    }
                  }
                ]
              },
              "success": true
            }
            ```
        """
        endpoint = self._join_url(self.report_api_suffix, "results")
        params = self._get_request_params(
            serial_number=serial_number,
            start_date=start_date,
            end_date=end_date,
            other={self.KeyNames.ALGORITHM_NAME: algorithm_name},
        )
        response = self._make_request(method="GET", endpoint=endpoint, params=params)
        decoded_response: Dict = self._decode_response(response=response, default={})
        return decoded_response

    def get_trend(self, serial_number: str, start_date: DateType, end_date: DateType, trend_name: str) -> Dict:
        """
        Gets the trend data from the `/trend` endpoint for requested
        drive serial number and time range.

        Args:
            serial_number: Drive Serial Number
            start_date: Start Date
            end_date: End Date
            trend_name: Name of the trend (aka algorithm name in type definition).

        Returns:
            Trend Data.

        Response Example:
            ```json
            {
              "payload": {
                "serialNumber": "2UEA902000692",
                "algorithm": "PowerTrend",
                "startDate": "2022-08-01T00:00:00+00:00",
                "endDate": "2022-08-30T00:00:00+00:00",
                "results": [
                  {
                    "date": "2022-08-01T07:00:00Z",
                    "details": {
                      "PowerTrendAmin": 0.0,
                      "PowerTrendAmax": 1807.388916,
                      "PowerTrendRms": 673.2872564275,
                      "PowerTrendDelta": 1807.388916,
                      "PowerTrendStd": 622.9944096008,
                      "PowerTrendCalculate_Average": 255.3305607908
                    }
                  }
                ]
              },
              "success": true
            }
            ```
        """

        endpoint = self._join_url(self.report_api_suffix, "trend")
        params = self._get_request_params(
            serial_number=serial_number,
            start_date=start_date,
            end_date=end_date,
            other={self.KeyNames.TREND_NAME: trend_name},
        )
        response = self._make_request(method="GET", endpoint=endpoint, params=params)
        decoded_response: Dict = self._decode_response(response=response, default={})
        return decoded_response

    def get_grouped_events(self, serial_number: str, start_date: DateType, end_date: DateType) -> Dict:
        """
        Gets the grouped events from the `/events/grouped` endpoint for requested
        drive serial number and time range.

        Args:
            serial_number: Drive Serial Number
            start_date: Start Date
            end_date: End Date

        Returns:
            Grouped events.

        Response Example:
            ```json
            {
              "payload": {
                "serialNumber": "2UEA902000692",
                "algorithm": "EventGroupingDay",
                "startDate": "2022-08-01T00:00:00+00:00",
                "endDate": "2022-08-30T00:00:00+00:00",
                "results": [
                  {
                    "date": "2022-08-01T00:00:00Z",
                    "details": {
                      "warningCount": 26,
                      "faultCount": 33,
                      "faultCountGrouped": 1,
                      "warningCountGrouped": 3,
                      "df": {}
                    }
                  }
                ]
              },
              "success": true
            }
            ```
        """
        endpoint = self._join_url(self.report_api_suffix, "/events/grouped")
        params = self._get_request_params(
            serial_number=serial_number,
            start_date=start_date,
            end_date=end_date,
        )
        response = self._make_request(method="GET", endpoint=endpoint, params=params)
        decoded_response: Dict = self._decode_response(response=response, default={})
        return decoded_response

    def get_scatter(self, serial_number: str, start_date: DateType, end_date: DateType, scatter_name: str) -> Dict:
        """
        Gets the scatter data from the `/scatter` endpoint for requested
        drive serial number and time range.

        Args:
            serial_number: Drive Serial Number
            start_date: Start Date
            end_date: End Date
            scatter_name: Name of the scatter (aka algorithm name in type definition).

        Returns:
            Scatter data.

        Response Example:
            ```json
            {
                "payload": {
                    "serialNumber": "2UEA902000692",
                    "algorithm": "Current_Volt",
                    "startDate": "2022-08-01T00:00:00+00:00",
                    "endDate": "2022-08-30T00:00:00+00:00",
                    "scatter": [
                        {"x": -5.0, "y": 14.0, "cnt": 4.0},
                        {"x": -5.0, "y": 28.0, "cnt": 2.0},
                        {"x": 0.0, "y": 14.0, "cnt": 8.0},
                        {"x": 0.0, "y": 6609.0, "cnt": 1.0},
                        {"x": 3.0, "y": 14.0, "cnt": 27.0},
                    ]
                },
                "success": true
            }
            ```
        """
        endpoint = self._join_url(self.report_api_suffix, "scatter")
        params = self._get_request_params(
            serial_number=serial_number,
            start_date=start_date,
            end_date=end_date,
            other={self.KeyNames.SCATTER_NAME: scatter_name},
        )
        response = self._make_request(method="GET", endpoint=endpoint, params=params)
        decoded_response: Dict = self._decode_response(response=response, default={})
        return decoded_response

    def get_drive_details(self, serial_number: str) -> Dict:
        """
        Gets drive details data from `/drive` endpoint for a provided drive serial number

        Args:
            serial_number: Drive Serial Number

        Returns:
            Drive details.

        Response example:
            ```json
            {
              "payload": {
                "drive": {
                  "serialNumber": "2UEA902000692",
                  "type": "acs2000",
                  "connection": null,
                  "firmwareVersion": "LDAI6251",
                  "loadingPackage": "LDOI6351",
                  "deviceName": "NETA-21 9050163 01",
                  "name": "NETA-21 9050163 01",
                },
                "gateway": {
                  "type": "Neta",
                  "serialNumber": "9050163",
                  "firmwareVersion": "3.30.0.0",
                  "packageVersion": "3.30-r1.1"
                },
                "sensor": null,
                "driveRegistration": {
                  "product": "ACS2000",
                },
                "preventiveSchedule": {
                  "dcCapacitor": null,
                  "igbt": null,
                  "coolingFan": null
                },
                "site": {
                  "country": "CHILE",
                  "countryCode": "CL",
                  "latitude": -26.8581658,
                  "longitude": -70.8133369
                }
              },
              "success": true
            }

            ```
        """
        endpoint = self._join_url(self.report_api_suffix, "drive")
        params = self._get_request_params(serial_number=serial_number)
        response = self._make_request(method="GET", endpoint=endpoint, params=params)
        decoded_response: Dict = self._decode_response(response=response, default={})
        return decoded_response

    def get_aggregates(
        self,
        serial_number: str,
        start_date: DateType,
        end_date: DateType,
        algorithm: str,
        algorithm_property: str,
        method: str,
    ) -> Dict:
        """
        Gets the 30 days aggregates for each using the `/aggregates` endpoint for requested
        drive serial number and time range.

        Args:
            serial_number: Drive Serial Number
            start_date: Start Date
            end_date: End Date
            algorithm: Name of the algorithm, as given in type definitions.
            algorithm_property: Name of the algorithm property. This can be any property
                that is produced by the specified algorithm.
            method: Name of the statistics method, as accepted by Statistics function. Transparent to API.

        Returns:
            Aggregated data.

        Response example:
            ```json
            {
              "payload": {
                "algorithmProperty": "run",
                "method": "max",
                "serialNumber": "2UEA902000692",
                "algorithm": "InherentAvailabilityDay",
                "startDate": "2022-08-01T00:00:00+00:00",
                "endDate": "2022-08-30T00:00:00+00:00",
                "results": [
                  {
                    "date": "2022-08-01T00:00:00Z",
                    "details": 64720000.0
                  },
                  {
                    "date": "2022-08-02T00:00:00Z",
                    "details": 64720000.0
                  }
                ]
              },
              "success": true
            }
            ```
        """
        endpoint = self._join_url(self.report_api_suffix, "aggregates")
        params = self._get_request_params(
            serial_number=serial_number,
            start_date=start_date,
            end_date=end_date,
            other={
                self.KeyNames.ALGORITHM: algorithm,
                self.KeyNames.ALGORITHM_PROPERTY: algorithm_property,
                self.KeyNames.METHOD: method,
            },
        )
        response = self._make_request(method="GET", endpoint=endpoint, params=params)
        decoded_response: Dict = self._decode_response(response=response, default={})
        return decoded_response

    def get_list_of_serial_numbers(
        self, email: Optional[str] = None, check_for_assigned_drives: bool = False
    ) -> List[str]:
        """
        Gets a list of serial numbers assigned to a user using the `Cloud Interface API`
        endpoint `/devices`.

        It requires user email to be provided, to identify the user.
        If email is not provided then the email from auth token is used.

        Args:
            email: User email to check. Default `<EMAIL_FROM_AUTH_TOKEN>`
            check_for_assigned_drives: If True then the CMD API will check for assigned drives.
                This takes a lot of time and should be done with caution. Default: `False`

        Returns:
            List of serial numbers that user has access to.

        """
        endpoint = self._join_url(self.cloud_api_suffix, "devices")

        email = email if email else self.auth_data.get(self.KeyNames.EMAIL, "")
        data = {"checkForAssignedDrives": check_for_assigned_drives, "email": email}

        response = self._make_request(method="POST", endpoint=endpoint, json_data=data)
        decoded_response: Dict = self._decode_response(response=response, default={})

        list_of_serial_numbers = decoded_response.get(self.KeyNames.PAYLOAD, [])
        is_success = decoded_response.get(self.KeyNames.SUCCESS, False)

        if not is_success or not list_of_serial_numbers:
            return []
        return list_of_serial_numbers

    def get_list_of_drive_details(self, serial_numbers: Sequence[str], email: Optional[str] = None) -> List[Dict]:
        """
        Gets a list of drive details assigned to a user using the `Cloud Interface API`
        endpoint `/devices/detail`.

        It requires list of serial numbers to identify the drives and user email to identify the user.
        If email is not provided then the email from auth token is used.

        Args:
            serial_numbers: List of drives serial numbers for which we want to get the details.
            email: Email of user that has access to the drives. Default `<EMAIL_FROM_AUTH_TOKEN>`

        Returns:
            List of drive details.

        """
        endpoint = self._join_url(self.cloud_api_suffix, "devices/detail")

        email = email if email else self.auth_data.get(self.KeyNames.EMAIL, "")
        data = {"serialNumbers": serial_numbers, "email": email}

        response = self._make_request(method="POST", endpoint=endpoint, json_data=data)
        decoded_response: Dict = self._decode_response(response=response, default={})

        list_of_drive_details = decoded_response.get(self.KeyNames.PAYLOAD, {}).get("items", [])
        is_success = decoded_response.get(self.KeyNames.SUCCESS, False)

        if not is_success or not list_of_drive_details:
            return []
        return list_of_drive_details

    def validate_auth_token(self) -> bool:
        """
        Validates current authorization token by calling the `/validate` endpoint.

        If the token is valid then the endpoint responds with the OK status and `{"success": True}` response

        Returns:
            True if the token is valid, False otherwise.

        """
        endpoint = self._join_url(self.report_api_suffix, "validate")
        response = self._make_request(method="GET", endpoint=endpoint)
        decoded_response: Dict = self._decode_response(response=response, default={})
        status_code = self._get_status_code(response=response)
        is_valid = (status_code == HTTPStatus.OK) and (decoded_response.get("success") is True)
        return is_valid

    def confirm_report_generation(self, token: str) -> bool:
        """
        Sends confirmation about report generation process being done.

        Report generation process is identified by `token` argument.

        Args:
            token: Report generation process identifier.

        Returns:
            True if confirmation is accepted, False otherwise.

        """
        endpoint = self._join_url(self.report_api_suffix, "completed")
        params = {"token": token}

        response = self._make_request(method="GET", endpoint=endpoint, params=params)
        decoded_response: Dict = self._decode_response(response=response, default={})
        status_code = self._get_status_code(response=response)
        self._log_response_errors(response_content=decoded_response)
        is_confirmed = (status_code == HTTPStatus.OK) and (decoded_response.get("success") is True)
        return is_confirmed

    def _get_openid_token(self, client_id: str, client_secret: str, token_url: str, **kwargs) -> Dict:
        """
        Gets OpenID token by using backend authentication flow.

        Args:
            client_id: Client ID
            client_secret: Client Secret
            token_url: Token URL

        Keyword Args:
            timeout (int): Timeout for the authentication process. Default: `60`
            content_type (str): Content type of the request. Default: `application/x-www-form-urlencoded`

        Returns:
             True if valid token is received, False otherwise.

        Notes:
            When authorizing at CIAM the User-Agent header is expected. That is the
            reason why we add the custom headers to the fetch_token method call.
        """
        timeout = kwargs.get("timeout", self._default_timeout)
        content_type = kwargs.get("content_type", "application/x-www-form-urlencoded")

        oauth_client = OAuth2Session(client=BackendApplicationClient(client_id=client_id))

        token_response = oauth_client.fetch_token(
            token_url=token_url,
            client_id=client_id,
            client_secret=client_secret,
            include_client_id=True,
            proxies=self._proxies,
            timeout=timeout,
            headers=self._prepare_headers(content_type=content_type),
        )
        return token_response

    def _log_response_errors(self, response_content: Dict) -> None:
        for error in response_content.get(self.KeyNames.ERRORS, []):
            log.debug(f"Error received from CMD API: {error}")

    @staticmethod
    def _get_email_from_auth_token(auth_token: str, field: str = "sub") -> Optional[str]:
        """
        Extracts email information from auth token.

        It supports the 'double at' case.
        e.g. if the field value is `admin@abb.com@carbon.super` then it correctly returns `admin@abb.com`

        Args:
            auth_token: Authentication token to examine
            field: Token field to examine. Default `sub`

        Returns:
            Optional email extracted from auth token

        """
        token = jwt.decode(auth_token, options={"verify_signature": False})
        value = token.get(field, "")
        if "@" not in value:
            return None

        elements = value.split("@")
        email = f"{elements[0]}@{elements[1]}"
        return email

`is_logged` `property`

Checks if the client is still logged in to the API.

`authenticate(client_id, client_secret, token_url, **kwargs)`

Authenticate to CMD API using Client ID and Client Secret parameters. To authenticate external identity provider is used. It is defined by token_url parameter

Parameters:

Name	Type	Description	Default
`client_id`	`str`	Client ID	required
`client_secret`	`str`	Client secret	required
`token_url`	`str`	Authentication token URL	required

Other Parameters:

Name	Type	Description
`timeout`	`float`	Timeout for the authentication process. Default: `60`

Returns:

Type	Description
`bool`	True if client is successfully authenticated and access token in obtained and set. Otherwise, False.

Source code in reportconnectors/api_client/cmd_api_client.py

def authenticate(self, client_id: str, client_secret: str, token_url: str, **kwargs) -> bool:
    """
    Authenticate to CMD API using Client ID and Client Secret parameters.
    To authenticate external identity provider is used. It is defined by `token_url` parameter

    Args:
        client_id: Client ID
        client_secret: Client secret
        token_url: Authentication token URL

    Keyword Args:
        timeout (float): Timeout for the authentication process. Default: `60`

    Returns:
        True if client is successfully authenticated and access token in obtained and set. Otherwise, False.
    """

    token_response = self._get_openid_token(
        client_id=client_id, client_secret=client_secret, token_url=token_url, **kwargs
    )

    self.auth_data[self.KeyNames.ACCESS_TOKEN] = token_response.get("access_token")
    self.auth_data[self.KeyNames.REFRESH_TOKEN] = token_response.get("refresh_token")
    _exp_timestamp = token_response.get("expires_at")
    _exp_at = (
        datetime.datetime.fromtimestamp(_exp_timestamp).replace(tzinfo=datetime.timezone.utc)
        if isinstance(_exp_timestamp, float)
        else None
    )
    self.auth_data[self.KeyNames.EXPIRATION_DATE] = _exp_at
    self.auth_data[self.KeyNames.EMAIL] = self._get_email_from_auth_token(auth_token=str(self.auth_token))
    return self.is_logged

`confirm_report_generation(token)`

Sends confirmation about report generation process being done.

Report generation process is identified by token argument.

Parameters:

Name	Type	Description	Default
`token`	`str`	Report generation process identifier.	required

Returns:

Type	Description
`bool`	True if confirmation is accepted, False otherwise.

Source code in reportconnectors/api_client/cmd_api_client.py

def confirm_report_generation(self, token: str) -> bool:
    """
    Sends confirmation about report generation process being done.

    Report generation process is identified by `token` argument.

    Args:
        token: Report generation process identifier.

    Returns:
        True if confirmation is accepted, False otherwise.

    """
    endpoint = self._join_url(self.report_api_suffix, "completed")
    params = {"token": token}

    response = self._make_request(method="GET", endpoint=endpoint, params=params)
    decoded_response: Dict = self._decode_response(response=response, default={})
    status_code = self._get_status_code(response=response)
    self._log_response_errors(response_content=decoded_response)
    is_confirmed = (status_code == HTTPStatus.OK) and (decoded_response.get("success") is True)
    return is_confirmed

`get_aggregates(serial_number, start_date, end_date, algorithm, algorithm_property, method)`

Gets the 30 days aggregates for each using the /aggregates endpoint for requested drive serial number and time range.

Parameters:

Name	Type	Description	Default
`serial_number`	`str`	Drive Serial Number	required
`start_date`	`DateType`	Start Date	required
`end_date`	`DateType`	End Date	required
`algorithm`	`str`	Name of the algorithm, as given in type definitions.	required
`algorithm_property`	`str`	Name of the algorithm property. This can be any property that is produced by the specified algorithm.	required
`method`	`str`	Name of the statistics method, as accepted by Statistics function. Transparent to API.	required

Returns:

Type	Description
`Dict`	Aggregated data.

Response example

{
  "payload": {
    "algorithmProperty": "run",
    "method": "max",
    "serialNumber": "2UEA902000692",
    "algorithm": "InherentAvailabilityDay",
    "startDate": "2022-08-01T00:00:00+00:00",
    "endDate": "2022-08-30T00:00:00+00:00",
    "results": [
      {
        "date": "2022-08-01T00:00:00Z",
        "details": 64720000.0
      },
      {
        "date": "2022-08-02T00:00:00Z",
        "details": 64720000.0
      }
    ]
  },
  "success": true
}

Source code in reportconnectors/api_client/cmd_api_client.py

def get_aggregates(
    self,
    serial_number: str,
    start_date: DateType,
    end_date: DateType,
    algorithm: str,
    algorithm_property: str,
    method: str,
) -> Dict:
    """
    Gets the 30 days aggregates for each using the `/aggregates` endpoint for requested
    drive serial number and time range.

    Args:
        serial_number: Drive Serial Number
        start_date: Start Date
        end_date: End Date
        algorithm: Name of the algorithm, as given in type definitions.
        algorithm_property: Name of the algorithm property. This can be any property
            that is produced by the specified algorithm.
        method: Name of the statistics method, as accepted by Statistics function. Transparent to API.

    Returns:
        Aggregated data.

    Response example:
        ```json
        {
          "payload": {
            "algorithmProperty": "run",
            "method": "max",
            "serialNumber": "2UEA902000692",
            "algorithm": "InherentAvailabilityDay",
            "startDate": "2022-08-01T00:00:00+00:00",
            "endDate": "2022-08-30T00:00:00+00:00",
            "results": [
              {
                "date": "2022-08-01T00:00:00Z",
                "details": 64720000.0
              },
              {
                "date": "2022-08-02T00:00:00Z",
                "details": 64720000.0
              }
            ]
          },
          "success": true
        }
        ```
    """
    endpoint = self._join_url(self.report_api_suffix, "aggregates")
    params = self._get_request_params(
        serial_number=serial_number,
        start_date=start_date,
        end_date=end_date,
        other={
            self.KeyNames.ALGORITHM: algorithm,
            self.KeyNames.ALGORITHM_PROPERTY: algorithm_property,
            self.KeyNames.METHOD: method,
        },
    )
    response = self._make_request(method="GET", endpoint=endpoint, params=params)
    decoded_response: Dict = self._decode_response(response=response, default={})
    return decoded_response

`get_drive_details(serial_number)`

Gets drive details data from /drive endpoint for a provided drive serial number

Parameters:

Name	Type	Description	Default
`serial_number`	`str`	Drive Serial Number	required

Returns:

Type	Description
`Dict`	Drive details.

Response example

{
  "payload": {
    "drive": {
      "serialNumber": "2UEA902000692",
      "type": "acs2000",
      "connection": null,
      "firmwareVersion": "LDAI6251",
      "loadingPackage": "LDOI6351",
      "deviceName": "NETA-21 9050163 01",
      "name": "NETA-21 9050163 01",
    },
    "gateway": {
      "type": "Neta",
      "serialNumber": "9050163",
      "firmwareVersion": "3.30.0.0",
      "packageVersion": "3.30-r1.1"
    },
    "sensor": null,
    "driveRegistration": {
      "product": "ACS2000",
    },
    "preventiveSchedule": {
      "dcCapacitor": null,
      "igbt": null,
      "coolingFan": null
    },
    "site": {
      "country": "CHILE",
      "countryCode": "CL",
      "latitude": -26.8581658,
      "longitude": -70.8133369
    }
  },
  "success": true
}

Source code in reportconnectors/api_client/cmd_api_client.py

def get_drive_details(self, serial_number: str) -> Dict:
    """
    Gets drive details data from `/drive` endpoint for a provided drive serial number

    Args:
        serial_number: Drive Serial Number

    Returns:
        Drive details.

    Response example:
        ```json
        {
          "payload": {
            "drive": {
              "serialNumber": "2UEA902000692",
              "type": "acs2000",
              "connection": null,
              "firmwareVersion": "LDAI6251",
              "loadingPackage": "LDOI6351",
              "deviceName": "NETA-21 9050163 01",
              "name": "NETA-21 9050163 01",
            },
            "gateway": {
              "type": "Neta",
              "serialNumber": "9050163",
              "firmwareVersion": "3.30.0.0",
              "packageVersion": "3.30-r1.1"
            },
            "sensor": null,
            "driveRegistration": {
              "product": "ACS2000",
            },
            "preventiveSchedule": {
              "dcCapacitor": null,
              "igbt": null,
              "coolingFan": null
            },
            "site": {
              "country": "CHILE",
              "countryCode": "CL",
              "latitude": -26.8581658,
              "longitude": -70.8133369
            }
          },
          "success": true
        }

        ```
    """
    endpoint = self._join_url(self.report_api_suffix, "drive")
    params = self._get_request_params(serial_number=serial_number)
    response = self._make_request(method="GET", endpoint=endpoint, params=params)
    decoded_response: Dict = self._decode_response(response=response, default={})
    return decoded_response

`get_grouped_events(serial_number, start_date, end_date)`

Gets the grouped events from the /events/grouped endpoint for requested drive serial number and time range.

Parameters:

Name	Type	Description	Default
`serial_number`	`str`	Drive Serial Number	required
`start_date`	`DateType`	Start Date	required
`end_date`	`DateType`	End Date	required

Returns:

Type	Description
`Dict`	Grouped events.

Response Example

{
  "payload": {
    "serialNumber": "2UEA902000692",
    "algorithm": "EventGroupingDay",
    "startDate": "2022-08-01T00:00:00+00:00",
    "endDate": "2022-08-30T00:00:00+00:00",
    "results": [
      {
        "date": "2022-08-01T00:00:00Z",
        "details": {
          "warningCount": 26,
          "faultCount": 33,
          "faultCountGrouped": 1,
          "warningCountGrouped": 3,
          "df": {}
        }
      }
    ]
  },
  "success": true
}

Source code in reportconnectors/api_client/cmd_api_client.py

def get_grouped_events(self, serial_number: str, start_date: DateType, end_date: DateType) -> Dict:
    """
    Gets the grouped events from the `/events/grouped` endpoint for requested
    drive serial number and time range.

    Args:
        serial_number: Drive Serial Number
        start_date: Start Date
        end_date: End Date

    Returns:
        Grouped events.

    Response Example:
        ```json
        {
          "payload": {
            "serialNumber": "2UEA902000692",
            "algorithm": "EventGroupingDay",
            "startDate": "2022-08-01T00:00:00+00:00",
            "endDate": "2022-08-30T00:00:00+00:00",
            "results": [
              {
                "date": "2022-08-01T00:00:00Z",
                "details": {
                  "warningCount": 26,
                  "faultCount": 33,
                  "faultCountGrouped": 1,
                  "warningCountGrouped": 3,
                  "df": {}
                }
              }
            ]
          },
          "success": true
        }
        ```
    """
    endpoint = self._join_url(self.report_api_suffix, "/events/grouped")
    params = self._get_request_params(
        serial_number=serial_number,
        start_date=start_date,
        end_date=end_date,
    )
    response = self._make_request(method="GET", endpoint=endpoint, params=params)
    decoded_response: Dict = self._decode_response(response=response, default={})
    return decoded_response

`get_list_of_drive_details(serial_numbers, email=None)`

Gets a list of drive details assigned to a user using the Cloud Interface API endpoint /devices/detail.

It requires list of serial numbers to identify the drives and user email to identify the user. If email is not provided then the email from auth token is used.

Parameters:

Name	Type	Description	Default
`serial_numbers`	`Sequence[str]`	List of drives serial numbers for which we want to get the details.	required
`email`	`Optional[str]`	Email of user that has access to the drives. Default `<EMAIL_FROM_AUTH_TOKEN>`	`None`

Returns:

Type	Description
`List[Dict]`	List of drive details.

Source code in reportconnectors/api_client/cmd_api_client.py

def get_list_of_drive_details(self, serial_numbers: Sequence[str], email: Optional[str] = None) -> List[Dict]:
    """
    Gets a list of drive details assigned to a user using the `Cloud Interface API`
    endpoint `/devices/detail`.

    It requires list of serial numbers to identify the drives and user email to identify the user.
    If email is not provided then the email from auth token is used.

    Args:
        serial_numbers: List of drives serial numbers for which we want to get the details.
        email: Email of user that has access to the drives. Default `<EMAIL_FROM_AUTH_TOKEN>`

    Returns:
        List of drive details.

    """
    endpoint = self._join_url(self.cloud_api_suffix, "devices/detail")

    email = email if email else self.auth_data.get(self.KeyNames.EMAIL, "")
    data = {"serialNumbers": serial_numbers, "email": email}

    response = self._make_request(method="POST", endpoint=endpoint, json_data=data)
    decoded_response: Dict = self._decode_response(response=response, default={})

    list_of_drive_details = decoded_response.get(self.KeyNames.PAYLOAD, {}).get("items", [])
    is_success = decoded_response.get(self.KeyNames.SUCCESS, False)

    if not is_success or not list_of_drive_details:
        return []
    return list_of_drive_details

`get_list_of_serial_numbers(email=None, check_for_assigned_drives=False)`

Gets a list of serial numbers assigned to a user using the Cloud Interface API endpoint /devices.

It requires user email to be provided, to identify the user. If email is not provided then the email from auth token is used.

Parameters:

Name	Type	Description	Default
`email`	`Optional[str]`	User email to check. Default `<EMAIL_FROM_AUTH_TOKEN>`	`None`
`check_for_assigned_drives`	`bool`	If True then the CMD API will check for assigned drives. This takes a lot of time and should be done with caution. Default: `False`	`False`

Returns:

Type	Description
`List[str]`	List of serial numbers that user has access to.

Source code in reportconnectors/api_client/cmd_api_client.py

def get_list_of_serial_numbers(
    self, email: Optional[str] = None, check_for_assigned_drives: bool = False
) -> List[str]:
    """
    Gets a list of serial numbers assigned to a user using the `Cloud Interface API`
    endpoint `/devices`.

    It requires user email to be provided, to identify the user.
    If email is not provided then the email from auth token is used.

    Args:
        email: User email to check. Default `<EMAIL_FROM_AUTH_TOKEN>`
        check_for_assigned_drives: If True then the CMD API will check for assigned drives.
            This takes a lot of time and should be done with caution. Default: `False`

    Returns:
        List of serial numbers that user has access to.

    """
    endpoint = self._join_url(self.cloud_api_suffix, "devices")

    email = email if email else self.auth_data.get(self.KeyNames.EMAIL, "")
    data = {"checkForAssignedDrives": check_for_assigned_drives, "email": email}

    response = self._make_request(method="POST", endpoint=endpoint, json_data=data)
    decoded_response: Dict = self._decode_response(response=response, default={})

    list_of_serial_numbers = decoded_response.get(self.KeyNames.PAYLOAD, [])
    is_success = decoded_response.get(self.KeyNames.SUCCESS, False)

    if not is_success or not list_of_serial_numbers:
        return []
    return list_of_serial_numbers

`get_results(serial_number, start_date, end_date, algorithm_name)`

Gets the results of given algorithm calculations from the /results endpoint for requested drive serial number and time range.

Parameters:

Name	Type	Description	Default
`serial_number`	`str`	Drive Serial Number	required
`start_date`	`DateType`	Start Date	required
`end_date`	`DateType`	End Date	required
`algorithm_name`	`str`	Name of the algorithm, as given in type definitions.	required

Returns:

Type	Description
`Dict`	Algorithm calculation results.

Response Example

{
  "payload": {
    "serialNumber": "2UEA902000692",
    "algorithm": "PowerTrend",
    "startDate": "2022-08-01T00:00:00+00:00",
    "endDate": "2022-08-30T00:00:00+00:00",
    "results": [
      {
        "date": "2022-08-01T07:00:00Z",
        "details": {
          "PowerTrendAmin": 0.0,
          "PowerTrendAmax": 1807.388916,
          "PowerTrendRms": 673.2872564275,
          "PowerTrendDelta": 1807.388916,
          "PowerTrendStd": 622.9944096008,
          "PowerTrendCalculate_Average": 255.3305607908
        }
      }
    ]
  },
  "success": true
}

Source code in reportconnectors/api_client/cmd_api_client.py

def get_results(self, serial_number: str, start_date: DateType, end_date: DateType, algorithm_name: str) -> Dict:
    """
    Gets the results of given algorithm calculations from the `/results` endpoint for requested
    drive serial number and time range.

    Args:
        serial_number: Drive Serial Number
        start_date: Start Date
        end_date: End Date
        algorithm_name: Name of the algorithm, as given in type definitions.

    Returns:
        Algorithm calculation results.

    Response Example:
        ```json
        {
          "payload": {
            "serialNumber": "2UEA902000692",
            "algorithm": "PowerTrend",
            "startDate": "2022-08-01T00:00:00+00:00",
            "endDate": "2022-08-30T00:00:00+00:00",
            "results": [
              {
                "date": "2022-08-01T07:00:00Z",
                "details": {
                  "PowerTrendAmin": 0.0,
                  "PowerTrendAmax": 1807.388916,
                  "PowerTrendRms": 673.2872564275,
                  "PowerTrendDelta": 1807.388916,
                  "PowerTrendStd": 622.9944096008,
                  "PowerTrendCalculate_Average": 255.3305607908
                }
              }
            ]
          },
          "success": true
        }
        ```
    """
    endpoint = self._join_url(self.report_api_suffix, "results")
    params = self._get_request_params(
        serial_number=serial_number,
        start_date=start_date,
        end_date=end_date,
        other={self.KeyNames.ALGORITHM_NAME: algorithm_name},
    )
    response = self._make_request(method="GET", endpoint=endpoint, params=params)
    decoded_response: Dict = self._decode_response(response=response, default={})
    return decoded_response

`get_scatter(serial_number, start_date, end_date, scatter_name)`

Gets the scatter data from the /scatter endpoint for requested drive serial number and time range.

Parameters:

Name	Type	Description	Default
`serial_number`	`str`	Drive Serial Number	required
`start_date`	`DateType`	Start Date	required
`end_date`	`DateType`	End Date	required
`scatter_name`	`str`	Name of the scatter (aka algorithm name in type definition).	required

Returns:

Type	Description
`Dict`	Scatter data.

Response Example

{
    "payload": {
        "serialNumber": "2UEA902000692",
        "algorithm": "Current_Volt",
        "startDate": "2022-08-01T00:00:00+00:00",
        "endDate": "2022-08-30T00:00:00+00:00",
        "scatter": [
            {"x": -5.0, "y": 14.0, "cnt": 4.0},
            {"x": -5.0, "y": 28.0, "cnt": 2.0},
            {"x": 0.0, "y": 14.0, "cnt": 8.0},
            {"x": 0.0, "y": 6609.0, "cnt": 1.0},
            {"x": 3.0, "y": 14.0, "cnt": 27.0},
        ]
    },
    "success": true
}

Source code in reportconnectors/api_client/cmd_api_client.py

def get_scatter(self, serial_number: str, start_date: DateType, end_date: DateType, scatter_name: str) -> Dict:
    """
    Gets the scatter data from the `/scatter` endpoint for requested
    drive serial number and time range.

    Args:
        serial_number: Drive Serial Number
        start_date: Start Date
        end_date: End Date
        scatter_name: Name of the scatter (aka algorithm name in type definition).

    Returns:
        Scatter data.

    Response Example:
        ```json
        {
            "payload": {
                "serialNumber": "2UEA902000692",
                "algorithm": "Current_Volt",
                "startDate": "2022-08-01T00:00:00+00:00",
                "endDate": "2022-08-30T00:00:00+00:00",
                "scatter": [
                    {"x": -5.0, "y": 14.0, "cnt": 4.0},
                    {"x": -5.0, "y": 28.0, "cnt": 2.0},
                    {"x": 0.0, "y": 14.0, "cnt": 8.0},
                    {"x": 0.0, "y": 6609.0, "cnt": 1.0},
                    {"x": 3.0, "y": 14.0, "cnt": 27.0},
                ]
            },
            "success": true
        }
        ```
    """
    endpoint = self._join_url(self.report_api_suffix, "scatter")
    params = self._get_request_params(
        serial_number=serial_number,
        start_date=start_date,
        end_date=end_date,
        other={self.KeyNames.SCATTER_NAME: scatter_name},
    )
    response = self._make_request(method="GET", endpoint=endpoint, params=params)
    decoded_response: Dict = self._decode_response(response=response, default={})
    return decoded_response

`get_trend(serial_number, start_date, end_date, trend_name)`

Gets the trend data from the /trend endpoint for requested drive serial number and time range.

Parameters:

Name	Type	Description	Default
`serial_number`	`str`	Drive Serial Number	required
`start_date`	`DateType`	Start Date	required
`end_date`	`DateType`	End Date	required
`trend_name`	`str`	Name of the trend (aka algorithm name in type definition).	required

Returns:

Type	Description
`Dict`	Trend Data.

Response Example

{
  "payload": {
    "serialNumber": "2UEA902000692",
    "algorithm": "PowerTrend",
    "startDate": "2022-08-01T00:00:00+00:00",
    "endDate": "2022-08-30T00:00:00+00:00",
    "results": [
      {
        "date": "2022-08-01T07:00:00Z",
        "details": {
          "PowerTrendAmin": 0.0,
          "PowerTrendAmax": 1807.388916,
          "PowerTrendRms": 673.2872564275,
          "PowerTrendDelta": 1807.388916,
          "PowerTrendStd": 622.9944096008,
          "PowerTrendCalculate_Average": 255.3305607908
        }
      }
    ]
  },
  "success": true
}

Source code in reportconnectors/api_client/cmd_api_client.py

def get_trend(self, serial_number: str, start_date: DateType, end_date: DateType, trend_name: str) -> Dict:
    """
    Gets the trend data from the `/trend` endpoint for requested
    drive serial number and time range.

    Args:
        serial_number: Drive Serial Number
        start_date: Start Date
        end_date: End Date
        trend_name: Name of the trend (aka algorithm name in type definition).

    Returns:
        Trend Data.

    Response Example:
        ```json
        {
          "payload": {
            "serialNumber": "2UEA902000692",
            "algorithm": "PowerTrend",
            "startDate": "2022-08-01T00:00:00+00:00",
            "endDate": "2022-08-30T00:00:00+00:00",
            "results": [
              {
                "date": "2022-08-01T07:00:00Z",
                "details": {
                  "PowerTrendAmin": 0.0,
                  "PowerTrendAmax": 1807.388916,
                  "PowerTrendRms": 673.2872564275,
                  "PowerTrendDelta": 1807.388916,
                  "PowerTrendStd": 622.9944096008,
                  "PowerTrendCalculate_Average": 255.3305607908
                }
              }
            ]
          },
          "success": true
        }
        ```
    """

    endpoint = self._join_url(self.report_api_suffix, "trend")
    params = self._get_request_params(
        serial_number=serial_number,
        start_date=start_date,
        end_date=end_date,
        other={self.KeyNames.TREND_NAME: trend_name},
    )
    response = self._make_request(method="GET", endpoint=endpoint, params=params)
    decoded_response: Dict = self._decode_response(response=response, default={})
    return decoded_response

`set_auth_token(auth_token, **kwargs)`

Sets the authorization token without the authentication process. It might be used when the authorization token is obtained from external source.

We expect this token to be Bearer token. To validate it, we use PyJWT library to decode it, and extract the expiration date

To verify if the token is valid, the validate_auth_token method is called afterward.

Parameters:

Name	Type	Description	Default
`auth_token`	`str`	Authentication token to be set.	required

Other Parameters:

Name	Type	Description
`refresh_token`	`str`	Refresh token to be set.

Returns:

Type	Description
`bool`	True if provided token is set and valid. Otherwise, False.

Source code in reportconnectors/api_client/cmd_api_client.py

def set_auth_token(self, auth_token: str, **kwargs) -> bool:
    """
    Sets the authorization token without the authentication process. It might be used when the authorization
    token is obtained from external source.

    We expect this token to be Bearer token. To validate it, we use PyJWT library to decode it, and
    extract the expiration date

    To verify if the token is valid, the `validate_auth_token` method is called afterward.

    Args:
        auth_token: Authentication token to be set.

    Keyword Args:
        refresh_token (str): Refresh token to be set.

    Returns:
        True if provided token is set and valid. Otherwise, False.

    """

    try:
        decoded_token = jwt.decode(auth_token, options={"verify_signature": False})
        _exp_timestamp = float(decoded_token.get("exp"))  # type: ignore
        expiration_date = datetime.datetime.fromtimestamp(_exp_timestamp).replace(tzinfo=datetime.timezone.utc)
    except (ValueError, TypeError, jwt.exceptions.InvalidTokenError):
        log.error("Invalid token provided.")
        return False

    # Set new tokens ...
    self.auth_data[self.KeyNames.ACCESS_TOKEN] = auth_token
    self.auth_data[self.KeyNames.REFRESH_TOKEN] = kwargs.get("refresh_token")
    self.auth_data[self.KeyNames.EXPIRATION_DATE] = expiration_date
    self.auth_data[self.KeyNames.EMAIL] = self._get_email_from_auth_token(auth_token=auth_token)
    # ... and validate them
    validation_results = self.validate_auth_token()
    return validation_results

`validate_auth_token()`

Validates current authorization token by calling the /validate endpoint.

If the token is valid then the endpoint responds with the OK status and {"success": True} response

Returns:

Type	Description
`bool`	True if the token is valid, False otherwise.

Source code in reportconnectors/api_client/cmd_api_client.py

def validate_auth_token(self) -> bool:
    """
    Validates current authorization token by calling the `/validate` endpoint.

    If the token is valid then the endpoint responds with the OK status and `{"success": True}` response

    Returns:
        True if the token is valid, False otherwise.

    """
    endpoint = self._join_url(self.report_api_suffix, "validate")
    response = self._make_request(method="GET", endpoint=endpoint)
    decoded_response: Dict = self._decode_response(response=response, default={})
    status_code = self._get_status_code(response=response)
    is_valid = (status_code == HTTPStatus.OK) and (decoded_response.get("success") is True)
    return is_valid

CMD API Client

is_logged property

authenticate(client_id, client_secret, token_url, **kwargs)

confirm_report_generation(token)

get_aggregates(serial_number, start_date, end_date, algorithm, algorithm_property, method)

get_drive_details(serial_number)

get_grouped_events(serial_number, start_date, end_date)

get_list_of_drive_details(serial_numbers, email=None)

get_list_of_serial_numbers(email=None, check_for_assigned_drives=False)

get_results(serial_number, start_date, end_date, algorithm_name)

get_scatter(serial_number, start_date, end_date, scatter_name)

get_trend(serial_number, start_date, end_date, trend_name)

set_auth_token(auth_token, **kwargs)

validate_auth_token()

`is_logged` `property`

`authenticate(client_id, client_secret, token_url, **kwargs)`

`confirm_report_generation(token)`

`get_aggregates(serial_number, start_date, end_date, algorithm, algorithm_property, method)`

`get_drive_details(serial_number)`

`get_grouped_events(serial_number, start_date, end_date)`

`get_list_of_drive_details(serial_numbers, email=None)`

`get_list_of_serial_numbers(email=None, check_for_assigned_drives=False)`

`get_results(serial_number, start_date, end_date, algorithm_name)`

`get_scatter(serial_number, start_date, end_date, scatter_name)`

`get_trend(serial_number, start_date, end_date, trend_name)`

`set_auth_token(auth_token, **kwargs)`

`validate_auth_token()`