ALM API Client

Bases: BaseAPIClient

Motion Asset Lifecycle Management (ALM) API Client.

Portal Documentation

It uses the CIAM API client as a base class to handle authentication.

So far it only supports the endpoints needed for the PM Kit maintenance plan.

Notes

There is a special method to get the PM Kit maintenance plan for an asset. It queries the asset details, recommended maintenance services and service events in parallel and then creates a PM Kit maintenance plan object from the data.

Source code in reportconnectors/api_client/alm/__init__.py

class AlmAPIClient(BaseAPIClient):
    """
    Motion Asset Lifecycle Management (ALM) API Client.

    [Portal](https://alm.motion.abb.com/portal/)
    [Documentation](https://alm.motion.abb.com/)

    It uses the CIAM API client as a base class to handle authentication.

    So far it only supports the endpoints needed for the PM Kit maintenance plan.

    Notes:
        There is a special method to get the PM Kit maintenance plan for an asset.
        It queries the asset details, recommended maintenance services and service events in parallel
        and then creates a PM Kit maintenance plan object from the data.
    """

    def __init__(self, url: str, ciam_url: str, ciam_api_client: Optional[CiamAPIClient] = None, **kwargs):
        """
        Initializes the ALM API client.

        Args:
            url: ALM API URL
            ciam_url: CIAM API URL

        Keyword Args:
            timeout (float): Timeout for the API requests. Default: `60`
            proxies (Dict[str, str]): Proxy settings. Default: `None`
            cert_path (Union[str, bool]): Path to the certificate file or `False` to disable certificate verification.
                Default: `False`
        """
        super().__init__(url=url, **kwargs)
        self.ciam_api_client = CiamAPIClient(url=ciam_url, **kwargs) if not ciam_api_client else ciam_api_client

    def authenticate(self, client_id: str, client_secret: str, password: str, username: str, **kwargs) -> bool:
        """
        Authenticates the client.

        Because the ALM service uses CIAM as an authentication provider,
        the validation of logged status is done through the CIAM API client.

        Args:
            client_id: Client ID
            client_secret: Client secret
            password: User password
            username: Username

        Keyword Args:
            token_url (str): Authentication token URL. If not provided, it is taken from the CIAM API client
            timeout (float): Timeout for the authentication process. Default: `60`

        Returns:
            True if the client is successfully authenticated and the access token is obtained and set.
            Otherwise, it returns False.
        """
        self.ciam_api_client.authenticate(
            client_id=client_id, client_secret=client_secret, password=password, username=username, **kwargs
        )
        self.auth_data = self.ciam_api_client.auth_data
        return self.is_logged

    @property
    def is_logged(self):
        """
        Checks if the client is logged in.
        """
        time_now = datetime.datetime.now(datetime.timezone.utc)
        _is_logged = (
            bool(self.auth_token)
            and isinstance(self.token_expiration_date, datetime.datetime)
            and (self.token_expiration_date > time_now)
        )
        return _is_logged

    def set_token(self, access_token: str, validate: bool = False, **kwargs) -> bool:
        """
        Sets the authorization tokens without the authentication process. It might be used when the
        tokens are obtained from external source.

        Args:
            access_token: Access token to be set.
            validate: If set to True, the token will be validated. Default: False

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

        status = self.ciam_api_client.set_token(access_token=access_token, validate=validate, **kwargs)
        if status:
            self.auth_data = dict(self.ciam_api_client.auth_data)
        return status

    @classmethod
    def from_access_token(cls, url: str, ciam_url: str, access_token: str, **kwargs) -> "AlmAPIClient":
        """
        Initialize the client with the provided access token.

        Args:
            url: URL of the API.
            ciam_url: URL of the CIAM API.
            access_token: Valid access token to the API.

        Keyword Args:
            refresh_token (str): Valid refresh token that will be used to automatically refresh the access token
            id_token (str): Valid ID token
            validate (bool): If True, the token will be validated. Default: `False`
            audience (str): Audience for the token validation. Default: `None`

        Returns:
            Initialized and authenticated ALM API client.
        """
        validate: bool = kwargs.pop("validate", False)
        audience: Optional[str] = kwargs.pop("audience", None)
        refresh_token: Optional[str] = kwargs.pop("refresh_token", None)
        id_token: Optional[str] = kwargs.pop("id_token", None)

        client = cls(url=url, ciam_url=ciam_url, **kwargs)
        client.set_token(
            access_token=access_token,
            refresh_token=refresh_token,
            id_token=id_token,
            validate=validate,
            audience=audience,
        )
        return client

    def get_asset_base_data(
        self, asset_id: Optional[str] = None, serial_number: Optional[str] = None, pt_id: Optional[str] = None
    ) -> AlmAssetBaseData:
        """
        Gets the asset base data response for an asset identified by the serial number.

        Args:
            asset_id: Asset serial number
            serial_number: Asset serial number
            pt_id: Powertrain asset ID

        Returns:
            Asset base data.
        """
        endpoint = self._build_endpoint_url(
            model_name=AlmAbbMoModel.INSTALLED_BASE.value, type_name="abb.mo.installedBase.asset@2"
        )
        if pt_id:
            params = {"filter": f"powertrainAssetId = '{pt_id}'"}
        elif serial_number:
            params = {"filter": f"properties.serialNumber = '{serial_number}'"}
        else:
            params = {"filter": f"properties.assetId = '{asset_id}'"}

        response = self._make_request(method="GET", endpoint=endpoint, params=params)
        response_models = self._decode_response_to_models(response=response, model_type=AlmAssetBaseData)
        asset_base_data = response_models[0]
        return asset_base_data

    def get_asset_core_details(
        self,
        asset_id: Optional[str] = None,
        serial_number: Optional[str] = None,
        device_type: Optional[AlmDeviceType] = None,
        strict_mode: bool = False,
    ) -> AlmCoreAssetData:
        """
        Gets the asset core details for the asset identified by either asset_id or serial_number.
        If both asset_id and serial_number are provided, asset_id has priority.

        If device_type is provided, it filters the results based on the device type.

        Notes:
            This can be useful when there are multiple assets with the same serial number but different device types.
            For example, in case of MV drives, there is another asset (LV drive) with the same serial number.
            To filter out the LV drive, the device_type="MvDrive" parameter can be used.

        Args:
            asset_id: Asset identifier in the ALM system
            serial_number: Asset serial number
            device_type: Device type to filter the results (optional).
            strict_mode: If True, then the device type must match exactly. Otherwise, if there is only
                one asset with the given serial number, it is returned regardless of the device type. Default: False

        Returns:
            Asset core details data.

        Raises:
            ValueError: If neither asset_id nor serial_number is provided
            ValueError: If the asset with the given parameters is not found
            ValueError: If multiple assets are found for the given parameters in strict mode
        """

        if asset_id is None and serial_number is None:
            raise ValueError("Either asset_id or serial_number must be provided to get the asset core details.")

        endpoint = self._build_endpoint_url(model_name=AlmAbbMoModel.CORE.value, type_name="abb.mo.core.asset@1")
        # Build the filter parameters based on provided arguments
        # Asset ID has priority over serial number if both are provided
        params = {}
        if serial_number:
            params["filter"] = f"properties.asset.serialNumber = '{serial_number}'"
        if asset_id:
            params["filter"] = f"properties.assetId = '{asset_id}'"
        response = self._make_request(method="GET", endpoint=endpoint, params=params)
        response_models = self._decode_response_to_models(response=response, model_type=AlmCoreAssetData)

        # If there is only one asset with the given serial number, return it regardless of the device type
        if not strict_mode and len(response_models) == 1:
            log.debug(
                "Only one asset found, and the strict mode is not active. Returning it regardless of device type."
            )
            return response_models[0]

        # Filter the results based on device type if provided
        if device_type:
            response_models = [
                model
                for model in response_models
                if model.device_type and model.device_type.value == device_type.value
            ]
        if not response_models:
            raise ValueError(
                f"Asset with the given parameters not found. {asset_id=}, {serial_number=}, {device_type=}."
            )

        if len(response_models) > 1:
            if strict_mode:
                raise ValueError(
                    f"Multiple assets found for the given parameters in strict mode: {asset_id=}, "
                    f"{serial_number=}, {device_type=}."
                )
            else:
                log.warning(
                    f"Multiple assets found for the given parameters: {asset_id=}, {serial_number=}, {device_type=}. "
                    "Returning the first one."
                )
        asset_core_data = response_models[0]
        return asset_core_data

    def get_asset_id_from_serial_number(
        self, serial_number: str, device_type: Optional[AlmDeviceType] = None, strict_mode: bool = False
    ) -> Optional[str]:
        """
        Gets the ALM asset identifier from based on the asset serial number.

        Args:
            serial_number: Asset serial number
            device_type: Device type to filter the results (optional).
            strict_mode: If True, then the device type must match exactly. Otherwise, if there is only
                one asset with the given serial number, it is returned regardless of the device type. Default: False

        Notes:
            Filtering by device type can be useful for MV Drives, where there are multiple assets with the same
            serial number but different device types.
            To filter out the LV drive, the device_type="MvDrive" parameter can be used.


        Returns:
            ALM asset identifier.
        """
        asset_core_data = self.get_asset_core_details(
            serial_number=serial_number, device_type=device_type, strict_mode=strict_mode
        )
        if asset_core_data.asset_id.value is None:
            raise ValueError(f"Asset ID is missing for the asset with serial number: {serial_number}")
        return asset_core_data.asset_id.value

    def get_recommended_maintenance_service(self, asset_id: str) -> List[AlmRecommendedService]:
        """
        Gets the recommended maintenance service data for the asset identified by the given asset_id.

        Notes:
            Since version 1.29.2 we are using the "abb.mo.productService.recommendedService@2" endpoint
            which returns a full list of recommended services for the asset.

        Args:
            asset_id: Asset identifier in the ALM system

        Returns:
            Recommended maintenance service data.
        """
        endpoint = self._build_endpoint_url(
            model_name=AlmAbbMoModel.PRODUCT_SERVICE.value,
            type_name="abb.mo.productService.recommendedService@2",
        )
        params = {"filter": f"properties.assetId = '{asset_id}'"}
        response = self._make_request(method="GET", endpoint=endpoint, params=params)
        response_models = self._decode_response_to_models(
            response=response, model_type=AlmRecommendedService, raise_on_empty=False
        )
        list_of_services = response_models
        return list_of_services

    def get_service_events(self, asset_id: str) -> List[AlmServiceEvent]:
        """
        Gets the service events for the asset identified by the given asset_id.

        Args:
            asset_id: Asset identifier in the ALM system

        Returns:
            List of service events.

        """
        endpoint = self._build_endpoint_url(
            model_name=AlmAbbMoModel.PRODUCT_SERVICE.value,
            type_name="abb.mo.productService.serviceEvent@2",
        )
        params = {"filter": f"properties.assetId = '{asset_id}'"}
        response = self._make_request(method="GET", endpoint=endpoint, params=params)
        response_models = self._decode_response_to_models(
            response=response, model_type=AlmServiceEvent, raise_on_empty=False
        )
        list_of_events = response_models if response_models else []
        return list_of_events

    def get_alm_service_data(
        self,
        serial_number: Optional[str] = None,
        asset_id: Optional[str] = None,
        device_type: Optional[AlmDeviceType] = None,
        **kwargs,
    ) -> AlmServiceData | None:
        """
        Gets the AlmServiceData object for the asset with the given serial number.

        It queries the asset details, recommended maintenance services and service events in parallel
        and then creates a PM Kit maintenance plan object from the data.

        Args:
            serial_number: Asset serial number
            asset_id: Asset identifier in the ALM system.
            device_type: Device type to filter the results (optional). When dealing with MV drives, it is recommended
                to set this parameter to AlmDeviceType.MV_DRIVE to avoid getting the LV drive data.

        Keyword Args:
            skip_service_events_exception (bool): If True, then if there is an error while getting service events,
                it will be ignored and the empty list is returned. Default: True
            include_service_events (bool): If True, then service events will be included in the result. If False,
                then service events will not be queried and the empty list will be returned. Default: True

        Returns:
            AlmServiceData object containing the asset details, recommended maintenance services and service events
            for the asset with the given serial number.
        """
        include_service_events: bool = kwargs.pop("include_service_events", True)
        skip_service_events_exception: bool = kwargs.pop("skip_service_events_exception", True)

        aid = self._find_alm_asset_id(asset_id, device_type, serial_number)

        # Get asset details, recommended maintenance services and service events in three parallel threads
        results: Dict[str, Any] = {"serial_number": serial_number, "asset_id": aid}
        with ThreadPoolExecutor(max_workers=3) as executor:
            futures: Dict[Future, str] = {
                executor.submit(self.get_asset_base_data, aid): "asset_base_data",
                executor.submit(self.get_recommended_maintenance_service, aid): "service_recommendations",
            }
            if include_service_events:
                futures[executor.submit(self.get_service_events, aid)] = "service_events"
            else:
                results["service_events"] = []

            for completed_future in as_completed(futures):
                name = futures[completed_future]
                try:
                    result = completed_future.result()
                except (requests.exceptions.HTTPError, ValueError) as e:
                    if name == "service_events" and skip_service_events_exception:
                        log.info(
                            f"Error while getting service events for {aid=}. Error: {e}. Returning empty list of service events."
                        )
                        result = []
                    else:
                        raise e
                results[name] = result

        try:
            output = AlmServiceData.model_validate(results)
            log.debug("Alm data created for serial number: %s", serial_number)
            return output

        except ValueError:
            log.info("Failed to create AlmServiceData model for serial number: %s", serial_number)
            return None

    def _find_alm_asset_id(
        self, asset_id: str | None, device_type: AlmDeviceType | None, serial_number: str | None
    ) -> str:
        """
        Finds the ALM asset ID based on the provided asset_id, device_type and serial_number.

        Args:
            asset_id: ALM asset ID. If provided, it is returned without any additional checks.
            device_type: Device type to filter the results when searching by serial number.
                This is useful for MV drives, where there are
            serial_number: Asset serial number. It is used to find the asset ID if the asset ID is not provided.
                If there are multiple assets with the same serial number,
                then device type is used to filter the results.

        Raises:
            ValueError: If neither asset_id nor serial_number is provided
            ValueError: If the asset with the given serial number is not found

        Returns:
            ALM asset ID

        """
        if asset_id:
            return asset_id
        elif asset_id is None and serial_number is not None:
            asset_id_from_serial_number = self.get_asset_id_from_serial_number(
                serial_number=serial_number, device_type=device_type
            )
            if asset_id_from_serial_number is None:
                raise ValueError(f"Asset ID is missing for the asset with serial number: {serial_number}")
            aid = asset_id_from_serial_number
        else:
            raise ValueError("Either asset_id or serial_number must be provided to get the PM Kit maintenance plan.")
        return aid

    @staticmethod
    def _decode_response_to_models(
        response: Optional[requests.Response], model_type: Type[AcceptedModelType], raise_on_empty: bool = True
    ) -> List[AcceptedModelType]:
        """
        Decodes the received response to the given model.

        Since all ALM responses have the data field with the list of properties, where each property is a specific data
        model, this method requires the desired model type to be provided.
        Based on the model type, it extracts the properties from the data field and converts them to the list
        of desired models.
        In many cases there is only a single model in the list, but it is still returned as a list.

        Args:
            response: Response object received from the ALM API.
            model_type: Specific model type expected in the `data[].properties` fields.
            raise_on_empty: Raise ValueError if the response data is empty. Default: True

        Returns:
            List of data models extracted from the general response

        Raises:
            ValueError: If the response data is empty
            TypeError: If the response is not valid Response object
            requests.exceptions.HTTPError: If the response status code is not 2xx
            ValidationError: If the response content cannot be parsed as JSON or does not match the expected model

        """
        if not isinstance(response, requests.Response):
            raise TypeError(f"Invalid response type: {type(response)}")

        try:
            # Check response status, raise exception on error
            response.raise_for_status()
            # Initialize the specific model class based on the model type
            # mypy complains about the following line, but it is correct
            specific_model_class = AlmGeneralResponse[model_type]  # type: ignore
            # Parse response content as JSON
            alm_response_model = specific_model_class.model_validate_json(response.text)

        except requests.exceptions.HTTPError as http_exp:
            log.exception(http_exp)
            if response.text:
                log.error(f"API Response: {response.text}")
            raise http_exp
        except ValidationError as validation_error:
            log.exception(validation_error)
            raise validation_error

        list_of_specific_models = [data.properties for data in alm_response_model.data]
        if raise_on_empty and not list_of_specific_models:
            raise ValueError(f"Response data is empty for the model type: {model_type}")
        return list_of_specific_models

    @staticmethod
    def _build_endpoint_url(model_name: str, type_name: str, version: str = "v1") -> str:
        """
        Builds the ALM endpoint. It uses the default ALM URL template and fills it with models, types
        and version details.

        Args:
            model_name: Name of the data model
            type_name: Name of the data type
            version: API version. Default: "v1"

        Returns:
            ALM endpoint URL
        """
        return f"/api/im/{version}/models/{model_name}/types/{type_name}/assets"

is_logged property

Checks if the client is logged in.

__init__(url, ciam_url, ciam_api_client=None, **kwargs)

Initializes the ALM API client.

Parameters:

Name	Type	Description	Default
url	str	ALM API URL	required
ciam_url	str	CIAM API URL	required

Other Parameters:

Name	Type	Description
timeout	float	Timeout for the API requests. Default: 60
proxies	Dict[str, str]	Proxy settings. Default: None
cert_path	Union[str, bool]	Path to the certificate file or False to disable certificate verification. Default: False

Source code in reportconnectors/api_client/alm/__init__.py

def __init__(self, url: str, ciam_url: str, ciam_api_client: Optional[CiamAPIClient] = None, **kwargs):
    """
    Initializes the ALM API client.

    Args:
        url: ALM API URL
        ciam_url: CIAM API URL

    Keyword Args:
        timeout (float): Timeout for the API requests. Default: `60`
        proxies (Dict[str, str]): Proxy settings. Default: `None`
        cert_path (Union[str, bool]): Path to the certificate file or `False` to disable certificate verification.
            Default: `False`
    """
    super().__init__(url=url, **kwargs)
    self.ciam_api_client = CiamAPIClient(url=ciam_url, **kwargs) if not ciam_api_client else ciam_api_client

authenticate(client_id, client_secret, password, username, **kwargs)

Authenticates the client.

Because the ALM service uses CIAM as an authentication provider, the validation of logged status is done through the CIAM API client.

Parameters:

Name	Type	Description	Default
client_id	str	Client ID	required
client_secret	str	Client secret	required
password	str	User password	required
username	str	Username	required

Other Parameters:

Name	Type	Description
token_url	str	Authentication token URL. If not provided, it is taken from the CIAM API client
timeout	float	Timeout for the authentication process. Default: 60

Returns:

Type	Description
bool	True if the client is successfully authenticated and the access token is obtained and set.
bool	Otherwise, it returns False.

Source code in reportconnectors/api_client/alm/__init__.py

def authenticate(self, client_id: str, client_secret: str, password: str, username: str, **kwargs) -> bool:
    """
    Authenticates the client.

    Because the ALM service uses CIAM as an authentication provider,
    the validation of logged status is done through the CIAM API client.

    Args:
        client_id: Client ID
        client_secret: Client secret
        password: User password
        username: Username

    Keyword Args:
        token_url (str): Authentication token URL. If not provided, it is taken from the CIAM API client
        timeout (float): Timeout for the authentication process. Default: `60`

    Returns:
        True if the client is successfully authenticated and the access token is obtained and set.
        Otherwise, it returns False.
    """
    self.ciam_api_client.authenticate(
        client_id=client_id, client_secret=client_secret, password=password, username=username, **kwargs
    )
    self.auth_data = self.ciam_api_client.auth_data
    return self.is_logged

from_access_token(url, ciam_url, access_token, **kwargs) classmethod

Initialize the client with the provided access token.

Parameters:

Name	Type	Description	Default
url	str	URL of the API.	required
ciam_url	str	URL of the CIAM API.	required
access_token	str	Valid access token to the API.	required

Other Parameters:

Name	Type	Description
refresh_token	str	Valid refresh token that will be used to automatically refresh the access token
id_token	str	Valid ID token
validate	bool	If True, the token will be validated. Default: False
audience	str	Audience for the token validation. Default: None

Returns:

Type	Description
AlmAPIClient	Initialized and authenticated ALM API client.

Source code in reportconnectors/api_client/alm/__init__.py

@classmethod
def from_access_token(cls, url: str, ciam_url: str, access_token: str, **kwargs) -> "AlmAPIClient":
    """
    Initialize the client with the provided access token.

    Args:
        url: URL of the API.
        ciam_url: URL of the CIAM API.
        access_token: Valid access token to the API.

    Keyword Args:
        refresh_token (str): Valid refresh token that will be used to automatically refresh the access token
        id_token (str): Valid ID token
        validate (bool): If True, the token will be validated. Default: `False`
        audience (str): Audience for the token validation. Default: `None`

    Returns:
        Initialized and authenticated ALM API client.
    """
    validate: bool = kwargs.pop("validate", False)
    audience: Optional[str] = kwargs.pop("audience", None)
    refresh_token: Optional[str] = kwargs.pop("refresh_token", None)
    id_token: Optional[str] = kwargs.pop("id_token", None)

    client = cls(url=url, ciam_url=ciam_url, **kwargs)
    client.set_token(
        access_token=access_token,
        refresh_token=refresh_token,
        id_token=id_token,
        validate=validate,
        audience=audience,
    )
    return client

get_alm_service_data(serial_number=None, asset_id=None, device_type=None, **kwargs)

Gets the AlmServiceData object for the asset with the given serial number.

It queries the asset details, recommended maintenance services and service events in parallel and then creates a PM Kit maintenance plan object from the data.

Parameters:

Name	Type	Description	Default
serial_number	Optional[str]	Asset serial number	None
asset_id	Optional[str]	Asset identifier in the ALM system.	None
device_type	Optional[AlmDeviceType]	Device type to filter the results (optional). When dealing with MV drives, it is recommended to set this parameter to AlmDeviceType.MV_DRIVE to avoid getting the LV drive data.	None

Other Parameters:

Name	Type	Description
skip_service_events_exception	bool	If True, then if there is an error while getting service events, it will be ignored and the empty list is returned. Default: True
include_service_events	bool	If True, then service events will be included in the result. If False, then service events will not be queried and the empty list will be returned. Default: True

Returns:

Type	Description
AlmServiceData | None	AlmServiceData object containing the asset details, recommended maintenance services and service events
AlmServiceData | None	for the asset with the given serial number.

Source code in reportconnectors/api_client/alm/__init__.py

def get_alm_service_data(
    self,
    serial_number: Optional[str] = None,
    asset_id: Optional[str] = None,
    device_type: Optional[AlmDeviceType] = None,
    **kwargs,
) -> AlmServiceData | None:
    """
    Gets the AlmServiceData object for the asset with the given serial number.

    It queries the asset details, recommended maintenance services and service events in parallel
    and then creates a PM Kit maintenance plan object from the data.

    Args:
        serial_number: Asset serial number
        asset_id: Asset identifier in the ALM system.
        device_type: Device type to filter the results (optional). When dealing with MV drives, it is recommended
            to set this parameter to AlmDeviceType.MV_DRIVE to avoid getting the LV drive data.

    Keyword Args:
        skip_service_events_exception (bool): If True, then if there is an error while getting service events,
            it will be ignored and the empty list is returned. Default: True
        include_service_events (bool): If True, then service events will be included in the result. If False,
            then service events will not be queried and the empty list will be returned. Default: True

    Returns:
        AlmServiceData object containing the asset details, recommended maintenance services and service events
        for the asset with the given serial number.
    """
    include_service_events: bool = kwargs.pop("include_service_events", True)
    skip_service_events_exception: bool = kwargs.pop("skip_service_events_exception", True)

    aid = self._find_alm_asset_id(asset_id, device_type, serial_number)

    # Get asset details, recommended maintenance services and service events in three parallel threads
    results: Dict[str, Any] = {"serial_number": serial_number, "asset_id": aid}
    with ThreadPoolExecutor(max_workers=3) as executor:
        futures: Dict[Future, str] = {
            executor.submit(self.get_asset_base_data, aid): "asset_base_data",
            executor.submit(self.get_recommended_maintenance_service, aid): "service_recommendations",
        }
        if include_service_events:
            futures[executor.submit(self.get_service_events, aid)] = "service_events"
        else:
            results["service_events"] = []

        for completed_future in as_completed(futures):
            name = futures[completed_future]
            try:
                result = completed_future.result()
            except (requests.exceptions.HTTPError, ValueError) as e:
                if name == "service_events" and skip_service_events_exception:
                    log.info(
                        f"Error while getting service events for {aid=}. Error: {e}. Returning empty list of service events."
                    )
                    result = []
                else:
                    raise e
            results[name] = result

    try:
        output = AlmServiceData.model_validate(results)
        log.debug("Alm data created for serial number: %s", serial_number)
        return output

    except ValueError:
        log.info("Failed to create AlmServiceData model for serial number: %s", serial_number)
        return None

get_asset_base_data(asset_id=None, serial_number=None, pt_id=None)

Gets the asset base data response for an asset identified by the serial number.

Parameters:

Name	Type	Description	Default
asset_id	Optional[str]	Asset serial number	None
serial_number	Optional[str]	Asset serial number	None
pt_id	Optional[str]	Powertrain asset ID	None

Returns:

Type	Description
AlmAssetBaseData	Asset base data.

Source code in reportconnectors/api_client/alm/__init__.py

def get_asset_base_data(
    self, asset_id: Optional[str] = None, serial_number: Optional[str] = None, pt_id: Optional[str] = None
) -> AlmAssetBaseData:
    """
    Gets the asset base data response for an asset identified by the serial number.

    Args:
        asset_id: Asset serial number
        serial_number: Asset serial number
        pt_id: Powertrain asset ID

    Returns:
        Asset base data.
    """
    endpoint = self._build_endpoint_url(
        model_name=AlmAbbMoModel.INSTALLED_BASE.value, type_name="abb.mo.installedBase.asset@2"
    )
    if pt_id:
        params = {"filter": f"powertrainAssetId = '{pt_id}'"}
    elif serial_number:
        params = {"filter": f"properties.serialNumber = '{serial_number}'"}
    else:
        params = {"filter": f"properties.assetId = '{asset_id}'"}

    response = self._make_request(method="GET", endpoint=endpoint, params=params)
    response_models = self._decode_response_to_models(response=response, model_type=AlmAssetBaseData)
    asset_base_data = response_models[0]
    return asset_base_data

get_asset_core_details(asset_id=None, serial_number=None, device_type=None, strict_mode=False)

Gets the asset core details for the asset identified by either asset_id or serial_number. If both asset_id and serial_number are provided, asset_id has priority.

If device_type is provided, it filters the results based on the device type.

Notes

This can be useful when there are multiple assets with the same serial number but different device types. For example, in case of MV drives, there is another asset (LV drive) with the same serial number. To filter out the LV drive, the device_type="MvDrive" parameter can be used.

Parameters:

Name	Type	Description	Default
asset_id	Optional[str]	Asset identifier in the ALM system	None
serial_number	Optional[str]	Asset serial number	None
device_type	Optional[AlmDeviceType]	Device type to filter the results (optional).	None
strict_mode	bool	If True, then the device type must match exactly. Otherwise, if there is only one asset with the given serial number, it is returned regardless of the device type. Default: False	False

Returns:

Type	Description
AlmCoreAssetData	Asset core details data.

Raises:

Type	Description
ValueError	If neither asset_id nor serial_number is provided
ValueError	If the asset with the given parameters is not found
ValueError	If multiple assets are found for the given parameters in strict mode

Source code in reportconnectors/api_client/alm/__init__.py

def get_asset_core_details(
    self,
    asset_id: Optional[str] = None,
    serial_number: Optional[str] = None,
    device_type: Optional[AlmDeviceType] = None,
    strict_mode: bool = False,
) -> AlmCoreAssetData:
    """
    Gets the asset core details for the asset identified by either asset_id or serial_number.
    If both asset_id and serial_number are provided, asset_id has priority.

    If device_type is provided, it filters the results based on the device type.

    Notes:
        This can be useful when there are multiple assets with the same serial number but different device types.
        For example, in case of MV drives, there is another asset (LV drive) with the same serial number.
        To filter out the LV drive, the device_type="MvDrive" parameter can be used.

    Args:
        asset_id: Asset identifier in the ALM system
        serial_number: Asset serial number
        device_type: Device type to filter the results (optional).
        strict_mode: If True, then the device type must match exactly. Otherwise, if there is only
            one asset with the given serial number, it is returned regardless of the device type. Default: False

    Returns:
        Asset core details data.

    Raises:
        ValueError: If neither asset_id nor serial_number is provided
        ValueError: If the asset with the given parameters is not found
        ValueError: If multiple assets are found for the given parameters in strict mode
    """

    if asset_id is None and serial_number is None:
        raise ValueError("Either asset_id or serial_number must be provided to get the asset core details.")

    endpoint = self._build_endpoint_url(model_name=AlmAbbMoModel.CORE.value, type_name="abb.mo.core.asset@1")
    # Build the filter parameters based on provided arguments
    # Asset ID has priority over serial number if both are provided
    params = {}
    if serial_number:
        params["filter"] = f"properties.asset.serialNumber = '{serial_number}'"
    if asset_id:
        params["filter"] = f"properties.assetId = '{asset_id}'"
    response = self._make_request(method="GET", endpoint=endpoint, params=params)
    response_models = self._decode_response_to_models(response=response, model_type=AlmCoreAssetData)

    # If there is only one asset with the given serial number, return it regardless of the device type
    if not strict_mode and len(response_models) == 1:
        log.debug(
            "Only one asset found, and the strict mode is not active. Returning it regardless of device type."
        )
        return response_models[0]

    # Filter the results based on device type if provided
    if device_type:
        response_models = [
            model
            for model in response_models
            if model.device_type and model.device_type.value == device_type.value
        ]
    if not response_models:
        raise ValueError(
            f"Asset with the given parameters not found. {asset_id=}, {serial_number=}, {device_type=}."
        )

    if len(response_models) > 1:
        if strict_mode:
            raise ValueError(
                f"Multiple assets found for the given parameters in strict mode: {asset_id=}, "
                f"{serial_number=}, {device_type=}."
            )
        else:
            log.warning(
                f"Multiple assets found for the given parameters: {asset_id=}, {serial_number=}, {device_type=}. "
                "Returning the first one."
            )
    asset_core_data = response_models[0]
    return asset_core_data

get_asset_id_from_serial_number(serial_number, device_type=None, strict_mode=False)

Gets the ALM asset identifier from based on the asset serial number.

Parameters:

Name	Type	Description	Default
serial_number	str	Asset serial number	required
device_type	Optional[AlmDeviceType]	Device type to filter the results (optional).	None
strict_mode	bool	If True, then the device type must match exactly. Otherwise, if there is only one asset with the given serial number, it is returned regardless of the device type. Default: False	False

Notes

Filtering by device type can be useful for MV Drives, where there are multiple assets with the same serial number but different device types. To filter out the LV drive, the device_type="MvDrive" parameter can be used.

Returns:

Type	Description
Optional[str]	ALM asset identifier.

Source code in reportconnectors/api_client/alm/__init__.py

def get_asset_id_from_serial_number(
    self, serial_number: str, device_type: Optional[AlmDeviceType] = None, strict_mode: bool = False
) -> Optional[str]:
    """
    Gets the ALM asset identifier from based on the asset serial number.

    Args:
        serial_number: Asset serial number
        device_type: Device type to filter the results (optional).
        strict_mode: If True, then the device type must match exactly. Otherwise, if there is only
            one asset with the given serial number, it is returned regardless of the device type. Default: False

    Notes:
        Filtering by device type can be useful for MV Drives, where there are multiple assets with the same
        serial number but different device types.
        To filter out the LV drive, the device_type="MvDrive" parameter can be used.


    Returns:
        ALM asset identifier.
    """
    asset_core_data = self.get_asset_core_details(
        serial_number=serial_number, device_type=device_type, strict_mode=strict_mode
    )
    if asset_core_data.asset_id.value is None:
        raise ValueError(f"Asset ID is missing for the asset with serial number: {serial_number}")
    return asset_core_data.asset_id.value

get_recommended_maintenance_service(asset_id)

Gets the recommended maintenance service data for the asset identified by the given asset_id.

Notes

Since version 1.29.2 we are using the "abb.mo.productService.recommendedService@2" endpoint which returns a full list of recommended services for the asset.

Parameters:

Name	Type	Description	Default
asset_id	str	Asset identifier in the ALM system	required

Returns:

Type	Description
List[AlmRecommendedService]	Recommended maintenance service data.

Source code in reportconnectors/api_client/alm/__init__.py

def get_recommended_maintenance_service(self, asset_id: str) -> List[AlmRecommendedService]:
    """
    Gets the recommended maintenance service data for the asset identified by the given asset_id.

    Notes:
        Since version 1.29.2 we are using the "abb.mo.productService.recommendedService@2" endpoint
        which returns a full list of recommended services for the asset.

    Args:
        asset_id: Asset identifier in the ALM system

    Returns:
        Recommended maintenance service data.
    """
    endpoint = self._build_endpoint_url(
        model_name=AlmAbbMoModel.PRODUCT_SERVICE.value,
        type_name="abb.mo.productService.recommendedService@2",
    )
    params = {"filter": f"properties.assetId = '{asset_id}'"}
    response = self._make_request(method="GET", endpoint=endpoint, params=params)
    response_models = self._decode_response_to_models(
        response=response, model_type=AlmRecommendedService, raise_on_empty=False
    )
    list_of_services = response_models
    return list_of_services

get_service_events(asset_id)

Gets the service events for the asset identified by the given asset_id.

Parameters:

Name	Type	Description	Default
asset_id	str	Asset identifier in the ALM system	required

Returns:

Type	Description
List[AlmServiceEvent]	List of service events.

Source code in reportconnectors/api_client/alm/__init__.py

def get_service_events(self, asset_id: str) -> List[AlmServiceEvent]:
    """
    Gets the service events for the asset identified by the given asset_id.

    Args:
        asset_id: Asset identifier in the ALM system

    Returns:
        List of service events.

    """
    endpoint = self._build_endpoint_url(
        model_name=AlmAbbMoModel.PRODUCT_SERVICE.value,
        type_name="abb.mo.productService.serviceEvent@2",
    )
    params = {"filter": f"properties.assetId = '{asset_id}'"}
    response = self._make_request(method="GET", endpoint=endpoint, params=params)
    response_models = self._decode_response_to_models(
        response=response, model_type=AlmServiceEvent, raise_on_empty=False
    )
    list_of_events = response_models if response_models else []
    return list_of_events

set_token(access_token, validate=False, **kwargs)

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

Parameters:

Name	Type	Description	Default
access_token	str	Access token to be set.	required
validate	bool	If set to True, the token will be validated. Default: False	False

Returns:

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

Source code in reportconnectors/api_client/alm/__init__.py

def set_token(self, access_token: str, validate: bool = False, **kwargs) -> bool:
    """
    Sets the authorization tokens without the authentication process. It might be used when the
    tokens are obtained from external source.

    Args:
        access_token: Access token to be set.
        validate: If set to True, the token will be validated. Default: False

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

    status = self.ciam_api_client.set_token(access_token=access_token, validate=validate, **kwargs)
    if status:
        self.auth_data = dict(self.ciam_api_client.auth_data)
    return status