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

    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 get_asset_base_data(self, asset_id: str) -> AlmAssetBaseData:
        """
        Gets the asset base data response for an asset identified by the serial number.

        Args:
            asset_id: Asset serial number

        Returns:
            Asset base data.
        """
        endpoint = self._build_endpoint_url(
            model_name=AlmAbbMoModel.INSTALLED_BASE.value, type_name="abb.mo.installedBase.asset@1"
        )
        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_asset_details(self, asset_id: str) -> AlmAssetDetails:
        """
        Gets the asset details for the asset identified by the given asset_id.

        Args:
            asset_id: Asset identifier in the ALM system

        Returns:
            Asset details.
        """
        endpoint = self._build_endpoint_url(
            model_name=AlmAbbMoModel.INSTALLED_BASE.value, type_name="abb.mo.installedBase.assetExtended@1"
        )
        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=AlmAssetDetails)
        asset_details = response_models[0]
        return asset_details

    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.

        Args:
            asset_id: Asset identifier in the ALM system

        Returns:
            Recommended maintenance service data.
        """
        endpoint = self._build_endpoint_url(
            model_name=AlmAbbMoModel.INSTALLED_BASE.value,
            type_name="abb.mo.installedBase.assetRecommendedServiceMaintenance@1",
        )
        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.assetServiceEvents@1",
        )
        params = {"filter": f"properties.mibAssetId = '{asset_id}'"}
        response = self._make_request(method="GET", endpoint=endpoint, params=params)
        response_models = self._decode_response_to_models(
            response=response, model_type=AlmServiceEvents, raise_on_empty=False
        )
        list_of_events = response_models[0].service_event if response_models else []
        return list_of_events

    def get_pm_kit_maintenance_plan(
        self,
        serial_number: Optional[str] = None,
        asset_id: Optional[str] = None,
        device_type: Optional[AlmDeviceType] = None,
    ) -> PmKitMaintenancePlan:
        """
        Gets the PM Kit maintenance plan 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.

        Returns:
            PM Kit maintenance plan.

        Raises:
            ValueError: If the asset ID is missing for the asset with the given serial number

        """
        if asset_id:
            aid = 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.")
        # Get asset details, recommended maintenance services and service events in three parallel threads
        results = {}
        with ThreadPoolExecutor(max_workers=3) as executor:
            futures: Dict[Future, str] = {
                executor.submit(self.get_asset_details, aid): "asset_details",
                executor.submit(self.get_recommended_maintenance_service, aid): "service_recommendations",
                executor.submit(self.get_service_events, aid): "service_events",
            }

            for completed_future in as_completed(futures):
                name = futures[completed_future]
                result = completed_future.result()
                results[name] = result

        output = PmKitMaintenancePlan.from_alm_data(
            asset_details=results["asset_details"],
            alm_service_recommendations=results["service_recommendations"],
            alm_service_events=results["service_events"],
        )
        return output

    @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, **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, **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)

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

get_asset_base_data(asset_id)

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

Parameters:

Name	Type	Description	Default
asset_id	str	Asset serial number	required

Returns:

Type	Description
AlmAssetBaseData	Asset base data.

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

def get_asset_base_data(self, asset_id: str) -> AlmAssetBaseData:
    """
    Gets the asset base data response for an asset identified by the serial number.

    Args:
        asset_id: Asset serial number

    Returns:
        Asset base data.
    """
    endpoint = self._build_endpoint_url(
        model_name=AlmAbbMoModel.INSTALLED_BASE.value, type_name="abb.mo.installedBase.asset@1"
    )
    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_details(asset_id)

Gets the asset details 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
AlmAssetDetails	Asset details.

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

def get_asset_details(self, asset_id: str) -> AlmAssetDetails:
    """
    Gets the asset details for the asset identified by the given asset_id.

    Args:
        asset_id: Asset identifier in the ALM system

    Returns:
        Asset details.
    """
    endpoint = self._build_endpoint_url(
        model_name=AlmAbbMoModel.INSTALLED_BASE.value, type_name="abb.mo.installedBase.assetExtended@1"
    )
    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=AlmAssetDetails)
    asset_details = response_models[0]
    return asset_details

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_pm_kit_maintenance_plan(serial_number=None, asset_id=None, device_type=None)

Gets the PM Kit maintenance plan 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

Returns:

Type	Description
PmKitMaintenancePlan	PM Kit maintenance plan.

Raises:

Type	Description
ValueError	If the asset ID is missing for the asset with the given serial number

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

def get_pm_kit_maintenance_plan(
    self,
    serial_number: Optional[str] = None,
    asset_id: Optional[str] = None,
    device_type: Optional[AlmDeviceType] = None,
) -> PmKitMaintenancePlan:
    """
    Gets the PM Kit maintenance plan 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.

    Returns:
        PM Kit maintenance plan.

    Raises:
        ValueError: If the asset ID is missing for the asset with the given serial number

    """
    if asset_id:
        aid = 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.")
    # Get asset details, recommended maintenance services and service events in three parallel threads
    results = {}
    with ThreadPoolExecutor(max_workers=3) as executor:
        futures: Dict[Future, str] = {
            executor.submit(self.get_asset_details, aid): "asset_details",
            executor.submit(self.get_recommended_maintenance_service, aid): "service_recommendations",
            executor.submit(self.get_service_events, aid): "service_events",
        }

        for completed_future in as_completed(futures):
            name = futures[completed_future]
            result = completed_future.result()
            results[name] = result

    output = PmKitMaintenancePlan.from_alm_data(
        asset_details=results["asset_details"],
        alm_service_recommendations=results["service_recommendations"],
        alm_service_events=results["service_events"],
    )
    return output

get_recommended_maintenance_service(asset_id)

Gets the recommended maintenance service data 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[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.

    Args:
        asset_id: Asset identifier in the ALM system

    Returns:
        Recommended maintenance service data.
    """
    endpoint = self._build_endpoint_url(
        model_name=AlmAbbMoModel.INSTALLED_BASE.value,
        type_name="abb.mo.installedBase.assetRecommendedServiceMaintenance@1",
    )
    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.assetServiceEvents@1",
    )
    params = {"filter": f"properties.mibAssetId = '{asset_id}'"}
    response = self._make_request(method="GET", endpoint=endpoint, params=params)
    response_models = self._decode_response_to_models(
        response=response, model_type=AlmServiceEvents, raise_on_empty=False
    )
    list_of_events = response_models[0].service_event if response_models else []
    return list_of_events