Powertrain Report API Client

The Powertrain Report API Client is a Python client designed to interact with the Powertrain Report API. This client is built on top of the base Powertrain API client, which provides methods to manage authentication and other common tasks.

The Powertrain Report API Client provides methods to access all available endpoints in the Report API. This API is a collection of endpoints from other microservices that provide all data related to analytics and reporting.

The Report API should be the only one you need to interact with to get all the data you need for analytics and reporting related activities.

Bases: BasePowertrainAPIClient

Powertrain Report API Client.

Documentation

Source code in reportconnectors/api_client/powertrain/report/__init__.py

class PowertrainReportAPIClient(BasePowertrainAPIClient):
    """
    Powertrain Report API Client.

    [Documentation](https://developer.powertrain.abb.com/api-details#api=api-report-1-0)
    """

    # Organization Endpoints

    def get_assets_structure(self, asset_ids: List[int]) -> ListOfOrganizationAssetStructure:
        """
        Gets organization structure for a list of assets.

        Args:
            asset_ids: List of asset IDs

        Returns:
            Response with a list of ReportAssetStructure objects
        """

        endpoint = "/api/report/Organization/Report/Asset/Structure"
        data = {"assetIds": asset_ids}
        response = self._make_request(method="POST", endpoint=endpoint, json_data=data)
        response_model = self._decode_response_to_model(response, ListOfOrganizationAssetStructure)
        return response_model

    def search_for_assets(
        self,
        page_number: int = 1,
        page_size: int = 1000,
        refresh_data: bool = False,
        name_filter: str = "",
        organization_id: Union[int, List[int], None] = None,
        site_id: Union[int, List[int], None] = None,
        asset_type: Union[str, List[str], None] = None,
    ) -> ListOfOrganizationAssetDetails:
        """
        Gets all assets of the organizations and/or sites the user has access to.
        At least an organization or site must be specified.
        If both are specified, then the superset is returned.

        Args:
            refresh_data: If True, it gets data from external sources and updates it in the Organization
                microservice database. Default is False.
            page_number: Page number of the paged results starting from 1. Default 1
            page_size: Page size of the paged results. Default 1000
            organization_id: Organization ID or list of organization IDs for the query.
            site_id: Site ID or list of site IDs for the query.
            asset_type: Asset Type or list of asset types for the query.
            name_filter: Filter assets by their name, asset serial number.

        Returns:
            ReportAssetSearch results
        """

        def _parse_input_list(_input, _type):
            if _input is None:
                return []
            if isinstance(_input, _type):
                return [_input]
            return _input

        organization_ids = _parse_input_list(organization_id, int)
        site_ids = _parse_input_list(site_id, int)
        asset_types = _parse_input_list(asset_type, str)

        if not (organization_ids or site_ids):
            raise ValueError("At least an organization or site must be specified")

        endpoint = "/api/report/Organization/Report/Asset/Search"
        data = {
            "pageNumber": page_number,
            "pageSize": page_size,
            "refreshData": refresh_data,
            "organizationIds": organization_ids,
            "siteIds": site_ids,
            "filter": name_filter,
            "assetTypes": asset_types,
        }
        response = self._make_request(method="POST", endpoint=endpoint, json_data=data)
        response_model = self._decode_response_to_model(response, ListOfOrganizationAssetDetails)
        return response_model

    def search_for_organizations(
        self,
        refresh_data: bool = False,
        page_number: int = 1,
        page_size: int = 1000,
        name_filter: str = "",
        only_favorites: bool = False,
    ) -> ListOfOrganizationDetails:
        """
        Gets all organizations matching the specified filter that the user has access to.

        Args:
            refresh_data: If True, it gets data from external sources and updates it in the Organization
                microservice database. Default is False.
            page_number: Page number of the paged results starting from 1. Default 1
            page_size: Page size of the paged results. Default 1000
            name_filter: Filter organizations by their name.
            only_favorites: If True, only favorites are returned.

        Returns:
            ReportOrganizationSearch results
        """

        endpoint = "/api/report/Organization/Report/Organization/Search"
        data = {
            "refreshData": refresh_data,
            "pageNumber": page_number,
            "pageSize": page_size,
            "nameFilter": name_filter,
            "onlyFavorites": only_favorites,
        }
        response = self._make_request(method="POST", endpoint=endpoint, json_data=data)
        response_model = self._decode_response_to_model(response, ListOfOrganizationDetails)
        return response_model

    def search_for_sites(
        self,
        organization_id: Union[int, List[int]],
        refresh_data: bool = False,
        page_number: int = 1,
        page_size: int = 1000,
    ) -> ListOfOrganizationSiteDetails:
        """
        Get all sites for the specified organizations.

        Args:
            refresh_data: If True, it gets data from external sources and updates it in the Organization
                microservice database. Default is False.
            page_number: Page number of the paged results starting from 1. Default 1
            page_size: Page size of the paged results. Default 1000
            organization_id: Organization ID or list of organization IDs.
        Returns:
            ListOfReportSites
        """

        endpoint = "/api/report/Organization/Report/Site/Search"
        organization_ids = organization_id if isinstance(organization_id, list) else [organization_id]
        data = {
            "refreshData": refresh_data,
            "pageNumber": page_number,
            "pageSize": page_size,
            "organizationsIds": organization_ids,
        }
        response = self._make_request(method="POST", endpoint=endpoint, json_data=data)
        response_model = self._decode_response_to_model(response, ListOfOrganizationSiteDetails)
        return response_model

    # Asset Endpoints

    def get_asset_details(self, asset_id: Union[int, List[int]]) -> ListOfAssetDetails:
        """
        Get asset details which include master data and asset properties

        Args:
            asset_id: Asset ID or list of Asset IDs.

        Returns:

            List of ReportAssetDetail objects
        """
        endpoint = "/api/report/Asset/Report/Asset/Details"
        asset_ids = asset_id if isinstance(asset_id, list) else [asset_id]

        data = {"assetIds": asset_ids}
        response = self._make_request(method="POST", endpoint=endpoint, json_data=data)
        response_model = self._decode_response_to_model(response, ListOfAssetDetails)
        return response_model

    # Subscription endpoints

    def get_capabilities(
        self,
        asset_id: Union[List[int], int],
        capability_code: Union[str, List[str]],
        organization_id: Optional[int] = None,
        check_subscription: bool = False,
    ) -> AssetCapabilitiesInfoResponse:
        """
        Gets the list of capabilities for specified assets and organization


        Args:
            organization_id: Organization Identifier
            asset_id: Asset identifier or list of asset identifiers.
            capability_code: Capability code or list of capability codes to filter by
            check_subscription: If True, it will check if the user has a subscription for the capability.
        """

        asset_ids = asset_id if isinstance(asset_id, list) else [asset_id]
        capability_codes = capability_code if isinstance(capability_code, list) else [capability_code]
        endpoint = "/api/report/Subscription/Report/Capability/Search"

        data: Dict[str, Union[int, bool, str, List]] = {
            "assetIds": asset_ids,
            "checkSubscription": check_subscription,
            "capabilityCodes": capability_codes,
        }
        if organization_id is not None:
            data["organizationId"] = organization_id

        response = self._make_request(method="POST", endpoint=endpoint, json_data=data)
        response_model = self._decode_response_to_model(response, AssetCapabilitiesInfoResponse)
        return response_model

    # Analytic Endpoints

    def get_analytic_algorithms_overview(
        self, asset_id: int, start_date: DateType, end_date: DateType
    ) -> AnalyticsAlgorithmsOverview:
        """
        Gets the overview list of analytics algorithms for the given asset.

        Args:
            asset_id: Asset ID
            start_date: Start date
            end_date: End date

        Returns:
            Analytics Algorithm overview
        """
        endpoint = f"/api/report/Analytics/Report/Algorithm/Results/Overview/{asset_id}"

        params = {
            "beginTime": start_date.strftime(self._datetime_format),
            "endTime": end_date.strftime(self._datetime_format),
        }

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

    def get_analytic_algorithm_results(
        self,
        asset_id: int,
        start_date: DateType,
        end_date: DateType,
        algorithm_name: Union[str, Sequence[str]],
        use_all_retentions: bool = False,
    ) -> AnalyticsAlgorithmResults:
        """
        Gets the algorithm(s) results for a given asset in the provided 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 results data model.
        """
        endpoint = f"/api/report/Analytics/Report/Algorithm/Results/{asset_id}"
        algorithm_names = [algorithm_name] if isinstance(algorithm_name, str) else algorithm_name
        params = {
            "beginTime": start_date.strftime(self._datetime_format),
            "endTime": end_date.strftime(self._datetime_format),
            "algorithmNames": algorithm_names,
            "useAllRetentions": use_all_retentions,
        }
        response = self._make_request(method="GET", endpoint=endpoint, params=params)
        response_model = self._decode_response_to_model(response, AnalyticsAlgorithmResults)
        return response_model

    def get_condition_index_results(
        self, asset_id: int, start_date: DateType, end_date: DateType
    ) -> AnalyticsConditionDetails:
        """
        Gets the condition index results for a given asset in the provided timerange.

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

        Returns:
            List of ConditionDetails results
        """

        endpoint = f"/api/report/Analytics/Report/AssetCondition/{asset_id}"
        params = {
            "beginTime": start_date.strftime(self._datetime_format),
            "endTime": end_date.strftime(self._datetime_format),
        }
        response = self._make_request(method="GET", endpoint=endpoint, params=params)
        response_model = self._decode_response_to_model(response, AnalyticsConditionDetails)
        return response_model

    # Timeseries Endpoints

    def get_timeseries_config(self, asset_id: int) -> ListOfTimeseriesConfigs:
        """
        Gets the timeseries config for a given asset

        Args:
            asset_id: Asset ID

        Returns:
            Timeseries config
        """

        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, ListOfTimeseriesConfigs)
        return response_model

    def get_aggregated_timeseries_results(
        self,
        asset_id: int,
        start_date: DateType,
        end_date: DateType,
        symbolic_name: Union[str, List[str]],
    ) -> ListOfTimeseriesAggregatedResults:
        """
        Gets the list of aggregated timeseries for given asset id and time range.

        The aggregation resolution depends on the requested time range.

        Args:
            asset_id: Asset identifier
            start_date: Start Date
            end_date: End
            symbolic_name: List of symbolic names or a single symbolic name.

        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),
        }

        symbolic_names = symbolic_name if isinstance(symbolic_name, list) else [symbolic_name]
        response = self._make_request(method="POST", endpoint=endpoint, json_data=symbolic_names, params=params)
        response_model = self._decode_response_to_model(response, ListOfTimeseriesAggregatedResults)
        return response_model

    def get_full_timeseries_results(
        self,
        asset_id: int,
        start_date: DateType,
        end_date: DateType,
        symbolic_name: Union[str, List[str]],
    ) -> ListOfTimeseriesFullResults:
        """
        Gets the list of raw timeseries for given asset id and time range.

        Note:
            Maximum timerange you can ask for in a single request depends on the asset type.
            It's 30 days for SmartSensor assets, and 1 day for Drive assets.
            If you want to overcome that, it has to be done above this client.

        Args:
            asset_id: Asset identifier
            start_date: Start Date
            end_date: End
            symbolic_name: List of symbolic names or a single symbolic name.

        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),
        }
        symbolic_names = symbolic_name if isinstance(symbolic_name, list) else [symbolic_name]
        response = self._make_request(method="POST", endpoint=endpoint, json_data=symbolic_names, params=params)
        response_model = self._decode_response_to_model(response, ListOfTimeseriesFullResults)
        return response_model

    # Type Config endpoints

    def get_bearing_info(self, bearing: str, manufacturer: Optional[str] = None) -> ListOfBearingInfos:
        """
        Gets the bearing info based on the provided bearing type and optional manufacturer.

        Args:
            bearing: Bearing identifier
            manufacturer: Manufacturer name. If not provided, then the all matching results are returned.

        Returns:
            List of matching bearing info data
        """

        endpoint = f"/api/report/TypeConfig/Report/AssetType/BearingInfo/{bearing}"
        params = {"manufacturer": manufacturer} if manufacturer else None

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

    def get_unit_conversions(self) -> ListOfUnitConversions:
        """
        Gets all unit conversion data in a single response.

        Returns:
            All unit conversion data in a single response.
        """

        endpoint = "/api/report/TypeConfig/Report/TypeConfig/UnitConversions"
        response = self._make_request(method="GET", endpoint=endpoint)
        response_model = self._decode_response_to_model(response, ListOfUnitConversions)
        return response_model

    # Events Endpoints

    def search_for_events(
        self,
        asset_id: Union[int, Sequence[int]],
        start_date: DateType,
        end_date: DateType,
        use_cursor: bool = False,
        page_size: int = 100,
        **kwargs,
    ) -> EventsResponse:
        """
        Returns the events for the given asset ids and date range.

        It's based on the `Event/Search` endpoint.

        It also allows filtering events by severity, status, cause, existence of comments, binary logger,
        data loggers, parameter backup, and message text.

        Results are paginated with a default page size of 100.
        However, if you set `use_cursor=True`, it will fetch all the events,
        by using the cursor returned in the response.

        Args:
            asset_id: Asset identifier or list of asset identifiers.
            start_date: Start date of the search.
            end_date: End date of the search.
            use_cursor: If True, it will fetch all the events using the cursor returned in the response. Otherwise,
                it will fetch only the first page with the results. Default: `False`.
            page_size: Number of events per page. Default: `100`.

        Keyword Args:
            severity_codes (List[str]): List of severity codes to filter the events
            status_codes (List[str]): List of status codes to filter the events
            cause_codes (List[str]): List of cause codes to filter the events
            has_comments (bool): If True, it will filter the events that have comments
            has_binary_logger (bool): If True, it will filter the events that have binary logger
            has_data_loggers (bool): If True, it will filter the events that have data loggers
            has_parameter_backup (bool): If True, it will filter the events that have parameter backup
            message_text_contains (str): Text to search in the message text
            cursor (str): Cursor that points to a specific results page

        Returns:
            EventSearchResponse object with the list of events and the next cursor.
        """
        asset_ids = asset_id if isinstance(asset_id, list) else [asset_id]
        data = {
            "assetIds": asset_ids,
            "timestampFrom": start_date.strftime(self._datetime_format),
            "timestampTo": end_date.strftime(self._datetime_format),
            "pageSize": page_size,
        }
        # Since API expect extra filtering parameters in camelCase and the kwargs are provided in the snake_case,
        # we need to convert the keys to camelCase and update the data payload
        converted_kwargs = {to_camel(k): v for k, v in kwargs.items()}
        data.update(converted_kwargs)

        endpoint = "/api/report/Report/Event/Search"
        response = self._make_request(method="POST", endpoint=endpoint, json_data=data)
        response_model = self._decode_response_to_model(response, EventsResponse)

        # If the `use_cursor` is True and the first response has the `next_cursor`,
        # we request the next page until the cursor is None.
        # Results is a merged response model.
        while use_cursor and response_model.next_cursor is not None:
            data["cursor"] = response_model.next_cursor
            next_response = self._make_request(method="POST", endpoint=endpoint, json_data=data)
            next_response_model = self._decode_response_to_model(next_response, EventsResponse)
            response_model.events += next_response_model.events
            response_model.next_cursor = next_response_model.next_cursor
        return response_model

    def get_event_comments(self, event_id: int) -> ListOfEventComments:
        """
        Gets the event-related comments.

        Args:
            event_id: Identifier of the event

        Returns:
            Asset properties.
        """

        endpoint = f"/api/report/Report/Event/{event_id}/Comments"
        response = self._make_request(method="GET", endpoint=endpoint)
        response_model = self._decode_response_to_model(response, ListOfEventComments)
        return response_model

    # FileService endpoints

    def get_file_attributes(
        self, asset_id: Union[int, List[int]], start_date: DateType, end_date: DateType
    ) -> ListOfFileAttributes:
        """
        Gets the list of raw data file attributes that describe the raw data files available for a given asset(s)
        in the provided time range.

        It uses the `/Report/Asset/RawData` endpoint for the requested asset_id and time range.
        The returned raw data file objects don't contain the file content, but only the download URL.

        If you want to download the file content, you have two options:

        1. Run the `.get_file()` method on the `RawDataFileAttributes` response element.
        2. Use the client method `.download_raw_data_file(file_id)` to get the file content and name.

        Args:
            asset_id: Asset identifier or list of asset identifiers
            start_date: Start Date
            end_date: End Date

        Returns:
            Response object that contains the list of the found raw data files attributes.
        """
        endpoint = "/api/report/FileService/Report/Asset/RawData"

        asset_ids = asset_id if isinstance(asset_id, list) else [asset_id]

        data = {
            "assetIds": asset_ids,
            "startTime": start_date.strftime(self._datetime_format),
            "endTime": end_date.strftime(self._datetime_format),
        }
        response = self._make_request(method="POST", endpoint=endpoint, json_data=data)
        response_model = self._decode_response_to_model(response, ListOfFileAttributes)
        return response_model

    def download_file_from_url(self, url: str, file_name: Optional[str] = None) -> FileRawData:
        """
        Downloads the raw data file content directly from the provided URL.

        The URL is a part of the response from the `get_raw_data_files_attributes` method.

        Args:
            url: The URL to download the raw data file content.
            file_name: Optional file name to use for the downloaded file.
                If not provided, the file name from the URL is used.

        Returns:
            CsvFile object with the raw data file content and name.
        """
        response = self._make_request(method="GET", url=url, add_authorization=False)
        decoded_response = response.text if response else ""

        if file_name is None:
            file_name = extract_file_name_from_url(url, default="RawDataFile.csv")
        csv_file = FileRawData(content=decoded_response, name=file_name)
        return csv_file

    # Sensor Endpoints

    def get_sensor_details(self, sensor_id: Union[int, List[int]]) -> ListOfSensorDetails:
        """
        Gets the sensor details based on the provided sensor identifier(s).

        Args:
            sensor_id: Sensor identifier or list of sensor identifiers.

        Returns:
            List of the Sensor Details
        """
        endpoint = "/api/report/Sensor/Report/Details"
        sensor_ids = sensor_id if isinstance(sensor_id, list) else [sensor_id]

        data = {"sensorIds": sensor_ids}
        response = self._make_request(method="POST", endpoint=endpoint, json_data=data)
        response_model = self._decode_response_to_model(response, ListOfSensorDetails)
        return response_model

    def get_asset_group(
        self,
        name_filter: Optional[str] = "",
        site_ids: Optional[List[int]] = None,
        organizations_ids: Optional[List[int]] = None,
        asset_group_ids: Optional[List[int]] = None,
        countries: Optional[List[str]] = None,
        country_codes: Optional[List[str]] = None,
        fleets_only: Optional[bool] = False,
        include_assets: Optional[bool] = False,
        refresh_data: Optional[bool] = False,
    ) -> ListReportAssetGroups:
        """Gets the asset group list based on the provided filters.
        Args:
            name_filter: Filter asset groups by their name.
            site_ids: List of site IDs to filter by.
            organizations_ids: List of organization IDs to filter by.
            asset_group_ids: List of asset group IDs to filter by.
            countries: List of country names to filter by.
            country_codes: List of country codes to filter by.
            fleets_only: If True, only fleets are returned.
            include_assets: If True, assets are included in the response.
            refresh_data: If True, it gets data from external sources and updates it in the pt api
        Returns:
            ListReportAssetGroups
        """
        data = {"fleetsOnly": fleets_only, "includeAssets": include_assets}
        query = ReportGetAssetGroupsQuery.model_validate(data)
        if refresh_data:
            query.refresh_data = refresh_data
        if name_filter:
            query.name_filter = name_filter
        if site_ids:
            query.site_ids = site_ids
        if organizations_ids:
            query.organizations_ids = organizations_ids
        if asset_group_ids:
            query.asset_group_ids = asset_group_ids
        if countries:
            query.countries = countries
        if country_codes:
            query.country_codes = country_codes

        endpoint = "/api/report/Organization/Report/AssetGroup/Search"
        data = query.model_dump(by_alias=True)
        response = self._make_request(method="POST", endpoint=endpoint, json_data=data)
        response_model = self._decode_response_to_model(response, ListReportAssetGroups)
        return response_model

