Skip to content

Powertrain Analytics API Client

Bases: BasePowertrainAPIClient

Powertrain Analytics API Client.

Swagger Documentation

Source code in reportconnectors/api_client/powertrain/analytics/__init__.py 
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
class PowertrainAnalyticsAPIClient(BasePowertrainAPIClient):
    """
    Powertrain Analytics API Client.

    [Swagger Documentation](https://motion-pt-dev-we-analytics-api.azurewebsites.net/swagger/index.html)
    """

    _status_prefix = "/analytics"

    class KeyNames(BasePowertrainAPIClient.KeyNames):
        START_DATE = "beginTime"
        END_DATE = "endTime"
        ALGORITHM_NAMES = "algorithmNames"

    def get_report_algorithm_results(
        self,
        asset_id: int,
        start_date: DateType,
        end_date: DateType,
        algorithm_name: Union[str, Sequence[str]],
        use_all_retentions: bool = False,
    ) -> AlgorithmResultsResponse:
        """
        Gets the results of given algorithm calculations from the `/report/algorithm/results/{assetId}`
        endpoint for requested asset_id and time range.

        Args:
            asset_id: Asset identifier
            start_date: Start Date
            end_date: End Date
            algorithm_name: Name of the algorithm, as given in type definitions. It might be a single name,
                or collection of names.
            use_all_retentions: If True, it will return all data for the algorithm, otherwise only the results.
                It can be used to get the latest CI model in the response.

        Returns:
            Algorithm calculation results.
        """
        endpoint = f"api/report/Analytics/Report/Algorithm/Results/{asset_id}"
        algorithm_names = [algorithm_name] if isinstance(algorithm_name, str) else algorithm_name
        params = {
            self.KeyNames.START_DATE: start_date.strftime(self._datetime_format),
            self.KeyNames.END_DATE: end_date.strftime(self._datetime_format),
            self.KeyNames.ALGORITHM_NAMES: algorithm_names,
            "useAllRetentions": use_all_retentions,
        }
        response = self._make_request(method="GET", endpoint=endpoint, params=params)
        response_model = self._decode_response_to_model(response, AlgorithmResultsResponse)
        return response_model

    def get_report_algorithms_overview(
        self, asset_id: int, start_date: DateType, end_date: DateType
    ) -> AlgorithmOverviewResponse:
        """
        Gets the list of algorithm available for an asset identified by asset id
        The data comes from `/report/algorithm/results/overview/{assetId}` endpoint for
        requested asset_id and time range.

        Args:
            asset_id: Asset identifier
            start_date: Start Date
            end_date: End Date

        Returns:
           Dictionary with the list of algorithms available for the asset.
        """
        endpoint = f"api/report/Analytics/Report/Algorithm/Results/Overview/{asset_id}"
        params = {
            self.KeyNames.START_DATE: start_date.strftime(self._datetime_format),
            self.KeyNames.END_DATE: end_date.strftime(self._datetime_format),
        }
        response = self._make_request(method="GET", endpoint=endpoint, params=params)
        response_model = self._decode_response_to_model(response, AlgorithmOverviewResponse)
        return response_model

    def get_report_asset_conditions(
        self, asset_id: int, start_date: DateType, end_date: DateType
    ) -> AssetConditionResponse:
        """
        Gets the asset conditions for the given asset_id and time range.

        Args:
            asset_id: Asset identifier
            start_date: Start Date
            end_date: End Date

        Returns:
            Asset conditions.
        """
        endpoint = f"api/report/Analytics/Report/AssetCondition/{asset_id}"
        params = {
            self.KeyNames.START_DATE: start_date.strftime(self._datetime_format),
            self.KeyNames.END_DATE: end_date.strftime(self._datetime_format),
        }
        response = self._make_request(method="GET", endpoint=endpoint, params=params)
        response_model = self._decode_response_to_model(response, AssetConditionResponse)
        return response_model

get_report_algorithm_results(asset_id, start_date, end_date, algorithm_name, use_all_retentions=False)

Gets the results of given algorithm calculations from the /report/algorithm/results/{assetId} endpoint for requested asset_id and time range.

Parameters:

Name Type Description Default
asset_id int

Asset identifier

 required
start_date DateType

Start Date

 required
end_date DateType

End Date

 required
algorithm_name Union[str, Sequence[str]]

Name of the algorithm, as given in type definitions. It might be a single name, or collection of names.

 required
