Skip to content

Powertrain Timeseries API Client

Bases: BasePowertrainAPIClient

Powertrain Timeseries API Client.

Swagger Documentation

Source code in reportconnectors/api_client/powertrain/timeseries/__init__.py 
 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
108
109
110
111
112
class PowertrainTimeseriesAPIClient(BasePowertrainAPIClient):
    """
    Powertrain Timeseries API Client.

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

    _status_prefix = "/timeseries"

    def get_timeseries_config(self, asset_id: int) -> TimeseriesConfigResponse:
        """
        Gets the list of available timeseries for asset identified by `asset_id`.
        Uses `/timeseries/report/config` endpoint

        Args:
            asset_id: Asset identifier

        Returns:
            Timeseries configuration.
        """
        endpoint = "api/report/Timeseries/Report/Config"
        params = {"assetId": asset_id}
        response = self._make_request(method="GET", endpoint=endpoint, params=params)
        response_model = self._decode_response_to_model(response, TimeseriesConfigResponse)
        return response_model

    def get_timeseries_limits(self, asset_id: int) -> AssetTimeseriesLimitsResponse:
        """
        Gets the collection of timeseries limits for asset identified by `asset_id`.
        Uses `/timeseries/limits` endpoint

        Args:
            asset_id: Asset identifier

        Returns:
            Timeseries limits.
        """
        endpoint = "/timeseries/timeseries/limits"
        params = {"assetId": asset_id}
        response = self._make_request(method="GET", endpoint=endpoint, params=params)
        response_model = self._decode_response_to_model(response, AssetTimeseriesLimitsResponse)
        return response_model

    def get_aggregated_timeseries(
        self, asset_id: int, start_date: DateType, end_date: DateType, timeseries_key: Union[List[str], str]
    ) -> AggregatedTimeseriesResponse:
        """
        Gets the list of aggregated timeseries for given asset id and time range.
        Uses `/timeseries/report/aggregated` endpoint.

        The aggregation resolution depends on the requested time range.

        Args:
            asset_id: Asset identifier
            start_date: Start Date
            end_date: End
            timeseries_key: List of timeseries keys or a single timeseries key.

        Returns:
            List of aggregated timeseries.
        """
        endpoint = "/api/report/Timeseries/Report/AggregatedData"
        params = {
            "assetId": asset_id,
            "from": start_date.strftime(self._datetime_format),
            "to": end_date.strftime(self._datetime_format),
        }

        timeseries_keys = timeseries_key if isinstance(timeseries_key, list) else [timeseries_key]
        response = self._make_request(method="POST", endpoint=endpoint, params=params, json_data=timeseries_keys)
        response_model = self._decode_response_to_model(response, AggregatedTimeseriesResponse)
        return response_model

    def get_raw_timeseries(
        self, asset_id: int, start_date: DateType, end_date: DateType, timeseries_key: Union[List[str], str]
    ) -> RawTimeseriesResponse:
        """
        Gets the list of raw timeseries for given asset id and time range.
        Uses `/timeseries/report/raw` endpoint.

        The aggregation resolution depends on the requested time range.

        Args:
            asset_id: Asset identifier
            start_date: Start Date
            end_date: End Date
            timeseries_key: Timeseries key or list of timeseries keys

        Returns:
            List of aggregated timeseries.
        """
        endpoint = "/api/report/Timeseries/Report/RawData"
        params = {
            "assetId": asset_id,
            "from": start_date.strftime(self._datetime_format),
            "to": end_date.strftime(self._datetime_format),
        }
        timeseries_keys = timeseries_key if isinstance(timeseries_key, list) else [timeseries_key]
        response = self._make_request(method="POST", endpoint=endpoint, params=params, json_data=timeseries_keys)
        response_model = self._decode_response_to_model(response, RawTimeseriesResponse)
        return response_model

get_aggregated_timeseries(asset_id, start_date, end_date, timeseries_key)

Gets the list of aggregated timeseries for given asset id and time range. Uses /timeseries/report/aggregated endpoint.

The aggregation resolution depends on the requested time range.

Parameters:

Name Type Description Default
asset_id int

Asset identifier

 required
start_date DateType

Start Date

 required
end_date DateType

End

 required
timeseries_key Union[List[str], str]

List of timeseries keys or a single timeseries key.

 required

Returns:

Type Description
AggregatedTimeseriesResponse

List of aggregated timeseries.
Source code in reportconnectors/api_client/powertrain/timeseries/__init__.py 
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
def get_aggregated_timeseries(
    self, asset_id: int, start_date: DateType, end_date: DateType, timeseries_key: Union[List[str], str]
) -> AggregatedTimeseriesResponse:
    """
    Gets the list of aggregated timeseries for given asset id and time range.
    Uses `/timeseries/report/aggregated` endpoint.

    The aggregation resolution depends on the requested time range.

    Args:
        asset_id: Asset identifier
        start_date: Start Date
        end_date: End
        timeseries_key: List of timeseries keys or a single timeseries key.

    Returns:
        List of aggregated timeseries.
    """
    endpoint = "/api/report/Timeseries/Report/AggregatedData"
    params = {
        "assetId": asset_id,
        "from": start_date.strftime(self._datetime_format),
        "to": end_date.strftime(self._datetime_format),
    }

    timeseries_keys = timeseries_key if isinstance(timeseries_key, list) else [timeseries_key]
    response = self._make_request(method="POST", endpoint=endpoint, params=params, json_data=timeseries_keys)
    response_model = self._decode_response_to_model(response, AggregatedTimeseriesResponse)
    return response_model

get_raw_timeseries(asset_id, start_date, end_date, timeseries_key)

Gets the list of raw timeseries for given asset id and time range. Uses /timeseries/report/raw endpoint.

The aggregation resolution depends on the requested 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
timeseries_key Union[List[str], str]

Timeseries key or list of timeseries keys

 required

Returns:

Type Description
RawTimeseriesResponse

List of aggregated timeseries.
Source code in reportconnectors/api_client/powertrain/timeseries/__init__.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def get_raw_timeseries(
    self, asset_id: int, start_date: DateType, end_date: DateType, timeseries_key: Union[List[str], str]
) -> RawTimeseriesResponse:
    """
    Gets the list of raw timeseries for given asset id and time range.
    Uses `/timeseries/report/raw` endpoint.

    The aggregation resolution depends on the requested time range.

    Args:
        asset_id: Asset identifier
        start_date: Start Date
        end_date: End Date
        timeseries_key: Timeseries key or list of timeseries keys

    Returns:
        List of aggregated timeseries.
    """
    endpoint = "/api/report/Timeseries/Report/RawData"
    params = {
        "assetId": asset_id,
        "from": start_date.strftime(self._datetime_format),
        "to": end_date.strftime(self._datetime_format),
    }
    timeseries_keys = timeseries_key if isinstance(timeseries_key, list) else [timeseries_key]
    response = self._make_request(method="POST", endpoint=endpoint, params=params, json_data=timeseries_keys)
    response_model = self._decode_response_to_model(response, RawTimeseriesResponse)
    return response_model

get_timeseries_config(asset_id)

Gets the list of available timeseries for asset identified by asset_id. Uses /timeseries/report/config endpoint

Parameters:

Name Type Description Default
asset_id int

Asset identifier

 required

Returns:

Type Description
TimeseriesConfigResponse

Timeseries configuration.
Source code in reportconnectors/api_client/powertrain/timeseries/__init__.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
def get_timeseries_config(self, asset_id: int) -> TimeseriesConfigResponse:
    """
    Gets the list of available timeseries for asset identified by `asset_id`.
    Uses `/timeseries/report/config` endpoint

    Args:
        asset_id: Asset identifier

    Returns:
        Timeseries configuration.
    """
    endpoint = "api/report/Timeseries/Report/Config"
    params = {"assetId": asset_id}
    response = self._make_request(method="GET", endpoint=endpoint, params=params)
    response_model = self._decode_response_to_model(response, TimeseriesConfigResponse)
    return response_model

get_timeseries_limits(asset_id)

Gets the collection of timeseries limits for asset identified by asset_id. Uses /timeseries/limits endpoint

Parameters:

Name Type Description Default
asset_id int

Asset identifier

 required

Returns:

Type Description
AssetTimeseriesLimitsResponse

Timeseries limits.
Source code in reportconnectors/api_client/powertrain/timeseries/__init__.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def get_timeseries_limits(self, asset_id: int) -> AssetTimeseriesLimitsResponse:
    """
    Gets the collection of timeseries limits for asset identified by `asset_id`.
    Uses `/timeseries/limits` endpoint

    Args:
        asset_id: Asset identifier

    Returns:
        Timeseries limits.
    """
    endpoint = "/timeseries/timeseries/limits"
    params = {"assetId": asset_id}
    response = self._make_request(method="GET", endpoint=endpoint, params=params)
    response_model = self._decode_response_to_model(response, AssetTimeseriesLimitsResponse)
    return response_model