download_file_from_url(url, file_name=None)

Downloads the raw data file content directly from the provided URL.

The URL is a part of the response from the get_raw_data_files_attributes method.

Parameters:

Name	Type	Description	Default
url	str	The URL to download the raw data file content.	required
file_name	Optional[str]	Optional file name to use for the downloaded file. If not provided, the file name from the URL is used.	None

Returns:

Type	Description
FileRawData	CsvFile object with the raw data file content and name.

Source code in reportconnectors/api_client/powertrain/report/__init__.py

def download_file_from_url(self, url: str, file_name: Optional[str] = None) -> FileRawData:
    """
    Downloads the raw data file content directly from the provided URL.

    The URL is a part of the response from the `get_raw_data_files_attributes` method.

    Args:
        url: The URL to download the raw data file content.
        file_name: Optional file name to use for the downloaded file.
            If not provided, the file name from the URL is used.

    Returns:
        CsvFile object with the raw data file content and name.
    """
    response = self._make_request(method="GET", url=url, add_authorization=False)
    decoded_response = response.text if response else ""

    if file_name is None:
        file_name = extract_file_name_from_url(url, default="RawDataFile.csv")
    csv_file = FileRawData(content=decoded_response, name=file_name)
    return csv_file

get_aggregated_timeseries_results(asset_id, start_date, end_date, symbolic_name)