use_all_retentions bool

If True, it will return all data for the algorithm, otherwise only the results. It can be used to get the latest CI model in the response.

 False

Returns:

Type Description
AlgorithmResultsResponse

Algorithm calculation results.
Source code in reportconnectors/api_client/powertrain/analytics/__init__.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def get_report_algorithm_results(
    self,
    asset_id: int,
    start_date: DateType,
    end_date: DateType,
    algorithm_name: Union[str, Sequence[str]],
    use_all_retentions: bool = False,
) -> AlgorithmResultsResponse:
    """
    Gets the results of given algorithm calculations from the `/report/algorithm/results/{assetId}`
    endpoint for requested asset_id and time range.

    Args:
        asset_id: Asset identifier
        start_date: Start Date
        end_date: End Date
        algorithm_name: Name of the algorithm, as given in type definitions. It might be a single name,
            or collection of names.
        use_all_retentions: If True, it will return all data for the algorithm, otherwise only the results.
            It can be used to get the latest CI model in the response.

    Returns:
        Algorithm calculation results.
    """
    endpoint = f"api/report/Analytics/Report/Algorithm/Results/{asset_id}"
    algorithm_names = [algorithm_name] if isinstance(algorithm_name, str) else algorithm_name
    params = {
        self.KeyNames.START_DATE: start_date.strftime(self._datetime_format),
        self.KeyNames.END_DATE: end_date.strftime(self._datetime_format),
        self.KeyNames.ALGORITHM_NAMES: algorithm_names,
        "useAllRetentions": use_all_retentions,
    }
    response = self._make_request(method="GET", endpoint=endpoint, params=params)
    response_model = self._decode_response_to_model(response, AlgorithmResultsResponse)
    return response_model

get_report_algorithms_overview(asset_id, start_date, end_date)

Gets the list of algorithm available for an asset identified by asset id The data comes from /report/algorithm/results/overview/{assetId} endpoint for requested asset_id and time range.

Parameters:

Name Type Description Default
asset_id int

Asset identifier

 required
start_date DateType

Start Date

 required
end_date DateType

End Date

 required

Returns:

Type Description
AlgorithmOverviewResponse

Dictionary with the list of algorithms available for the asset.
Source code in reportconnectors/api_client/powertrain/analytics/__init__.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def get_report_algorithms_overview(
    self, asset_id: int, start_date: DateType, end_date: DateType
) -> AlgorithmOverviewResponse:
    """
    Gets the list of algorithm available for an asset identified by asset id
    The data comes from `/report/algorithm/results/overview/{assetId}` endpoint for
    requested asset_id and time range.

    Args:
        asset_id: Asset identifier
        start_date: Start Date
        end_date: End Date

    Returns:
       Dictionary with the list of algorithms available for the asset.
    """
    endpoint = f"api/report/Analytics/Report/Algorithm/Results/Overview/{asset_id}"
    params = {
        self.KeyNames.START_DATE: start_date.strftime(self._datetime_format),
        self.KeyNames.END_DATE: end_date.strftime(self._datetime_format),
    }
    response = self._make_request(method="GET", endpoint=endpoint, params=params)
    response_model = self._decode_response_to_model(response, AlgorithmOverviewResponse)
    return response_model

get_report_asset_conditions(asset_id, start_date, end_date)

Gets the asset conditions for the given asset_id and time range.

Parameters:

Name Type Description Default
asset_id int

Asset identifier

 required
start_date DateType

Start Date

 required
end_date DateType

End Date

 required

Returns:

Type Description
AssetConditionResponse

Asset conditions.
Source code in reportconnectors/api_client/powertrain/analytics/__init__.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def get_report_asset_conditions(
    self, asset_id: int, start_date: DateType, end_date: DateType
) -> AssetConditionResponse:
    """
    Gets the asset conditions for the given asset_id and time range.

    Args:
        asset_id: Asset identifier
        start_date: Start Date
        end_date: End Date

    Returns:
        Asset conditions.
    """
    endpoint = f"api/report/Analytics/Report/AssetCondition/{asset_id}"
    params = {
        self.KeyNames.START_DATE: start_date.strftime(self._datetime_format),
        self.KeyNames.END_DATE: end_date.strftime(self._datetime_format),
    }
    response = self._make_request(method="GET", endpoint=endpoint, params=params)
    response_model = self._decode_response_to_model(response, AssetConditionResponse)
    return response_model