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 
 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
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
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, symbolic_name: 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 of the requested timeseries data
            end_date: End date of the requested timeseries data
            symbolic_name: Symbolic name identifying the requested timeseries.
                It can be found in the timeseries config response for the asset.

        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),
            "symbolicName": symbolic_name,
        }

        response = self._make_request(method="GET", endpoint=endpoint, params=params)
        response_model = self._decode_response_to_model(response, AggregatedTimeseriesResponse)
        return response_model

    def get_timeseries_limits_status(
        self,
        asset_ids: Union[int, List[int], None] = None,
        asset_serial_numbers: Union[str, List[str], None] = None,
    ) -> AssetLimitStatusResponse:
        """
        Gets the current limit-breach status for the requested assets.

        At least one of `asset_ids` or `asset_serial_numbers` must be provided.

        Uses the ``/timeseries/timeseries/limits/status`` endpoint.

        Args:
            asset_ids: Asset identifier or list of asset identifiers.
            asset_serial_numbers: Asset serial number or list of asset serial numbers.

        Returns:
            AssetLimitStatusResponse containing per-asset timeseries limit status data.

        Raises:
            ValueError: If neither ``asset_ids`` nor ``asset_serial_numbers`` is provided.
        """
        if asset_ids is None and asset_serial_numbers is None:
            raise ValueError("At least one of 'asset_ids' or 'asset_serial_numbers' must be provided.")

        data = {}
        if asset_ids is not None:
            if isinstance(asset_ids, int):
                data["assetIds"] = [asset_ids]
            else:
                data["assetIds"] = [int(a) for a in asset_ids]

        if asset_serial_numbers is not None:
            data["assetSerialNumber"] = (
                [asset_serial_numbers]
                if isinstance(asset_serial_numbers, int)
                else [int(a) for a in asset_serial_numbers]
            )

        endpoint = "/timeseries/timeseries/limits/status"
        response = self._make_request(method="POST", endpoint=endpoint, json_data=data)
        response_model = self._decode_response_to_model(response, AssetLimitStatusResponse)
        return response_model

    def get_raw_timeseries(
        self, asset_id: int, start_date: DateType, end_date: DateType, symbolic_name: 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 of the requested timeseries data
            end_date: End date of the requested timeseries data
            symbolic_name: Symbolic name identifying the requested timeseries.
                It can be found in the timeseries config response for the asset.

        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),
            "symbolicName": symbolic_name,
        }
        response = self._make_request(method="GET", endpoint=endpoint, params=params)
        response_model = self._decode_response_to_model(response, RawTimeseriesResponse)
        return response_model

get_aggregated_timeseries(asset_id, start_date, end_date, symbolic_name)

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 of the requested timeseries data

 required
end_date DateType

End date of the requested timeseries data

 required
symbolic_name str

Symbolic name identifying the requested timeseries. It can be found in the timeseries config response for the asset.

 required

Returns:

Type Description
AggregatedTimeseriesResponse

List of aggregated timeseries.
Source code in reportconnectors/api_client/powertrain/timeseries/__init__.py 
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
def get_aggregated_timeseries(
    self, asset_id: int, start_date: DateType, end_date: DateType, symbolic_name: 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 of the requested timeseries data
        end_date: End date of the requested timeseries data
        symbolic_name: Symbolic name identifying the requested timeseries.
            It can be found in the timeseries config response for the asset.

    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),
        "symbolicName": symbolic_name,
    }

    response = self._make_request(method="GET", endpoint=endpoint, params=params)
    response_model = self._decode_response_to_model(response, AggregatedTimeseriesResponse)
    return response_model

get_raw_timeseries(asset_id, start_date, end_date, symbolic_name)

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 of the requested timeseries data

 required
end_date DateType

End date of the requested timeseries data

 required
symbolic_name str

Symbolic name identifying the requested timeseries. It can be found in the timeseries config response for the asset.

 required

Returns:

Type Description
RawTimeseriesResponse

List of aggregated timeseries.
Source code in reportconnectors/api_client/powertrain/timeseries/__init__.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
def get_raw_timeseries(
    self, asset_id: int, start_date: DateType, end_date: DateType, symbolic_name: 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 of the requested timeseries data
        end_date: End date of the requested timeseries data
        symbolic_name: Symbolic name identifying the requested timeseries.
            It can be found in the timeseries config response for the asset.

    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),
        "symbolicName": symbolic_name,
    }
    response = self._make_request(method="GET", endpoint=endpoint, params=params)
    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 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
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 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
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

get_timeseries_limits_status(asset_ids=None, asset_serial_numbers=None)

Gets the current limit-breach status for the requested assets.

At least one of asset_ids or asset_serial_numbers must be provided.

Uses the /timeseries/timeseries/limits/status endpoint.

Parameters:

Name Type Description Default
asset_ids Union[int, List[int], None]

Asset identifier or list of asset identifiers.

 None
asset_serial_numbers Union[str, List[str], None]

Asset serial number or list of asset serial numbers.

 None

Returns:

Type Description
AssetLimitStatusResponse

AssetLimitStatusResponse containing per-asset timeseries limit status data.

Raises:

Type Description
ValueError

If neither asset_ids nor asset_serial_numbers is provided.
Source code in reportconnectors/api_client/powertrain/timeseries/__init__.py 
 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
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
def get_timeseries_limits_status(
    self,
    asset_ids: Union[int, List[int], None] = None,
    asset_serial_numbers: Union[str, List[str], None] = None,
) -> AssetLimitStatusResponse:
    """
    Gets the current limit-breach status for the requested assets.

    At least one of `asset_ids` or `asset_serial_numbers` must be provided.

    Uses the ``/timeseries/timeseries/limits/status`` endpoint.

    Args:
        asset_ids: Asset identifier or list of asset identifiers.
        asset_serial_numbers: Asset serial number or list of asset serial numbers.

    Returns:
        AssetLimitStatusResponse containing per-asset timeseries limit status data.

    Raises:
        ValueError: If neither ``asset_ids`` nor ``asset_serial_numbers`` is provided.
    """
    if asset_ids is None and asset_serial_numbers is None:
        raise ValueError("At least one of 'asset_ids' or 'asset_serial_numbers' must be provided.")

    data = {}
    if asset_ids is not None:
        if isinstance(asset_ids, int):
            data["assetIds"] = [asset_ids]
        else:
            data["assetIds"] = [int(a) for a in asset_ids]

    if asset_serial_numbers is not None:
        data["assetSerialNumber"] = (
            [asset_serial_numbers]
            if isinstance(asset_serial_numbers, int)
            else [int(a) for a in asset_serial_numbers]
        )

    endpoint = "/timeseries/timeseries/limits/status"
    response = self._make_request(method="POST", endpoint=endpoint, json_data=data)
    response_model = self._decode_response_to_model(response, AssetLimitStatusResponse)
    return response_model