Gets the list of aggregated timeseries for given asset id and time range.

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
symbolic_name	Union[str, List[str]]	List of symbolic names or a single symbolic name.	required

Returns:

Type	Description
ListOfTimeseriesAggregatedResults	List of aggregated timeseries.

Source code in reportconnectors/api_client/powertrain/report/__init__.py

def get_aggregated_timeseries_results(
    self,
    asset_id: int,
    start_date: DateType,
    end_date: DateType,
    symbolic_name: Union[str, List[str]],
) -> ListOfTimeseriesAggregatedResults:
    """
    Gets the list of aggregated timeseries for given asset id and time range.

    The aggregation resolution depends on the requested time range.

    Args:
        asset_id: Asset identifier
        start_date: Start Date
        end_date: End
        symbolic_name: List of symbolic names or a single symbolic name.

    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),
    }

    symbolic_names = symbolic_name if isinstance(symbolic_name, list) else [symbolic_name]
    response = self._make_request(method="POST", endpoint=endpoint, json_data=symbolic_names, params=params)
    response_model = self._decode_response_to_model(response, ListOfTimeseriesAggregatedResults)
    return response_model

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

Gets the algorithm(s) results for a given asset in the provided 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
AnalyticsAlgorithmResults	Algorithm results data model.

Source code in reportconnectors/api_client/powertrain/report/__init__.py

def get_analytic_algorithm_results(
    self,
    asset_id: int,
    start_date: DateType,
    end_date: DateType,
    algorithm_name: Union[str, Sequence[str]],
    use_all_retentions: bool = False,
) -> AnalyticsAlgorithmResults:
    """
    Gets the algorithm(s) results for a given asset in the provided 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 results data model.
    """
    endpoint = f"/api/report/Analytics/Report/Algorithm/Results/{asset_id}"
    algorithm_names = [algorithm_name] if isinstance(algorithm_name, str) else algorithm_name
    params = {
        "beginTime": start_date.strftime(self._datetime_format),
        "endTime": end_date.strftime(self._datetime_format),
        "algorithmNames": algorithm_names,
        "useAllRetentions": use_all_retentions,
    }
    response = self._make_request(method="GET", endpoint=endpoint, params=params)
    response_model = self._decode_response_to_model(response, AnalyticsAlgorithmResults)
    return response_model

get_analytic_algorithms_overview(asset_id, start_date, end_date)

Gets the overview list of analytics algorithms for the given asset.

Parameters:

Name	Type	Description	Default
asset_id	int	Asset ID	required
start_date	DateType	Start date	required
end_date	DateType	End date	required

Returns:

Type	Description
AnalyticsAlgorithmsOverview	Analytics Algorithm overview

Source code in reportconnectors/api_client/powertrain/report/__init__.py

def get_analytic_algorithms_overview(
    self, asset_id: int, start_date: DateType, end_date: DateType
) -> AnalyticsAlgorithmsOverview:
    """
    Gets the overview list of analytics algorithms for the given asset.

    Args:
        asset_id: Asset ID
        start_date: Start date
        end_date: End date

    Returns:
        Analytics Algorithm overview
    """
    endpoint = f"/api/report/Analytics/Report/Algorithm/Results/Overview/{asset_id}"

    params = {
        "beginTime": start_date.strftime(self._datetime_format),
        "endTime": end_date.strftime(self._datetime_format),
    }

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

get_asset_details(asset_id)

Get asset details which include master data and asset properties

Parameters:

Name	Type	Description	Default
asset_id	Union[int, List[int]]	Asset ID or list of Asset IDs.	required

Returns:

List of ReportAssetDetail objects

Source code in reportconnectors/api_client/powertrain/report/__init__.py

def get_asset_details(self, asset_id: Union[int, List[int]]) -> ListOfAssetDetails:
    """
    Get asset details which include master data and asset properties

    Args:
        asset_id: Asset ID or list of Asset IDs.

    Returns:

        List of ReportAssetDetail objects
    """
    endpoint = "/api/report/Asset/Report/Asset/Details"
    asset_ids = asset_id if isinstance(asset_id, list) else [asset_id]

    data = {"assetIds": asset_ids}
    response = self._make_request(method="POST", endpoint=endpoint, json_data=data)
    response_model = self._decode_response_to_model(response, ListOfAssetDetails)
    return response_model

get_asset_group(name_filter='', site_ids=None, organizations_ids=None, asset_group_ids=None, countries=None, country_codes=None, fleets_only=False, include_assets=False, refresh_data=False)

Gets the asset group list based on the provided filters. Args: name_filter: Filter asset groups by their name. site_ids: List of site IDs to filter by. organizations_ids: List of organization IDs to filter by. asset_group_ids: List of asset group IDs to filter by. countries: List of country names to filter by. country_codes: List of country codes to filter by. fleets_only: If True, only fleets are returned. include_assets: If True, assets are included in the response. refresh_data: If True, it gets data from external sources and updates it in the pt api Returns: ListReportAssetGroups

Source code in reportconnectors/api_client/powertrain/report/__init__.py

def get_asset_group(
    self,
    name_filter: Optional[str] = "",
    site_ids: Optional[List[int]] = None,
    organizations_ids: Optional[List[int]] = None,
    asset_group_ids: Optional[List[int]] = None,
    countries: Optional[List[str]] = None,
    country_codes: Optional[List[str]] = None,
    fleets_only: Optional[bool] = False,
    include_assets: Optional[bool] = False,
    refresh_data: Optional[bool] = False,
) -> ListReportAssetGroups:
    """Gets the asset group list based on the provided filters.
    Args:
        name_filter: Filter asset groups by their name.
        site_ids: List of site IDs to filter by.
        organizations_ids: List of organization IDs to filter by.
        asset_group_ids: List of asset group IDs to filter by.
        countries: List of country names to filter by.
        country_codes: List of country codes to filter by.
        fleets_only: If True, only fleets are returned.
        include_assets: If True, assets are included in the response.
        refresh_data: If True, it gets data from external sources and updates it in the pt api
    Returns:
        ListReportAssetGroups
    """
    data = {"fleetsOnly": fleets_only, "includeAssets": include_assets}
    query = ReportGetAssetGroupsQuery.model_validate(data)
    if refresh_data:
        query.refresh_data = refresh_data
    if name_filter:
        query.name_filter = name_filter
    if site_ids:
        query.site_ids = site_ids
    if organizations_ids:
        query.organizations_ids = organizations_ids
    if asset_group_ids:
        query.asset_group_ids = asset_group_ids
    if countries:
        query.countries = countries
    if country_codes:
        query.country_codes = country_codes

    endpoint = "/api/report/Organization/Report/AssetGroup/Search"
    data = query.model_dump(by_alias=True)
    response = self._make_request(method="POST", endpoint=endpoint, json_data=data)
    response_model = self._decode_response_to_model(response, ListReportAssetGroups)
    return response_model

get_assets_structure(asset_ids)

Gets organization structure for a list of assets.

Parameters:

Name	Type	Description	Default
asset_ids	List[int]	List of asset IDs	required

Returns:

Type	Description
ListOfOrganizationAssetStructure	Response with a list of ReportAssetStructure objects

Source code in reportconnectors/api_client/powertrain/report/__init__.py

def get_assets_structure(self, asset_ids: List[int]) -> ListOfOrganizationAssetStructure:
    """
    Gets organization structure for a list of assets.

    Args:
        asset_ids: List of asset IDs

    Returns:
        Response with a list of ReportAssetStructure objects
    """

    endpoint = "/api/report/Organization/Report/Asset/Structure"
    data = {"assetIds": asset_ids}
    response = self._make_request(method="POST", endpoint=endpoint, json_data=data)
    response_model = self._decode_response_to_model(response, ListOfOrganizationAssetStructure)
    return response_model

get_bearing_info(bearing, manufacturer=None)

Gets the bearing info based on the provided bearing type and optional manufacturer.

Parameters:

Name	Type	Description	Default
bearing	str	Bearing identifier	required
manufacturer	Optional[str]	Manufacturer name. If not provided, then the all matching results are returned.	None

Returns:

Type	Description
ListOfBearingInfos	List of matching bearing info data

Source code in reportconnectors/api_client/powertrain/report/__init__.py

def get_bearing_info(self, bearing: str, manufacturer: Optional[str] = None) -> ListOfBearingInfos:
    """
    Gets the bearing info based on the provided bearing type and optional manufacturer.

    Args:
        bearing: Bearing identifier
        manufacturer: Manufacturer name. If not provided, then the all matching results are returned.

    Returns:
        List of matching bearing info data
    """

    endpoint = f"/api/report/TypeConfig/Report/AssetType/BearingInfo/{bearing}"
    params = {"manufacturer": manufacturer} if manufacturer else None

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

get_capabilities(asset_id, capability_code, organization_id=None, check_subscription=False)

Gets the list of capabilities for specified assets and organization

Parameters:

Name	Type	Description	Default
organization_id	Optional[int]	Organization Identifier	None
asset_id	Union[List[int], int]	Asset identifier or list of asset identifiers.	required
capability_code	Union[str, List[str]]	Capability code or list of capability codes to filter by	required
check_subscription	bool	If True, it will check if the user has a subscription for the capability.	False

Source code in reportconnectors/api_client/powertrain/report/__init__.py

def get_capabilities(
    self,
    asset_id: Union[List[int], int],
    capability_code: Union[str, List[str]],
    organization_id: Optional[int] = None,
    check_subscription: bool = False,
) -> AssetCapabilitiesInfoResponse:
    """
    Gets the list of capabilities for specified assets and organization


    Args:
        organization_id: Organization Identifier
        asset_id: Asset identifier or list of asset identifiers.
        capability_code: Capability code or list of capability codes to filter by
        check_subscription: If True, it will check if the user has a subscription for the capability.
    """

    asset_ids = asset_id if isinstance(asset_id, list) else [asset_id]
    capability_codes = capability_code if isinstance(capability_code, list) else [capability_code]
    endpoint = "/api/report/Subscription/Report/Capability/Search"

    data: Dict[str, Union[int, bool, str, List]] = {
        "assetIds": asset_ids,
        "checkSubscription": check_subscription,
        "capabilityCodes": capability_codes,
    }
    if organization_id is not None:
        data["organizationId"] = organization_id

    response = self._make_request(method="POST", endpoint=endpoint, json_data=data)
    response_model = self._decode_response_to_model(response, AssetCapabilitiesInfoResponse)
    return response_model

get_condition_index_results(asset_id, start_date, end_date)

Gets the condition index results for a given asset in the provided timerange.

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
AnalyticsConditionDetails	List of ConditionDetails results

Source code in reportconnectors/api_client/powertrain/report/__init__.py

def get_condition_index_results(
    self, asset_id: int, start_date: DateType, end_date: DateType
) -> AnalyticsConditionDetails:
    """
    Gets the condition index results for a given asset in the provided timerange.

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

    Returns:
        List of ConditionDetails results
    """

    endpoint = f"/api/report/Analytics/Report/AssetCondition/{asset_id}"
    params = {
        "beginTime": start_date.strftime(self._datetime_format),
        "endTime": end_date.strftime(self._datetime_format),
    }
    response = self._make_request(method="GET", endpoint=endpoint, params=params)
    response_model = self._decode_response_to_model(response, AnalyticsConditionDetails)
    return response_model

get_event_comments(event_id)

Gets the event-related comments.

Parameters:

Name	Type	Description	Default
event_id	int	Identifier of the event	required

Returns:

Type	Description
ListOfEventComments	Asset properties.

Source code in reportconnectors/api_client/powertrain/report/__init__.py

def get_event_comments(self, event_id: int) -> ListOfEventComments:
    """
    Gets the event-related comments.

    Args:
        event_id: Identifier of the event

    Returns:
        Asset properties.
    """

    endpoint = f"/api/report/Report/Event/{event_id}/Comments"
    response = self._make_request(method="GET", endpoint=endpoint)
    response_model = self._decode_response_to_model(response, ListOfEventComments)
    return response_model

get_file_attributes(asset_id, start_date, end_date)

Gets the list of raw data file attributes that describe the raw data files available for a given asset(s) in the provided time range.

It uses the /Report/Asset/RawData endpoint for the requested asset_id and time range. The returned raw data file objects don't contain the file content, but only the download URL.

If you want to download the file content, you have two options:

1. Run the .get_file() method on the RawDataFileAttributes response element.
2. Use the client method .download_raw_data_file(file_id) to get the file content and name.

Parameters:

Name	Type	Description	Default
asset_id	Union[int, List[int]]	Asset identifier or list of asset identifiers	required
start_date	DateType	Start Date	required
end_date	DateType	End Date	required

Returns:

Type	Description
ListOfFileAttributes	Response object that contains the list of the found raw data files attributes.

Source code in reportconnectors/api_client/powertrain/report/__init__.py

def get_file_attributes(
    self, asset_id: Union[int, List[int]], start_date: DateType, end_date: DateType
) -> ListOfFileAttributes:
    """
    Gets the list of raw data file attributes that describe the raw data files available for a given asset(s)
    in the provided time range.

    It uses the `/Report/Asset/RawData` endpoint for the requested asset_id and time range.
    The returned raw data file objects don't contain the file content, but only the download URL.

    If you want to download the file content, you have two options:

    1. Run the `.get_file()` method on the `RawDataFileAttributes` response element.
    2. Use the client method `.download_raw_data_file(file_id)` to get the file content and name.

    Args:
        asset_id: Asset identifier or list of asset identifiers
        start_date: Start Date
        end_date: End Date

    Returns:
        Response object that contains the list of the found raw data files attributes.
    """
    endpoint = "/api/report/FileService/Report/Asset/RawData"

    asset_ids = asset_id if isinstance(asset_id, list) else [asset_id]

    data = {
        "assetIds": asset_ids,
        "startTime": start_date.strftime(self._datetime_format),
        "endTime": end_date.strftime(self._datetime_format),
    }
    response = self._make_request(method="POST", endpoint=endpoint, json_data=data)
    response_model = self._decode_response_to_model(response, ListOfFileAttributes)
    return response_model

get_full_timeseries_results(asset_id, start_date, end_date, symbolic_name)

Gets the list of raw timeseries for given asset id and time range.

Note

Maximum timerange you can ask for in a single request depends on the asset type. It's 30 days for SmartSensor assets, and 1 day for Drive assets. If you want to overcome that, it has to be done above this client.

Parameters:

Name	Type	Description	Default
asset_id	int	Asset identifier	required
start_date	DateType	Start Date	required
end_date	DateType	End	required
symbolic_name	Union[str, List[str]]	List of symbolic names or a single symbolic name.	required

Returns:

Type	Description
ListOfTimeseriesFullResults	List of aggregated timeseries.

Source code in reportconnectors/api_client/powertrain/report/__init__.py

def get_full_timeseries_results(
    self,
    asset_id: int,
    start_date: DateType,
    end_date: DateType,
    symbolic_name: Union[str, List[str]],
) -> ListOfTimeseriesFullResults:
    """
    Gets the list of raw timeseries for given asset id and time range.

    Note:
        Maximum timerange you can ask for in a single request depends on the asset type.
        It's 30 days for SmartSensor assets, and 1 day for Drive assets.
        If you want to overcome that, it has to be done above this client.

    Args:
        asset_id: Asset identifier
        start_date: Start Date
        end_date: End
        symbolic_name: List of symbolic names or a single symbolic name.

    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),
    }
    symbolic_names = symbolic_name if isinstance(symbolic_name, list) else [symbolic_name]
    response = self._make_request(method="POST", endpoint=endpoint, json_data=symbolic_names, params=params)
    response_model = self._decode_response_to_model(response, ListOfTimeseriesFullResults)
    return response_model

get_sensor_details(sensor_id)

Gets the sensor details based on the provided sensor identifier(s).

Parameters:

Name	Type	Description	Default
sensor_id	Union[int, List[int]]	Sensor identifier or list of sensor identifiers.	required

Returns:

Type	Description
ListOfSensorDetails	List of the Sensor Details

Source code in reportconnectors/api_client/powertrain/report/__init__.py

def get_sensor_details(self, sensor_id: Union[int, List[int]]) -> ListOfSensorDetails:
    """
    Gets the sensor details based on the provided sensor identifier(s).

    Args:
        sensor_id: Sensor identifier or list of sensor identifiers.

    Returns:
        List of the Sensor Details
    """
    endpoint = "/api/report/Sensor/Report/Details"
    sensor_ids = sensor_id if isinstance(sensor_id, list) else [sensor_id]

    data = {"sensorIds": sensor_ids}
    response = self._make_request(method="POST", endpoint=endpoint, json_data=data)
    response_model = self._decode_response_to_model(response, ListOfSensorDetails)
    return response_model

get_timeseries_config(asset_id)

Gets the timeseries config for a given asset

Parameters:

Name	Type	Description	Default
asset_id	int	Asset ID	required

Returns:

Type	Description
ListOfTimeseriesConfigs	Timeseries config

Source code in reportconnectors/api_client/powertrain/report/__init__.py

def get_timeseries_config(self, asset_id: int) -> ListOfTimeseriesConfigs:
    """
    Gets the timeseries config for a given asset

    Args:
        asset_id: Asset ID

    Returns:
        Timeseries config
    """

    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, ListOfTimeseriesConfigs)
    return response_model

get_unit_conversions()

Gets all unit conversion data in a single response.

Returns:

Type	Description
ListOfUnitConversions	All unit conversion data in a single response.

Source code in reportconnectors/api_client/powertrain/report/__init__.py

def get_unit_conversions(self) -> ListOfUnitConversions:
    """
    Gets all unit conversion data in a single response.

    Returns:
        All unit conversion data in a single response.
    """

    endpoint = "/api/report/TypeConfig/Report/TypeConfig/UnitConversions"
    response = self._make_request(method="GET", endpoint=endpoint)
    response_model = self._decode_response_to_model(response, ListOfUnitConversions)
    return response_model

search_for_assets(page_number=1, page_size=1000, refresh_data=False, name_filter='', organization_id=None, site_id=None, asset_type=None)

Gets all assets of the organizations and/or sites the user has access to. At least an organization or site must be specified. If both are specified, then the superset is returned.

Parameters:

Name	Type	Description	Default
refresh_data	bool	If True, it gets data from external sources and updates it in the Organization microservice database. Default is False.	False
page_number	int	Page number of the paged results starting from 1. Default 1	1
page_size	int	Page size of the paged results. Default 1000	1000
organization_id	Union[int, List[int], None]	Organization ID or list of organization IDs for the query.	None
site_id	Union[int, List[int], None]	Site ID or list of site IDs for the query.	None
asset_type	Union[str, List[str], None]	Asset Type or list of asset types for the query.	None
name_filter	str	Filter assets by their name, asset serial number.	''

Returns:

Type	Description
ListOfOrganizationAssetDetails	ReportAssetSearch results

Source code in reportconnectors/api_client/powertrain/report/__init__.py

def search_for_assets(
    self,
    page_number: int = 1,
    page_size: int = 1000,
    refresh_data: bool = False,
    name_filter: str = "",
    organization_id: Union[int, List[int], None] = None,
    site_id: Union[int, List[int], None] = None,
    asset_type: Union[str, List[str], None] = None,
) -> ListOfOrganizationAssetDetails:
    """
    Gets all assets of the organizations and/or sites the user has access to.
    At least an organization or site must be specified.
    If both are specified, then the superset is returned.

    Args:
        refresh_data: If True, it gets data from external sources and updates it in the Organization
            microservice database. Default is False.
        page_number: Page number of the paged results starting from 1. Default 1
        page_size: Page size of the paged results. Default 1000
        organization_id: Organization ID or list of organization IDs for the query.
        site_id: Site ID or list of site IDs for the query.
        asset_type: Asset Type or list of asset types for the query.
        name_filter: Filter assets by their name, asset serial number.

    Returns:
        ReportAssetSearch results
    """

    def _parse_input_list(_input, _type):
        if _input is None:
            return []
        if isinstance(_input, _type):
            return [_input]
        return _input

    organization_ids = _parse_input_list(organization_id, int)
    site_ids = _parse_input_list(site_id, int)
    asset_types = _parse_input_list(asset_type, str)

    if not (organization_ids or site_ids):
        raise ValueError("At least an organization or site must be specified")

    endpoint = "/api/report/Organization/Report/Asset/Search"
    data = {
        "pageNumber": page_number,
        "pageSize": page_size,
        "refreshData": refresh_data,
        "organizationIds": organization_ids,
        "siteIds": site_ids,
        "filter": name_filter,
        "assetTypes": asset_types,
    }
    response = self._make_request(method="POST", endpoint=endpoint, json_data=data)
    response_model = self._decode_response_to_model(response, ListOfOrganizationAssetDetails)
    return response_model

search_for_events(asset_id, start_date, end_date, use_cursor=False, page_size=100, **kwargs)

Returns the events for the given asset ids and date range.

It's based on the Event/Search endpoint.

It also allows filtering events by severity, status, cause, existence of comments, binary logger, data loggers, parameter backup, and message text.

Results are paginated with a default page size of 100. However, if you set use_cursor=True, it will fetch all the events, by using the cursor returned in the response.

Parameters:

Name	Type	Description	Default
asset_id	Union[int, Sequence[int]]	Asset identifier or list of asset identifiers.	required
start_date	DateType	Start date of the search.	required
end_date	DateType	End date of the search.	required
use_cursor	bool	If True, it will fetch all the events using the cursor returned in the response. Otherwise, it will fetch only the first page with the results. Default: False.	False
page_size	int	Number of events per page. Default: 100.	100

Other Parameters:

Name	Type	Description
severity_codes	List[str]	List of severity codes to filter the events
status_codes	List[str]	List of status codes to filter the events
cause_codes	List[str]	List of cause codes to filter the events
has_comments	bool	If True, it will filter the events that have comments
has_binary_logger	bool	If True, it will filter the events that have binary logger
has_data_loggers	bool	If True, it will filter the events that have data loggers
has_parameter_backup	bool	If True, it will filter the events that have parameter backup
message_text_contains	str	Text to search in the message text
cursor	str	Cursor that points to a specific results page

Returns:

Type	Description
EventsResponse	EventSearchResponse object with the list of events and the next cursor.

Source code in reportconnectors/api_client/powertrain/report/__init__.py

def search_for_events(
    self,
    asset_id: Union[int, Sequence[int]],
    start_date: DateType,
    end_date: DateType,
    use_cursor: bool = False,
    page_size: int = 100,
    **kwargs,
) -> EventsResponse:
    """
    Returns the events for the given asset ids and date range.

    It's based on the `Event/Search` endpoint.

    It also allows filtering events by severity, status, cause, existence of comments, binary logger,
    data loggers, parameter backup, and message text.

    Results are paginated with a default page size of 100.
    However, if you set `use_cursor=True`, it will fetch all the events,
    by using the cursor returned in the response.

    Args:
        asset_id: Asset identifier or list of asset identifiers.
        start_date: Start date of the search.
        end_date: End date of the search.
        use_cursor: If True, it will fetch all the events using the cursor returned in the response. Otherwise,
            it will fetch only the first page with the results. Default: `False`.
        page_size: Number of events per page. Default: `100`.

    Keyword Args:
        severity_codes (List[str]): List of severity codes to filter the events
        status_codes (List[str]): List of status codes to filter the events
        cause_codes (List[str]): List of cause codes to filter the events
        has_comments (bool): If True, it will filter the events that have comments
        has_binary_logger (bool): If True, it will filter the events that have binary logger
        has_data_loggers (bool): If True, it will filter the events that have data loggers
        has_parameter_backup (bool): If True, it will filter the events that have parameter backup
        message_text_contains (str): Text to search in the message text
        cursor (str): Cursor that points to a specific results page

    Returns:
        EventSearchResponse object with the list of events and the next cursor.
    """
    asset_ids = asset_id if isinstance(asset_id, list) else [asset_id]
    data = {
        "assetIds": asset_ids,
        "timestampFrom": start_date.strftime(self._datetime_format),
        "timestampTo": end_date.strftime(self._datetime_format),
        "pageSize": page_size,
    }
    # Since API expect extra filtering parameters in camelCase and the kwargs are provided in the snake_case,
    # we need to convert the keys to camelCase and update the data payload
    converted_kwargs = {to_camel(k): v for k, v in kwargs.items()}
    data.update(converted_kwargs)

    endpoint = "/api/report/Report/Event/Search"
    response = self._make_request(method="POST", endpoint=endpoint, json_data=data)
    response_model = self._decode_response_to_model(response, EventsResponse)

    # If the `use_cursor` is True and the first response has the `next_cursor`,
    # we request the next page until the cursor is None.
    # Results is a merged response model.
    while use_cursor and response_model.next_cursor is not None:
        data["cursor"] = response_model.next_cursor
        next_response = self._make_request(method="POST", endpoint=endpoint, json_data=data)
        next_response_model = self._decode_response_to_model(next_response, EventsResponse)
        response_model.events += next_response_model.events
        response_model.next_cursor = next_response_model.next_cursor
    return response_model

search_for_organizations(refresh_data=False, page_number=1, page_size=1000, name_filter='', only_favorites=False)

Gets all organizations matching the specified filter that the user has access to.

Parameters:

Name	Type	Description	Default
refresh_data	bool	If True, it gets data from external sources and updates it in the Organization microservice database. Default is False.	False
page_number	int	Page number of the paged results starting from 1. Default 1	1
page_size	int	Page size of the paged results. Default 1000	1000
name_filter	str	Filter organizations by their name.	''
only_favorites	bool	If True, only favorites are returned.	False

Returns:

Type	Description
ListOfOrganizationDetails	ReportOrganizationSearch results

Source code in reportconnectors/api_client/powertrain/report/__init__.py

def search_for_organizations(
    self,
    refresh_data: bool = False,
    page_number: int = 1,
    page_size: int = 1000,
    name_filter: str = "",
    only_favorites: bool = False,
) -> ListOfOrganizationDetails:
    """
    Gets all organizations matching the specified filter that the user has access to.

    Args:
        refresh_data: If True, it gets data from external sources and updates it in the Organization
            microservice database. Default is False.
        page_number: Page number of the paged results starting from 1. Default 1
        page_size: Page size of the paged results. Default 1000
        name_filter: Filter organizations by their name.
        only_favorites: If True, only favorites are returned.

    Returns:
        ReportOrganizationSearch results
    """

    endpoint = "/api/report/Organization/Report/Organization/Search"
    data = {
        "refreshData": refresh_data,
        "pageNumber": page_number,
        "pageSize": page_size,
        "nameFilter": name_filter,
        "onlyFavorites": only_favorites,
    }
    response = self._make_request(method="POST", endpoint=endpoint, json_data=data)
    response_model = self._decode_response_to_model(response, ListOfOrganizationDetails)
    return response_model

search_for_sites(organization_id, refresh_data=False, page_number=1, page_size=1000)

Get all sites for the specified organizations.

Parameters:

Name	Type	Description	Default
refresh_data	bool	If True, it gets data from external sources and updates it in the Organization microservice database. Default is False.	False
page_number	int	Page number of the paged results starting from 1. Default 1	1
page_size	int	Page size of the paged results. Default 1000	1000
organization_id	Union[int, List[int]]	Organization ID or list of organization IDs.	required

Returns: ListOfReportSites

Source code in reportconnectors/api_client/powertrain/report/__init__.py

def search_for_sites(
    self,
    organization_id: Union[int, List[int]],
    refresh_data: bool = False,
    page_number: int = 1,
    page_size: int = 1000,
) -> ListOfOrganizationSiteDetails:
    """
    Get all sites for the specified organizations.

    Args:
        refresh_data: If True, it gets data from external sources and updates it in the Organization
            microservice database. Default is False.
        page_number: Page number of the paged results starting from 1. Default 1
        page_size: Page size of the paged results. Default 1000
        organization_id: Organization ID or list of organization IDs.
    Returns:
        ListOfReportSites
    """

    endpoint = "/api/report/Organization/Report/Site/Search"
    organization_ids = organization_id if isinstance(organization_id, list) else [organization_id]
    data = {
        "refreshData": refresh_data,
        "pageNumber": page_number,
        "pageSize": page_size,
        "organizationsIds": organization_ids,
    }
    response = self._make_request(method="POST", endpoint=endpoint, json_data=data)
    response_model = self._decode_response_to_model(response, ListOfOrganizationSiteDetails)
    return response_model