SmartSensor API Client

Bases: BaseAPIClient

Smart Sensor API Client.

It is used to communicate with SmartSensor API that provides the data related to SmartSensor assets and organizations.

API documentation can be found here

Source code in reportconnectors/api_client/smartsensor_api_client.py

class SmartSensorAPIClient(BaseAPIClient):
    """
    Smart Sensor API Client.

    It is used to communicate with SmartSensor API  that provides the data related to SmartSensor
    assets and organizations.

    API documentation can be found [here](https://api.smartsensor.abb.com/swagger/)

    """

    __version__ = "1.3.0"
    _non_cachable_endpoints: Set[str] = {
        "Auth/",
    }
    _default_timeout: float = 120.0  # in seconds
    _default_retries: int = 3
    _http_statuses_to_retry: Set[HTTPStatus] = {HTTPStatus.FORBIDDEN}
    _feature_codes = {
        "Asset/List": "EXT_ListAssets",
        "Measurement/Value": "EXT_AssetTrendData",
        "Measurement/List/Value": "EXT_AssetTrendData",
        "Measurement/HealthInterval/": "EXT_ConfigureAssetHealth",
        "Asset/": "EXT_ViewAssetDetails",
        "ConditionIndex": "EXT_AssetConditionIndex",
        "EventLog": "EXT_AssetEventLog",
        "TimeBasedAnalytics": "EXT_TimeBasedAnalytics",
        "Sensor/Feature": "WEB_SensorFeatureConfiguration",
    }

    class KeyNames(BaseAPIClient.KeyNames):
        USER_CLAIMS = "user_claims"
        RAW_DATA_CSV = "RawDataCollectionCSV"
        RAW_DATA_BINARY = "RawDataCollectionResult"

    def __init__(self, url: str, **kwargs) -> None:
        """
        Initializes the SmartSensor API Client.
        Most of the initialization is done in the parent class BaseAPIClient

        Additionally, the specific SmartSensor API related attributes are initialized here.

        Args:
            url: Smart Sensor API Url

        Keyword Args:
            proxies (Dict): Dictionary with proxy configuration. Following keys are supported:
                `http_proxy`, `https_proxy`, `no_proxy`
            store_request (bool): If True, all obtained request-response pairs will be stored in memory.
                They might be dumped to a JSON file by calling `.dump_stored_requests` method.
                It is useful to collect the communication for the sake of regression tests.
            cache_file_path (Union[str | Path]): Path to the SQLite database file used as a permanent storage.
                If provided, then all request-response pairs will be cached in the SQLite DB.
                It is a useful feature for development purposes, but it's not recommended for production use
                since the cache is never invalidated.
            auto_refresh (bool): If True and `refresh_token` is present, then the client will automatically try
                to refresh the access token when it is about to expire (10 minutes before the deadline).
                Default: `True`

        """
        super().__init__(url=url, **kwargs)

        self.is_super_admin: Optional[bool] = None
        self.organization_id: Optional[int] = None
        self.user_id: Optional[int] = None
        self._organizations: List = []
        self._plants: Dict = {}
        self._units: Dict = {}
        self._token_refresh_dates: List = []

    @property
    def is_logged(self) -> bool:
        """
        Checks if the client is still logged in to the API.
        """

        _is_logged = (
            bool(self.auth_token)
            and isinstance(self.token_expiration_date, datetime.datetime)
            and (self.token_expiration_date > datetime.datetime.now(datetime.timezone.utc))
        )
        return _is_logged

    def logout(self):
        """
        Logs out the client from the API. It runs the parent class logout procedure fist,
        and then in resets the custom attributes of Smart Sensor API client
        """

        super().logout()
        self.user_id = None
        self.organization_id = None
        self.is_super_admin = None

    # type: ignore
    def authenticate(
        self,
        key: Optional[str] = None,
        username: Optional[str] = None,
        password: Optional[str] = None,
        access_token: Optional[str] = None,
        refresh_token: Optional[str] = None,
        id_token: Optional[str] = None,
        client_id: Optional[str] = None,
        client_secret: Optional[str] = None,
    ) -> bool:
        """
        Authenticates user based on OAuth2 using `/Auth/BearerOAuth2` endpoint.

        There are several available methods. They are supported in following order:

        1. API Key (key has to be provided)
        2. Username & Password (username and password has to be provided)
        3. Access Token (access_token has to be provided)
        4. ID Token (id_token and client_id has to be provided)
        5. Client ID & Client Secret (client_id and client_secret has to be provided)
        6. Refresh Token and Auth Token

        Args:
            key: Access API KEY generated on SmartSensorPortal
            username: Username (ABB email)
            password:  Password (ABB password)
            access_token: OAuth Access token from external identity provider. e.g. CIAM
            refresh_token: Refresh token
            id_token: OAuth ID token from external identity provider
            client_id: Client ID
            client_secret: Client Secret

        Returns:
            True if client is successfully authenticated and access token in obtained and set. Otherwise, False.
        """

        endpoint = "Auth/BearerOAuth2"

        #  1. API Key (key has to be provided)
        if key is not None:
            data = dict(grant_type="api_key", api_key=key)
        #  2. Username & Password (username and password has to be provided)
        elif (username is not None) and (password is not None):
            data = dict(grant_type="password", username=username, password=password)
        #  3. Access Token (access_token has to be provided)
        elif access_token is not None:
            data = dict(grant_type="access_token", access_token=access_token)
        #  4. ID Token (id_token and client_id has to be provided)
        elif id_token is not None and client_id is not None:
            data = dict(grant_type="id_token", id_token=id_token, client_id=client_id)
        # 5. Client ID & Client Secret (client_id and client_secret has to be provided)
        elif client_id is not None and client_secret is not None:
            data = dict(grant_type="client_credentials", client_id=client_id, client_secret=client_secret)
        # 6. Refresh Token
        elif refresh_token is not None and self.auth_token is not None:
            data = dict(grant_type="refresh_token", refresh_token=refresh_token, auth_token=self.auth_token)
        else:
            log.info(
                "Please use one of the following authentication methods:\n"
                "1. API Key (key has to be provided)\n"
                "2. Username & Password (username and password has to be provided)\n"
                "3. Access Token (access_token has to be provided)\n"
                "4. ID Token (id_token and client_id has to be provided)\n"
                "5. Client ID & Client Secret (client_id and client_secret has to be provided)\n"
                "6. Refresh token (refresh_token has to be provided and auth_token has to be set)"
            )
            return False

        response = self._make_request(method="POST", endpoint=endpoint, json_data=data)
        decoded_response: Dict = self._decode_response(response=response, default={})
        if not decoded_response:
            return False

        # Store authentication response
        self.auth_data[self.KeyNames.AUTH_RESPONSE] = decoded_response
        # Tokens
        self.auth_data[self.KeyNames.ACCESS_TOKEN] = decoded_response.get("access_token", None)
        self.auth_data[self.KeyNames.REFRESH_TOKEN] = decoded_response.get("refresh_token", None)

        # Token expiration date
        _expires_in = decoded_response.get("expires_in", 0)
        _expires_at = to_py_time(decoded_response.get("timestamp", None))
        if _expires_in and _expires_at:
            exp_date = _expires_at.replace(tzinfo=datetime.timezone.utc) + datetime.timedelta(seconds=_expires_in)
            self.auth_data[self.KeyNames.EXPIRATION_DATE] = exp_date

        # User data
        _user_claims = decoded_response.get("user_claims", {})
        self.auth_data[self.KeyNames.USER_CLAIMS] = _user_claims
        self.organization_id = _user_claims.get("OrganizationID", None)
        self.is_super_admin = _user_claims.get("IsSuperAdmin", False)
        self.user_id = _user_claims.get("UserID", None)

        return self.is_logged

    def set_auth_token(self, auth_token: str, use_api_data: bool = False, **kwargs) -> bool:
        """
        Sets the authorization token to SmartSensor API.

        To be used when authorization token is obtained from outside. Calling .authenticate() is not required
        when the auth token is present.

        Args:
            auth_token: Valid authorization token to SmartSensor API.
            use_api_data: If True then the extra data (refresh_token, is_super_admin, etc.) will be obtained from
                `User/Token` endpoint. Otherwise, it has to be provided in kwargs.

        Keyword Args:
            refresh_token (str): Valid refresh token
            user_id (int): User ID
            organization_id (int): Organization ID
            is_super_admin (bool): "Is Super Admin" flag from SmartSensor

        Returns:
            True if authorization token is set correctly, False otherwise.
        """

        try:
            decoded_token = jwt.decode(auth_token, options={"verify_signature": False})
            _exp_timestamp = float(decoded_token.get("exp"))  # type: ignore
            expiration_date = datetime.datetime.fromtimestamp(_exp_timestamp).replace(tzinfo=datetime.timezone.utc)
        except (ValueError, TypeError, jwt.exceptions.InvalidTokenError):
            log.error("Invalid token provided.")
            return False

        # Set new token ...
        self.auth_data[self.KeyNames.ACCESS_TOKEN] = auth_token
        self.auth_data[self.KeyNames.EXPIRATION_DATE] = expiration_date

        # ... and other extra parameters
        if use_api_data:
            auth_details = self.get_auth_details()
            self.auth_data[self.KeyNames.REFRESH_TOKEN] = auth_details.get("refreshToken")
            self.is_super_admin = auth_details.get("isSuperAdmin", False)
            self.user_id = auth_details.get("userID", None)
            self.organization_id = auth_details.get("organizationID", None)
        else:
            self.auth_data[self.KeyNames.REFRESH_TOKEN] = kwargs.get("refresh_token")
            self.is_super_admin = kwargs.get("is_super_admin", False)
            self.user_id = kwargs.get("user_id", None)
            self.organization_id = kwargs.get("organization_id", None)
        return self.is_logged

    def _make_access_token_refresh(self, force: bool = False, refresh_token: Optional[str] = None) -> bool:
        """
        Refreshes the access token using the existing refresh token.

        Because it overrides the base class method, it will be called by `_process_request` method just before calling
        `_send_request` method.

        In default mode, it will only try to refresh the token if two conditions are met:

        1. `auto_refresh` parameter is True
        2. Token's remaining time is less than auto_refresh_delta (10 min by default)

        Token refresh can be forced by calling the function with force = True.

        Args:
            force: If true then the token refresh process is executed regardless of the above conditions.
            refresh_token: Custom refresh token to be used. If not provided, then the `self.refresh_token` property
                is used.

        Returns:
            True, if the token refresh process is successful. False, otherwise.
        """
        _auto_refresh_delta = datetime.timedelta(minutes=10)

        # Check required inputs
        _refresh_token = refresh_token if refresh_token else self.refresh_token
        if not (_refresh_token and self.auth_token):
            return False
        # Check if it's time to refresh
        _expiration_date = self.auth_data.get(self.KeyNames.EXPIRATION_DATE)
        if not isinstance(_expiration_date, datetime.datetime):
            return False
        _time_to_expire = _expiration_date - datetime.datetime.now(datetime.timezone.utc)
        is_time_to_refresh = force or (self._auto_refresh and _time_to_expire < _auto_refresh_delta)
        if not is_time_to_refresh:
            return False

        # Finally perform the token refreshment
        _status = self.authenticate(refresh_token=_refresh_token)
        if _status:
            self._token_refresh_dates.append(datetime.datetime.now(datetime.timezone.utc))
            log.debug(f"Access token has been refreshed. New expiration date is: {self.token_expiration_date}")
        return _status

    def get_auth_details(self) -> Dict:
        """
        Gets authentication information for a currently used access token.

        Returns:
            Dictionary with auth details. Default: empty dict.
        """
        response = self._make_request(method="GET", endpoint="/Auth/Token")
        decoded_response: Dict = self._decode_response(response=response, default={})
        return decoded_response

    def get_user_data(self) -> Dict:
        """
        Gets the data of currently logged user.

        Returns:
            Dictionary with user data. Default: empty dict.
        """
        endpoint = "User"
        response = self._make_request(method="GET", endpoint=endpoint)
        decoded_response: Dict = self._decode_response(response=response, default={})
        return decoded_response

    def get_organizations_list(self, use_organization_history: bool = True) -> List[Dict]:
        """
        Gets the list of organizations.
        For SuperAdmin users `/Organization` endpoint is used. For standard users the list of available organizations
        is read from `User/OrganizationHistory` endpoint.

        Args:
            use_organization_history: If true, then the list of available organizations for standard users
                is read from `User/OrganizationHistory` endpoint. Otherwise, `Organization` endpoint is always used.
        """
        _endpoint = (
            "Organization" if (self.is_super_admin or not use_organization_history) else "User/OrganizationHistory"
        )

        response = self._make_request(method="GET", endpoint=_endpoint)
        decoded_response: List = self._decode_response(response=response, default=[])

        # Store the results locally for organization name look-up
        self._organizations = decoded_response
        return decoded_response

    def get_organization_name(self, organization_id: int) -> Optional[str]:
        """
        Get the working organization name based on provided organization ID

        Args:
            organization_id: ID of organization

        Returns:
            Organization name if found. Otherwise, `None` is returned.
        """

        def _search_organization_in_user_history(_id: int) -> Dict:
            organizations = self.get_organizations_list(use_organization_history=True)
            # Find the organization data by its id
            for org_data in organizations:
                if org_data.get("organizationID", None) == _id:
                    return org_data
            return {}

        # super admins can use /Organization/{id} endpoint which is much faster than listing all organizations and
        # searching those organizations by id
        # regular users cannot do that, so we will be searching in user organization history it this case
        if self.is_super_admin:
            organization_data = self.get_organization_data(organization_id=organization_id)
        else:
            organization_data = _search_organization_in_user_history(_id=organization_id)

        if not organization_data:
            log.debug(f"Organization (id={organization_id}) not found")
            return None

        organization_name = organization_data.get("organizationName", None)
        return organization_name

    def set_organization_id(self, organization_id: int) -> bool:
        """
        Sets the new working organization for a current user, based on provided organization ID.

        Args:
            organization_id: ID of organization to be switched to.

        Returns:
            True if the organization change was successful. Otherwise, false.
        """

        organization_name = self.get_organization_name(organization_id)
        if not organization_name:
            # Return false if there is no matching organization ID
            return False

        return self.set_organization(organization_name)

    def set_organization(self, organization_name: str) -> bool:
        """
        Sets the new working organization for a current user, based on provided organization ID.

        Args:
            organization_name: Name of organization to be switched to.

        Returns:
            True if the organization change was successful. Otherwise, false.
        """

        data = {"organizationName": organization_name}
        response = self._make_request(method="PUT", endpoint="User/Organization", json_data=data)
        decoded_response: Dict = self._decode_response(response=response, default={})
        status_code = self._get_status_code(response=response)
        _new_org_id = decoded_response.get("organizationID", None)
        if (status_code == HTTPStatus.OK) and _new_org_id:
            self.organization_id = _new_org_id
            return True
        else:
            return False

    def get_plants_list(self, organization_id: Optional[int] = None) -> List[Dict]:
        """
        Gets a list of all plants in the organization.
        If no organization id is provided, then the currently selected one is used.

        Args:
            organization_id: ID of an organization

        Returns:
            List of plants' data. If nothing found, the empty list is returned.

        """
        endpoint = "Plant"
        if organization_id is None:
            log.debug(f"Using default Organization(id={self.organization_id})")
            organization_id = self.organization_id

        params = {"organizationID": organization_id}
        response = self._make_request(method="GET", endpoint=endpoint, params=params)
        decoded_response: List = self._decode_response(response=response, default=[])
        # Store plants' data in the `_plants[org_id]` attribute.
        self._plants[organization_id] = decoded_response
        return decoded_response

    def get_asset_details(self, asset_id: Union[str, int]) -> Dict:
        """
        Gets detailed asset info using `Asset/{id}` endpoint

        Args:
            asset_id: ID of an asset

        Returns:
            Asset's details. If not found, empty dict is returned.
        """
        endpoint = f"Asset/{asset_id}"
        response = self._make_request(method="GET", endpoint=endpoint, feature_code="EXT_ViewAssetDetails")
        decoded_response: Dict = self._decode_response(response=response, default={})
        return decoded_response

    def get_asset_data(self, asset_id: Union[str, int]) -> Dict:
        """
        Gets asset's data using `Asset/Data{id}` endpoint

        Args:
            asset_id: ID of an asset

        Returns:
            Asset's data. If not found, empty dict is returned.
        """
        endpoint = f"Asset/Data/{asset_id}"
        response = self._make_request(method="GET", endpoint=endpoint)
        decoded_response: Dict = self._decode_response(response=response, default={})
        return decoded_response

    def get_asset_data_by_sensor_id(self, sensor_id: str) -> Dict:
        """
        Gets asset's data using `Asset/Sensor` endpoint. Sensor ID is used as an identifier.

        Args:
            sensor_id: ID of a sensor.

        Returns:
            Asset's data. If not found, empty dict is returned.
        """

        endpoint = "Asset/Sensor"
        params = {"identifier": sensor_id}
        response = self._make_request(method="GET", endpoint=endpoint, params=params)
        decoded_response: Dict = self._decode_response(response=response, default={})
        return decoded_response

    def get_asset_device(self, asset_id: Union[str, int]) -> Dict:
        """
        Gets Asset's device data using `Asset/Device/{id}` endpoint.

        Args:
            asset_id: ID of an asset.

        Returns:
            Asset's device data. If not found, empty dict is returned.
        """
        endpoint = f"Asset/Device/{asset_id}"
        response = self._make_request(method="GET", endpoint=endpoint)
        decoded_response: Dict = self._decode_response(response=response, default={})
        return decoded_response

    def get_asset_properties(self, asset_id: Union[str, int]) -> List[Dict]:
        """
        Gets Asset properties list using `Asset/Property/{id}` endpoint.

        Args:
            asset_id: ID of an asset.

        Returns:
            List of asset properties. If nothing found, the empty list is returned.
        """

        endpoint = f"Asset/Property/{asset_id}"
        response = self._make_request(method="GET", endpoint=endpoint)
        decoded_response: List = self._decode_response(response=response, default=[])
        return decoded_response

    def find_asset_id(self, identifier: Union[int, str]) -> Optional[int]:
        """
        Gets Asset ID based on provided identifier. Identifier might be either 'Asset ID' (then it is only validated)
        or a 'Sensor ID' (then the `.get_asset_data_by_sensor_id method` is used)

        Args:
            identifier: Asset ID or Sensor ID

        Returns:
            Asset ID if asset is found. Otherwise, `None` is returned.
        """
        # Try by Asset ID first
        response = self.get_asset_data(asset_id=identifier)
        asset_id = self._check_response_for_asset_id(response=response)
        if asset_id:
            return asset_id
        # Try by Sensor ID
        response = self.get_asset_data_by_sensor_id(sensor_id=str(identifier))
        asset_id = self._check_response_for_asset_id(response=response)
        return asset_id

    @staticmethod
    def _check_response_for_asset_id(response: Optional[Dict]) -> Optional[int]:
        """Checks the response dictionary for the presence of asset id key"""
        if isinstance(response, dict) and "assetID" in response:
            asset_id = response.get("assetID")
            if isinstance(asset_id, int):
                return asset_id
        return None

    def get_sensor_details(self, sensor_identifier: str, sensor_type_id: int) -> Dict:
        """
        Gets the sensor details using the `Sensor/Details/{id}` endpoint

        Args:
            sensor_identifier: Sensor ID
            sensor_type_id: Sensor Type ID

        Returns:
            Dictionary with sensor details. If nothing found, the empty dict is returned.

        """
        endpoint = f"Sensor/Details/{sensor_identifier}"
        params = {"sensorTypeID": sensor_type_id}
        response = self._make_request(method="GET", endpoint=endpoint, params=params)
        decoded_response: Dict = self._decode_response(response=response, default={})
        return decoded_response

    def get_sensor_properties(self, sensor_identifier: str, sensor_type_id: int) -> List[Dict]:
        """
         Gets sensor properties, as a subset of sensor details.

         Args:
            sensor_identifier: Sensor ID
            sensor_type_id: Sensor Type ID

        Returns:
            List with sensor properties. If nothing found, the empty list is returned.


        """
        sensor_details = self.get_sensor_details(sensor_identifier=sensor_identifier, sensor_type_id=sensor_type_id)
        if "properties" in sensor_details:
            return sensor_details["properties"]
        return []

    def get_sensor_feature(
        self,
        sensor_identifier: str,
        feature_types: str,
        sensor_type_id: int,
        start_date: Optional[datetime.datetime] = None,
        end_date: Optional[datetime.datetime] = None,
    ) -> List[Dict]:
        """
        Gets the sensor feature using `Sensor/Feature/{id}` endpoint

        Args:
            sensor_identifier: Sensor ID
            sensor_type_id: Sensor Type ID
            feature_types: Feature types, As a string with comma separated list of features.
            start_date: Start Date. Used with time based features.
            end_date: End Date. User with time based features.

        Returns:
            List of features. If not found, then the empty list is returned.

        """

        endpoint = f"Sensor/Feature/{sensor_identifier}"
        params = {"sensorTypeID": sensor_type_id, "featureTypes": feature_types}
        if start_date is not None:
            params["from"] = to_api_time(start_date)
        if end_date is not None:
            params["to"] = to_api_time(end_date)

        response = self._make_request(method="GET", endpoint=endpoint, params=params)
        decoded_response: List = self._decode_response(response=response, default=[])
        return decoded_response

    def get_sensor_latest_features(
        self,
        sensor_identifier: str,
        sensor_type_id: int,
        feature_types: str = "",
        include_complex_objects: bool = False,
    ) -> List[Dict]:
        """
        Gets the list of the latest features using `Sensor/Feature/Value/{id}` endpoint

        Args:
            sensor_identifier: Sensor ID
            sensor_type_id: Sensor Type ID
            feature_types: Feature types, As a string with comma separated list of features.
            include_complex_objects: If True, then the complex objects (e.g. raw data content) will be included
                in the response, otherwise only the headers of those complex objects are present. Default: `False`

        Returns:
            List of the latest sensor features. If not found then the empty list is returned.

        """
        endpoint = f"Sensor/Feature/Value/{sensor_identifier}"
        params = {
            "sensorTypeID": sensor_type_id,
            "featureTypes": feature_types,
            "complexObject": include_complex_objects,
        }
        response = self._make_request(method="GET", endpoint=endpoint, params=params)
        decoded_response: List = self._decode_response(response=response, default=[])
        return decoded_response

    def send_sensor_feature(
        self, sensor_identifier: str, sensor_type_id: int, features: List[Dict], method: MethodType = "POST"
    ) -> bool:
        """
        Sends the sensor feature to SmartSensor API using `Sensor/Feature/{id}` endpoint

        Args:
            sensor_identifier: Sensor ID
            sensor_type_id: Sensor Type ID
            features: List of features to send.
            method: HTTP method to be used while sending the features. Default method: POST

        Returns:
            True if sending was successful. Otherwise, `False` is returned.

        """

        endpoint = "Sensor/Feature"
        data = {"sensorIdentifier": sensor_identifier, "sensorTypeID": sensor_type_id, "features": features}
        response = self._make_request(method=method, endpoint=endpoint, json_data=data)
        status_code = self._get_status_code(response=response)
        _is_successful = status_code == HTTPStatus.OK
        return _is_successful

    def get_assets_list(
        self,
        organization_id: Optional[int] = None,
        plant_id: Optional[int] = None,
        asset_group_id: Optional[int] = None,
    ) -> List[Dict]:
        """
        Gets the list of assets in organization.

        If no organization id is provided, then the currently selected one is used.

        Optionally, the list might be filtered down by providing Plant ID or Asset Group ID.

        Args:
            organization_id: ID of organization.
            plant_id: ID of Plant
            asset_group_id: ID of Asset Group.

        Returns:
            List of assets in organization. If nothing is found, then the empty list is returned.
        """

        endpoint = "Asset/List"
        params = {"organizationID": organization_id if organization_id is not None else self.organization_id}
        if plant_id is not None:
            params["plantID"] = plant_id
        if asset_group_id is not None:
            params["assetGroupID"] = asset_group_id

        response = self._make_request(method="GET", endpoint=endpoint, params=params)
        decoded_response: List = self._decode_response(response=response, default=[])
        return decoded_response

    def get_assets_details_list(self, organization_id: int) -> List[Dict]:
        """
        Gets detailed asset information from all the accessible assets in the specified organization

        Args:
            organization_id: ID of organization.

        Returns:
            List of asset details in organization. If nothing is found, then the empty list is returned.

        """
        endpoint = "Asset"
        params = {"organizationID": organization_id}
        response = self._make_request(
            method="GET", endpoint=endpoint, params=params, feature_code="EXT_ViewAssetDetails"
        )
        decoded_response: List = self._decode_response(response=response, default=[])
        return decoded_response

    def get_asset_measurement(self, asset_id: Union[str, int]) -> List[Dict]:
        """
        Gets a collection of the latest measurements from a specified Asset using `Asset/Measurement/{id}` endpoint

        Args:
            asset_id: ID of an Asset.

        Returns:
            List of the latest measurements. If not found, then the empty list is returned.

        """
        endpoint = f"Asset/Measurement/{asset_id}"
        response = self._make_request(method="GET", endpoint=endpoint)
        decoded_response: List = self._decode_response(response=response, default=[])
        return decoded_response

    def get_measurements_for_single_asset(
        self,
        asset_id: Union[str, int],
        measurement_ids: Union[Sequence[int], Sequence[str], int, str],
        start_date: datetime.datetime,
        end_date: datetime.datetime,
    ) -> List[Dict]:
        """
        Gets measurement data for a single asset in a provided time range.

        It is using `Measurement/Value` endpoint, that allows data retrieval for a single asset only.

        To get the data for several assets at once use `get_measurements_for_many_assets`

        Args:
            asset_id: ID of an asset.
            measurement_ids: IDs of the measurements that should be received.
            start_date: Start date of the measurements.
            end_date: End date of the measurements.

        Returns:
            List of measurement data for single asset. If not found then the empty list is returned.
        """
        endpoint = "Measurement/Value"
        parsed_measurement_ids = parse_measurement_ids(measurement_ids)
        if parsed_measurement_ids is None:
            return []
        params = {
            "assetID": int(asset_id),
            "measurementTypes": parsed_measurement_ids,
            "from": to_api_time(start_date),
            "to": to_api_time(end_date),
        }
        response = self._make_request(method="GET", endpoint=endpoint, params=params)
        decoded_response: List = self._decode_response(response=response, default=[])
        return decoded_response

    def get_measurements_for_multiple_assets(
        self,
        asset_ids: Union[Sequence[str], Sequence[int]],
        measurement_ids: Union[Sequence[int], Sequence[str]],
        start_date: datetime.datetime,
        end_date: datetime.datetime,
        delta: datetime.timedelta = datetime.timedelta(days=90),
    ) -> List:
        """
        Gets measurement data for multiple assets in a provided time range.

        It is using POST `Measurement/List/Value` endpoint, that allows data retrieval for a multiple asset ids
        at once.

        This highly increases the performance, and should be used whenever it's possible.

        Args:
            asset_ids: List of asset ids to get the measurements for. Provided as a list of int
                or comma-separated string.
            measurement_ids: IDs of the measurements that should be received.
            start_date: Start date of the measurements.
            end_date: End date of the measurements.
            delta: Maximum time range to be requested in a single request. Due to SmartSensor API limitations
                only requests with time range less than 90 days are accepted. This method provides automatic support
                for greater time ranges, but the max delta for a single request has to be provided. Default: 90 days.

        Returns:
            List of measurement data for many assets. If not found then the empty list is returned.

        """
        endpoint = "Measurement/List/Value"
        if not measurement_ids:
            return []

        sub_time_ranges = split_time_range(start_date=start_date, end_date=end_date, delta=delta)
        responses = []
        for sub_start_date, sub_end_date in sub_time_ranges:
            log.debug(f"Collecting data for time range: {sub_start_date} / {sub_end_date}")
            params = {"from": to_api_time(sub_start_date), "to": to_api_time(sub_end_date)}
            data = {
                "assetIDList": sorted([str(asset_id) for asset_id in asset_ids]),
                "measurementTypes": sorted([str(measurement_id) for measurement_id in measurement_ids]),
            }
            response = self._make_request(method="POST", endpoint=endpoint, params=params, json_data=data)
            decoded_response: List = self._decode_response(response=response, default=[])
            if not decoded_response:
                continue
            responses.append(decoded_response)

        if not responses:
            return []
        merged_response = merge_measurement_value_responses(responses)
        return merged_response

    def get_measurement_threshold(
        self, asset_id: Union[str, int], measurement_ids: Union[Sequence[int], Sequence[str], int, str]
    ) -> List[Dict]:
        """
        Gets threshold values for a given measurement type and asset id,
        using `Measurement/HealthInterval/{id}` endpoint.

        Args:
            asset_id: ID of an asset.
            measurement_ids: IDs of measurements.

        Returns:
            List of measurement thresholds. If not found then the empty list is returned.
        """

        endpoint = f"Measurement/HealthInterval/{asset_id}"
        parsed_measurement_ids = parse_measurement_ids(measurement_ids)
        if parsed_measurement_ids is None:
            return []
        params = {"measurementTypes": parsed_measurement_ids}
        response = self._make_request(method="GET", endpoint=endpoint, params=params)
        decoded_response: List = self._decode_response(response=response, default=[])
        return decoded_response

    def get_measurement_units(self, force: bool = False) -> Dict:
        """
        Gets the list of units for all available measurements types, using `Measurement/Unit` endpoint.
        Due to the static nature of this data, by default it is read only once from the API. Any subsequent requests
        are dismissed, and the first response is returned instead. To override it, use `force=True` argument.

        Args:
            force: If true then each method call is calling the API. Otherwise, only the first one does, and subsequent
                calls are serverd with the first response.

        Returns:
            Dictionary with measurement unit data. If not found, then the empty dict is returned.

        """
        endpoint = "Measurement/Unit"

        if self._units and not force:
            return self._units

        response = self._make_request(method="GET", endpoint=endpoint)
        decoded_response: List = self._decode_response(response=response, default=[])
        for element in decoded_response:
            _group_id = element.get("measurementUnitGroup", {}).get("measurementUnitGroupID", None)
            _list = element.get("measurementUnits", None)
            if (_group_id is None) or (_list is None):
                continue
            _single_unit_dict = {}
            for _unit_definition in _list:
                _standard = _unit_definition.get("measurementUnitStandard", "").casefold()
                _symbol = _unit_definition.get("measurementUnitSymbol", "")
                _factor = _unit_definition.get("measurementUnitConversionRate", 1.0)
                _offset = _unit_definition.get("measurementUnitConversionOffset", 0.0)
                if _standard == "":
                    continue
                _single_unit_dict[_standard] = {"symbol": _symbol, "factor": _factor, "offset": _offset}
                # MANUAL PATCH for Fahrenheits
                # Since Smart Sensor API provides `ConversionRate` only, it is impossible to convert from °C to °F
                # To solve this problem, offset field was created, and appropriate values were added manually
                if _standard == "imperial" and _group_id == 4:
                    _single_unit_dict[_standard] = {"symbol": "°F", "factor": 0.5555, "offset": 32.0}
            self._units[_group_id] = _single_unit_dict
        return self._units

    def get_measurement_unit(self, measurement_unit_group: int, standard: str = "metric") -> Dict:
        """
        Gets the unit data for a specific measurement id. The output comes from the `get_measurement_units` response.

        Args:
            measurement_unit_group: ID of a measurement group.
            standard: Unit standard. Supported: `metric` and `imperial`. Default: `metric

        Returns:
            Unit data. If not found then the empty dict is returned.


        """
        no_unit_response = {"symbol": "", "factor": 1.0, "offset": 0.0}
        measurement_units = self.get_measurement_units()

        single_unit_dict = measurement_units.get(measurement_unit_group, {})
        # Default standard is `metric` standard
        default_standard_dict = single_unit_dict.get("metric", no_unit_response)
        standard_dict = single_unit_dict.get(standard.casefold(), default_standard_dict)
        return standard_dict

    def get_organization_data(self, organization_id: int) -> Dict:
        """
        Gets the data about the organization. It only works for super admins.

        Args:
            organization_id: ID of an organization.

        Returns:
            Organization data if found. Otherwise, Empty dict.

        """
        endpoint = f"Organization/{organization_id}"
        response = self._make_request(method="GET", endpoint=endpoint)
        decoded_response: Dict = self._decode_response(response=response, default={})
        return decoded_response

    def get_event_logs(
        self,
        event_log_type: int,
        filter_closed: Optional[bool] = None,
        closing_reason: Optional[int] = None,
        asset_id: Optional[Union[str, int]] = None,
        start_date: Optional[datetime.datetime] = None,
        end_date: Optional[datetime.datetime] = None,
        number_of_events: Optional[int] = None,
        organization_id: Optional[int] = None,
        asset_type_id: Optional[int] = None,
        group_events: bool = False,
        sub_type_id: Optional[int] = None,
    ) -> List:
        """
        Gets the event logs of the specified organization as a list. Different filtering options can be chosen.
        """
        endpoint = "EventLog"
        params = {
            "eventLogType": event_log_type,
            "filterClosed": filter_closed,
            "closingReason": closing_reason,
            "assetID": asset_id,
            "numberOfEvents": number_of_events,
            "organizationID": organization_id,
            "assetTypeID": asset_type_id,
            "groupEvents": group_events,
            "subTypeID": sub_type_id,
        }
        if start_date is not None:
            params["from"] = to_api_time(start_date)
        if end_date is not None:
            params["to"] = to_api_time(end_date)

        response = self._make_request(method="GET", endpoint=endpoint, params=params)
        decoded_response: List = self._decode_response(response=response, default=[])
        return decoded_response

    def get_asset_health(
        self,
        plant_id: Optional[int] = None,
        asset_group_id: Optional[int] = None,
        organization_id: Optional[int] = None,
        assets_without_group: bool = False,
    ) -> List:
        """
        Gets a list of assets with health information, using `Asset/Health` endpoint.

        Args:
            plant_id: ID of a Plant
            asset_group_id: ID of an Asset Group.
            organization_id: ID of an organization.
            assets_without_group: If true, assets that don't belong to any group will be included in the response.
                Otherwise, only the assets that belong to any Asset Group will be included.

        Returns:
            List of assets health information. If not found then the empty list is returned.
        """

        endpoint = "Asset/Health"
        params = {
            "plantID": plant_id,
            "assetGroupID": asset_group_id,
            "organizationID": organization_id,
            "assetsWithoutGroup": assets_without_group,
        }
        response = self._make_request(method="GET", endpoint=endpoint, params=params)
        decoded_response: List = self._decode_response(response=response, default=[])
        return decoded_response

    def get_plant_data(self, plant_id: int) -> Dict:
        """
        Gets the plant data, using `Plant/{id}` endpoint

        Args:
            plant_id: ID of a Plant

        Returns:
            Plant data. If not found then the empty dict is returned.
        """
        endpoint = f"Plant/{plant_id}"
        response = self._make_request(method="GET", endpoint=endpoint)
        decoded_response: Dict = self._decode_response(response=response, default={})
        return decoded_response

    def get_raw_data_for_asset(
        self,
        asset_id: int,
        mode: Literal["binary", "csv"] = "csv",
        start_date: Optional[datetime.datetime] = None,
        end_date: Optional[datetime.datetime] = None,
        fetch_files: bool = True,
    ) -> List[Dict]:
        """
        Gets list of raw data objects. It accepts two modes: "csv" and "binary".

        If `start_date` and `end_date` are provided, it returns all objects found between those days,
        otherwise only the latest raw data object is returned.

        If `fetch_files` parameter is set to False while the start and end dates are provided then the returned
        list of raw data objects doesn't contain the actual contend of raw data file, but only list their names.
        That is helpful when you want only to list all available raw data, but you don't need to download it.

        Args:
            asset_id: ID of an asset.
            mode: Raw data mode. Default: `csv`
            start_date: Start date
            end_date: End date
            fetch_files: If True the actual content of raw data is included in the response. Otherwise, only the
                file names are included.

        Returns:
            List of "Raw Data" data for a given asset. If nothing is found, then the empty list is returned.

        """
        params: Dict[str, Union[str, int]] = {"assetID": asset_id}
        _type = self.KeyNames.RAW_DATA_BINARY if mode.casefold() == "binary" else self.KeyNames.RAW_DATA_CSV

        if isinstance(start_date, datetime.datetime) and isinstance(end_date, datetime.datetime):
            endpoint = f"TimeBasedAnalytics/KPI/Values/RawData/{_type}"
            params["from"] = to_api_time(start_date)
            params["to"] = to_api_time(end_date)
            params["fetchFileContents"] = fetch_files
        else:
            endpoint = f"TimeBasedAnalytics/KPI/RawData/{_type}"

        response = self._make_request(method="GET", endpoint=endpoint, params=params)
        decoded_response: List = self._decode_response(response=response, default=[])
        return decoded_response

    def send_report_confirmation(self, token: str) -> bool:
        """
        Sends a confirmation to the `Report/Completed` endpoint, when the report generation process is ready.

        This function can be used in ReportAzureFunction to confirm when the report generation process is done.

        Args:
            token: Report generation process token.

        Returns:
            Delivery status as a bool. True if the response status code was 200, False otherwise.
        """

        endpoint = "Report/Completed"
        params = {"token": token}
        response = self._make_request(method="GET", endpoint=endpoint, params=params)
        status_code = self._get_status_code(response=response)
        is_status_code = status_code == HTTPStatus.OK
        return is_status_code

    def get_sites_list(self, organization_id: int) -> List[Dict]:
        """
        Gets the list of sites from the organization identified by organizationID.

        Args:
            organization_id: ID of an organization.

        Returns:
            List of dictionaries containing information about Sites that belong to the organization.
        """

        endpoint = "Site"
        params = {"organizationID": organization_id}

        response = self._make_request(method="GET", endpoint=endpoint, params=params)
        decoded_response: List = self._decode_response(response=response, default=[])
        return decoded_response

    def get_bearing_info(self, bearing_number: str, manufacturer: Optional[str] = None) -> List[Dict]:
        """
        Gets the list of the bearing information.

        Args:
            bearing_number: Bearing number.
            manufacturer: Manufacturer name. Optional.

        Returns:
            List of dictionaries containing information about bearings.
        """

        endpoint = f"/AssetType/BearingInfo/{bearing_number}"
        params = {"manufacturer": manufacturer} if manufacturer else None

        response = self._make_request(method="GET", endpoint=endpoint, params=params)
        decoded_response: List = self._decode_response(response=response, default=[])
        return decoded_response

is_logged property

Checks if the client is still logged in to the API.

__init__(url, **kwargs)

Initializes the SmartSensor API Client. Most of the initialization is done in the parent class BaseAPIClient

Additionally, the specific SmartSensor API related attributes are initialized here.

Parameters:

Name	Type	Description	Default
url	str	Smart Sensor API Url	required

Other Parameters:

Name	Type	Description
proxies	Dict	Dictionary with proxy configuration. Following keys are supported: http_proxy, https_proxy, no_proxy
store_request	bool	If True, all obtained request-response pairs will be stored in memory. They might be dumped to a JSON file by calling .dump_stored_requests method. It is useful to collect the communication for the sake of regression tests.
cache_file_path	Union[str | Path]	Path to the SQLite database file used as a permanent storage. If provided, then all request-response pairs will be cached in the SQLite DB. It is a useful feature for development purposes, but it's not recommended for production use since the cache is never invalidated.
auto_refresh	bool	If True and refresh_token is present, then the client will automatically try to refresh the access token when it is about to expire (10 minutes before the deadline). Default: True

Source code in reportconnectors/api_client/smartsensor_api_client.py

def __init__(self, url: str, **kwargs) -> None:
    """
    Initializes the SmartSensor API Client.
    Most of the initialization is done in the parent class BaseAPIClient

    Additionally, the specific SmartSensor API related attributes are initialized here.

    Args:
        url: Smart Sensor API Url

    Keyword Args:
        proxies (Dict): Dictionary with proxy configuration. Following keys are supported:
            `http_proxy`, `https_proxy`, `no_proxy`
        store_request (bool): If True, all obtained request-response pairs will be stored in memory.
            They might be dumped to a JSON file by calling `.dump_stored_requests` method.
            It is useful to collect the communication for the sake of regression tests.
        cache_file_path (Union[str | Path]): Path to the SQLite database file used as a permanent storage.
            If provided, then all request-response pairs will be cached in the SQLite DB.
            It is a useful feature for development purposes, but it's not recommended for production use
            since the cache is never invalidated.
        auto_refresh (bool): If True and `refresh_token` is present, then the client will automatically try
            to refresh the access token when it is about to expire (10 minutes before the deadline).
            Default: `True`

    """
    super().__init__(url=url, **kwargs)

    self.is_super_admin: Optional[bool] = None
    self.organization_id: Optional[int] = None
    self.user_id: Optional[int] = None
    self._organizations: List = []
    self._plants: Dict = {}
    self._units: Dict = {}
    self._token_refresh_dates: List = []

authenticate(key=None, username=None, password=None, access_token=None, refresh_token=None, id_token=None, client_id=None, client_secret=None)

Authenticates user based on OAuth2 using /Auth/BearerOAuth2 endpoint.

There are several available methods. They are supported in following order:

1. API Key (key has to be provided)
2. Username & Password (username and password has to be provided)
3. Access Token (access_token has to be provided)
4. ID Token (id_token and client_id has to be provided)
5. Client ID & Client Secret (client_id and client_secret has to be provided)
6. Refresh Token and Auth Token

Parameters:

Name	Type	Description	Default
key	Optional[str]	Access API KEY generated on SmartSensorPortal	None
username	Optional[str]	Username (ABB email)	None
password	Optional[str]	Password (ABB password)	None
access_token	Optional[str]	OAuth Access token from external identity provider. e.g. CIAM	None
refresh_token	Optional[str]	Refresh token	None
id_token	Optional[str]	OAuth ID token from external identity provider	None
client_id	Optional[str]	Client ID	None
client_secret	Optional[str]	Client Secret	None

Returns:

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

Source code in reportconnectors/api_client/smartsensor_api_client.py

def authenticate(
    self,
    key: Optional[str] = None,
    username: Optional[str] = None,
    password: Optional[str] = None,
    access_token: Optional[str] = None,
    refresh_token: Optional[str] = None,
    id_token: Optional[str] = None,
    client_id: Optional[str] = None,
    client_secret: Optional[str] = None,
) -> bool:
    """
    Authenticates user based on OAuth2 using `/Auth/BearerOAuth2` endpoint.

    There are several available methods. They are supported in following order:

    1. API Key (key has to be provided)
    2. Username & Password (username and password has to be provided)
    3. Access Token (access_token has to be provided)
    4. ID Token (id_token and client_id has to be provided)
    5. Client ID & Client Secret (client_id and client_secret has to be provided)
    6. Refresh Token and Auth Token

    Args:
        key: Access API KEY generated on SmartSensorPortal
        username: Username (ABB email)
        password:  Password (ABB password)
        access_token: OAuth Access token from external identity provider. e.g. CIAM
        refresh_token: Refresh token
        id_token: OAuth ID token from external identity provider
        client_id: Client ID
        client_secret: Client Secret

    Returns:
        True if client is successfully authenticated and access token in obtained and set. Otherwise, False.
    """

    endpoint = "Auth/BearerOAuth2"

    #  1. API Key (key has to be provided)
    if key is not None:
        data = dict(grant_type="api_key", api_key=key)
    #  2. Username & Password (username and password has to be provided)
    elif (username is not None) and (password is not None):
        data = dict(grant_type="password", username=username, password=password)
    #  3. Access Token (access_token has to be provided)
    elif access_token is not None:
        data = dict(grant_type="access_token", access_token=access_token)
    #  4. ID Token (id_token and client_id has to be provided)
    elif id_token is not None and client_id is not None:
        data = dict(grant_type="id_token", id_token=id_token, client_id=client_id)
    # 5. Client ID & Client Secret (client_id and client_secret has to be provided)
    elif client_id is not None and client_secret is not None:
        data = dict(grant_type="client_credentials", client_id=client_id, client_secret=client_secret)
    # 6. Refresh Token
    elif refresh_token is not None and self.auth_token is not None:
        data = dict(grant_type="refresh_token", refresh_token=refresh_token, auth_token=self.auth_token)
    else:
        log.info(
            "Please use one of the following authentication methods:\n"
            "1. API Key (key has to be provided)\n"
            "2. Username & Password (username and password has to be provided)\n"
            "3. Access Token (access_token has to be provided)\n"
            "4. ID Token (id_token and client_id has to be provided)\n"
            "5. Client ID & Client Secret (client_id and client_secret has to be provided)\n"
            "6. Refresh token (refresh_token has to be provided and auth_token has to be set)"
        )
        return False

    response = self._make_request(method="POST", endpoint=endpoint, json_data=data)
    decoded_response: Dict = self._decode_response(response=response, default={})
    if not decoded_response:
        return False

    # Store authentication response
    self.auth_data[self.KeyNames.AUTH_RESPONSE] = decoded_response
    # Tokens
    self.auth_data[self.KeyNames.ACCESS_TOKEN] = decoded_response.get("access_token", None)
    self.auth_data[self.KeyNames.REFRESH_TOKEN] = decoded_response.get("refresh_token", None)

    # Token expiration date
    _expires_in = decoded_response.get("expires_in", 0)
    _expires_at = to_py_time(decoded_response.get("timestamp", None))
    if _expires_in and _expires_at:
        exp_date = _expires_at.replace(tzinfo=datetime.timezone.utc) + datetime.timedelta(seconds=_expires_in)
        self.auth_data[self.KeyNames.EXPIRATION_DATE] = exp_date

    # User data
    _user_claims = decoded_response.get("user_claims", {})
    self.auth_data[self.KeyNames.USER_CLAIMS] = _user_claims
    self.organization_id = _user_claims.get("OrganizationID", None)
    self.is_super_admin = _user_claims.get("IsSuperAdmin", False)
    self.user_id = _user_claims.get("UserID", None)

    return self.is_logged

set_auth_token(auth_token, use_api_data=False, **kwargs)

Sets the authorization token to SmartSensor API.

To be used when authorization token is obtained from outside. Calling .authenticate() is not required when the auth token is present.

Parameters:

Name	Type	Description	Default
auth_token	str	Valid authorization token to SmartSensor API.	required
use_api_data	bool	If True then the extra data (refresh_token, is_super_admin, etc.) will be obtained from User/Token endpoint. Otherwise, it has to be provided in kwargs.	False

Other Parameters:

Name	Type	Description
refresh_token	str	Valid refresh token
user_id	int	User ID
organization_id	int	Organization ID
is_super_admin	bool	"Is Super Admin" flag from SmartSensor

Returns:

Type	Description
bool	True if authorization token is set correctly, False otherwise.

Source code in reportconnectors/api_client/smartsensor_api_client.py

def set_auth_token(self, auth_token: str, use_api_data: bool = False, **kwargs) -> bool:
    """
    Sets the authorization token to SmartSensor API.

    To be used when authorization token is obtained from outside. Calling .authenticate() is not required
    when the auth token is present.

    Args:
        auth_token: Valid authorization token to SmartSensor API.
        use_api_data: If True then the extra data (refresh_token, is_super_admin, etc.) will be obtained from
            `User/Token` endpoint. Otherwise, it has to be provided in kwargs.

    Keyword Args:
        refresh_token (str): Valid refresh token
        user_id (int): User ID
        organization_id (int): Organization ID
        is_super_admin (bool): "Is Super Admin" flag from SmartSensor

    Returns:
        True if authorization token is set correctly, False otherwise.
    """

    try:
        decoded_token = jwt.decode(auth_token, options={"verify_signature": False})
        _exp_timestamp = float(decoded_token.get("exp"))  # type: ignore
        expiration_date = datetime.datetime.fromtimestamp(_exp_timestamp).replace(tzinfo=datetime.timezone.utc)
    except (ValueError, TypeError, jwt.exceptions.InvalidTokenError):
        log.error("Invalid token provided.")
        return False

    # Set new token ...
    self.auth_data[self.KeyNames.ACCESS_TOKEN] = auth_token
    self.auth_data[self.KeyNames.EXPIRATION_DATE] = expiration_date

    # ... and other extra parameters
    if use_api_data:
        auth_details = self.get_auth_details()
        self.auth_data[self.KeyNames.REFRESH_TOKEN] = auth_details.get("refreshToken")
        self.is_super_admin = auth_details.get("isSuperAdmin", False)
        self.user_id = auth_details.get("userID", None)
        self.organization_id = auth_details.get("organizationID", None)
    else:
        self.auth_data[self.KeyNames.REFRESH_TOKEN] = kwargs.get("refresh_token")
        self.is_super_admin = kwargs.get("is_super_admin", False)
        self.user_id = kwargs.get("user_id", None)
        self.organization_id = kwargs.get("organization_id", None)
    return self.is_logged

logout()

Logs out the client from the API. It runs the parent class logout procedure fist, and then in resets the custom attributes of Smart Sensor API client

Source code in reportconnectors/api_client/smartsensor_api_client.py

def logout(self):
    """
    Logs out the client from the API. It runs the parent class logout procedure fist,
    and then in resets the custom attributes of Smart Sensor API client
    """

    super().logout()
    self.user_id = None
    self.organization_id = None
    self.is_super_admin = None

_make_access_token_refresh(force=False, refresh_token=None)

Refreshes the access token using the existing refresh token.

Because it overrides the base class method, it will be called by _process_request method just before calling _send_request method.

In default mode, it will only try to refresh the token if two conditions are met:

1. auto_refresh parameter is True
2. Token's remaining time is less than auto_refresh_delta (10 min by default)

Token refresh can be forced by calling the function with force = True.

Parameters:

Name	Type	Description	Default
force	bool	If true then the token refresh process is executed regardless of the above conditions.	False
refresh_token	Optional[str]	Custom refresh token to be used. If not provided, then the self.refresh_token property is used.	None

Returns:

Type	Description
bool	True, if the token refresh process is successful. False, otherwise.

Source code in reportconnectors/api_client/smartsensor_api_client.py

def _make_access_token_refresh(self, force: bool = False, refresh_token: Optional[str] = None) -> bool:
    """
    Refreshes the access token using the existing refresh token.

    Because it overrides the base class method, it will be called by `_process_request` method just before calling
    `_send_request` method.

    In default mode, it will only try to refresh the token if two conditions are met:

    1. `auto_refresh` parameter is True
    2. Token's remaining time is less than auto_refresh_delta (10 min by default)

    Token refresh can be forced by calling the function with force = True.

    Args:
        force: If true then the token refresh process is executed regardless of the above conditions.
        refresh_token: Custom refresh token to be used. If not provided, then the `self.refresh_token` property
            is used.

    Returns:
        True, if the token refresh process is successful. False, otherwise.
    """
    _auto_refresh_delta = datetime.timedelta(minutes=10)

    # Check required inputs
    _refresh_token = refresh_token if refresh_token else self.refresh_token
    if not (_refresh_token and self.auth_token):
        return False
    # Check if it's time to refresh
    _expiration_date = self.auth_data.get(self.KeyNames.EXPIRATION_DATE)
    if not isinstance(_expiration_date, datetime.datetime):
        return False
    _time_to_expire = _expiration_date - datetime.datetime.now(datetime.timezone.utc)
    is_time_to_refresh = force or (self._auto_refresh and _time_to_expire < _auto_refresh_delta)
    if not is_time_to_refresh:
        return False

    # Finally perform the token refreshment
    _status = self.authenticate(refresh_token=_refresh_token)
    if _status:
        self._token_refresh_dates.append(datetime.datetime.now(datetime.timezone.utc))
        log.debug(f"Access token has been refreshed. New expiration date is: {self.token_expiration_date}")
    return _status

get_asset_data(asset_id)

Gets asset's data using Asset/Data{id} endpoint

Parameters:

Name	Type	Description	Default
asset_id	Union[str, int]	ID of an asset	required

Returns:

Type	Description
Dict	Asset's data. If not found, empty dict is returned.

Source code in reportconnectors/api_client/smartsensor_api_client.py

def get_asset_data(self, asset_id: Union[str, int]) -> Dict:
    """
    Gets asset's data using `Asset/Data{id}` endpoint

    Args:
        asset_id: ID of an asset

    Returns:
        Asset's data. If not found, empty dict is returned.
    """
    endpoint = f"Asset/Data/{asset_id}"
    response = self._make_request(method="GET", endpoint=endpoint)
    decoded_response: Dict = self._decode_response(response=response, default={})
    return decoded_response

get_asset_data_by_sensor_id(sensor_id)

Gets asset's data using Asset/Sensor endpoint. Sensor ID is used as an identifier.

Parameters:

Name	Type	Description	Default
sensor_id	str	ID of a sensor.	required

Returns:

Type	Description
Dict	Asset's data. If not found, empty dict is returned.

Source code in reportconnectors/api_client/smartsensor_api_client.py

def get_asset_data_by_sensor_id(self, sensor_id: str) -> Dict:
    """
    Gets asset's data using `Asset/Sensor` endpoint. Sensor ID is used as an identifier.

    Args:
        sensor_id: ID of a sensor.

    Returns:
        Asset's data. If not found, empty dict is returned.
    """

    endpoint = "Asset/Sensor"
    params = {"identifier": sensor_id}
    response = self._make_request(method="GET", endpoint=endpoint, params=params)
    decoded_response: Dict = self._decode_response(response=response, default={})
    return decoded_response

get_asset_details(asset_id)

Gets detailed asset info using Asset/{id} endpoint

Parameters:

Name	Type	Description	Default
asset_id	Union[str, int]	ID of an asset	required

Returns:

Type	Description
Dict	Asset's details. If not found, empty dict is returned.

Source code in reportconnectors/api_client/smartsensor_api_client.py

def get_asset_details(self, asset_id: Union[str, int]) -> Dict:
    """
    Gets detailed asset info using `Asset/{id}` endpoint

    Args:
        asset_id: ID of an asset

    Returns:
        Asset's details. If not found, empty dict is returned.
    """
    endpoint = f"Asset/{asset_id}"
    response = self._make_request(method="GET", endpoint=endpoint, feature_code="EXT_ViewAssetDetails")
    decoded_response: Dict = self._decode_response(response=response, default={})
    return decoded_response

get_asset_device(asset_id)

Gets Asset's device data using Asset/Device/{id} endpoint.

Parameters:

Name	Type	Description	Default
asset_id	Union[str, int]	ID of an asset.	required

Returns:

Type	Description
Dict	Asset's device data. If not found, empty dict is returned.

Source code in reportconnectors/api_client/smartsensor_api_client.py

def get_asset_device(self, asset_id: Union[str, int]) -> Dict:
    """
    Gets Asset's device data using `Asset/Device/{id}` endpoint.

    Args:
        asset_id: ID of an asset.

    Returns:
        Asset's device data. If not found, empty dict is returned.
    """
    endpoint = f"Asset/Device/{asset_id}"
    response = self._make_request(method="GET", endpoint=endpoint)
    decoded_response: Dict = self._decode_response(response=response, default={})
    return decoded_response

get_asset_health(plant_id=None, asset_group_id=None, organization_id=None, assets_without_group=False)

Gets a list of assets with health information, using Asset/Health endpoint.

Parameters:

Name	Type	Description	Default
plant_id	Optional[int]	ID of a Plant	None
asset_group_id	Optional[int]	ID of an Asset Group.	None
organization_id	Optional[int]	ID of an organization.	None
assets_without_group	bool	If true, assets that don't belong to any group will be included in the response. Otherwise, only the assets that belong to any Asset Group will be included.	False

Returns:

Type	Description
List	List of assets health information. If not found then the empty list is returned.

Source code in reportconnectors/api_client/smartsensor_api_client.py

def get_asset_health(
    self,
    plant_id: Optional[int] = None,
    asset_group_id: Optional[int] = None,
    organization_id: Optional[int] = None,
    assets_without_group: bool = False,
) -> List:
    """
    Gets a list of assets with health information, using `Asset/Health` endpoint.

    Args:
        plant_id: ID of a Plant
        asset_group_id: ID of an Asset Group.
        organization_id: ID of an organization.
        assets_without_group: If true, assets that don't belong to any group will be included in the response.
            Otherwise, only the assets that belong to any Asset Group will be included.

    Returns:
        List of assets health information. If not found then the empty list is returned.
    """

    endpoint = "Asset/Health"
    params = {
        "plantID": plant_id,
        "assetGroupID": asset_group_id,
        "organizationID": organization_id,
        "assetsWithoutGroup": assets_without_group,
    }
    response = self._make_request(method="GET", endpoint=endpoint, params=params)
    decoded_response: List = self._decode_response(response=response, default=[])
    return decoded_response

get_asset_measurement(asset_id)

Gets a collection of the latest measurements from a specified Asset using Asset/Measurement/{id} endpoint

Parameters:

Name	Type	Description	Default
asset_id	Union[str, int]	ID of an Asset.	required

Returns:

Type	Description
List[Dict]	List of the latest measurements. If not found, then the empty list is returned.

Source code in reportconnectors/api_client/smartsensor_api_client.py

def get_asset_measurement(self, asset_id: Union[str, int]) -> List[Dict]:
    """
    Gets a collection of the latest measurements from a specified Asset using `Asset/Measurement/{id}` endpoint

    Args:
        asset_id: ID of an Asset.

    Returns:
        List of the latest measurements. If not found, then the empty list is returned.

    """
    endpoint = f"Asset/Measurement/{asset_id}"
    response = self._make_request(method="GET", endpoint=endpoint)
    decoded_response: List = self._decode_response(response=response, default=[])
    return decoded_response

get_asset_properties(asset_id)

Gets Asset properties list using Asset/Property/{id} endpoint.

Parameters:

Name	Type	Description	Default
asset_id	Union[str, int]	ID of an asset.	required

Returns:

Type	Description
List[Dict]	List of asset properties. If nothing found, the empty list is returned.

Source code in reportconnectors/api_client/smartsensor_api_client.py

def get_asset_properties(self, asset_id: Union[str, int]) -> List[Dict]:
    """
    Gets Asset properties list using `Asset/Property/{id}` endpoint.

    Args:
        asset_id: ID of an asset.

    Returns:
        List of asset properties. If nothing found, the empty list is returned.
    """

    endpoint = f"Asset/Property/{asset_id}"
    response = self._make_request(method="GET", endpoint=endpoint)
    decoded_response: List = self._decode_response(response=response, default=[])
    return decoded_response

get_assets_details_list(organization_id)

Gets detailed asset information from all the accessible assets in the specified organization

Parameters:

Name	Type	Description	Default
organization_id	int	ID of organization.	required

Returns:

Type	Description
List[Dict]	List of asset details in organization. If nothing is found, then the empty list is returned.

Source code in reportconnectors/api_client/smartsensor_api_client.py

def get_assets_details_list(self, organization_id: int) -> List[Dict]:
    """
    Gets detailed asset information from all the accessible assets in the specified organization

    Args:
        organization_id: ID of organization.

    Returns:
        List of asset details in organization. If nothing is found, then the empty list is returned.

    """
    endpoint = "Asset"
    params = {"organizationID": organization_id}
    response = self._make_request(
        method="GET", endpoint=endpoint, params=params, feature_code="EXT_ViewAssetDetails"
    )
    decoded_response: List = self._decode_response(response=response, default=[])
    return decoded_response

get_assets_list(organization_id=None, plant_id=None, asset_group_id=None)

Gets the list of assets in organization.

If no organization id is provided, then the currently selected one is used.

Optionally, the list might be filtered down by providing Plant ID or Asset Group ID.

Parameters:

Name	Type	Description	Default
organization_id	Optional[int]	ID of organization.	None
plant_id	Optional[int]	ID of Plant	None
asset_group_id	Optional[int]	ID of Asset Group.	None

Returns:

Type	Description
List[Dict]	List of assets in organization. If nothing is found, then the empty list is returned.

Source code in reportconnectors/api_client/smartsensor_api_client.py

def get_assets_list(
    self,
    organization_id: Optional[int] = None,
    plant_id: Optional[int] = None,
    asset_group_id: Optional[int] = None,
) -> List[Dict]:
    """
    Gets the list of assets in organization.

    If no organization id is provided, then the currently selected one is used.

    Optionally, the list might be filtered down by providing Plant ID or Asset Group ID.

    Args:
        organization_id: ID of organization.
        plant_id: ID of Plant
        asset_group_id: ID of Asset Group.

    Returns:
        List of assets in organization. If nothing is found, then the empty list is returned.
    """

    endpoint = "Asset/List"
    params = {"organizationID": organization_id if organization_id is not None else self.organization_id}
    if plant_id is not None:
        params["plantID"] = plant_id
    if asset_group_id is not None:
        params["assetGroupID"] = asset_group_id

    response = self._make_request(method="GET", endpoint=endpoint, params=params)
    decoded_response: List = self._decode_response(response=response, default=[])
    return decoded_response

get_auth_details()

Gets authentication information for a currently used access token.

Returns:

Type	Description
Dict	Dictionary with auth details. Default: empty dict.

Source code in reportconnectors/api_client/smartsensor_api_client.py

def get_auth_details(self) -> Dict:
    """
    Gets authentication information for a currently used access token.

    Returns:
        Dictionary with auth details. Default: empty dict.
    """
    response = self._make_request(method="GET", endpoint="/Auth/Token")
    decoded_response: Dict = self._decode_response(response=response, default={})
    return decoded_response

get_event_logs(event_log_type, filter_closed=None, closing_reason=None, asset_id=None, start_date=None, end_date=None, number_of_events=None, organization_id=None, asset_type_id=None, group_events=False, sub_type_id=None)

Gets the event logs of the specified organization as a list. Different filtering options can be chosen.

Source code in reportconnectors/api_client/smartsensor_api_client.py

def get_event_logs(
    self,
    event_log_type: int,
    filter_closed: Optional[bool] = None,
    closing_reason: Optional[int] = None,
    asset_id: Optional[Union[str, int]] = None,
    start_date: Optional[datetime.datetime] = None,
    end_date: Optional[datetime.datetime] = None,
    number_of_events: Optional[int] = None,
    organization_id: Optional[int] = None,
    asset_type_id: Optional[int] = None,
    group_events: bool = False,
    sub_type_id: Optional[int] = None,
) -> List:
    """
    Gets the event logs of the specified organization as a list. Different filtering options can be chosen.
    """
    endpoint = "EventLog"
    params = {
        "eventLogType": event_log_type,
        "filterClosed": filter_closed,
        "closingReason": closing_reason,
        "assetID": asset_id,
        "numberOfEvents": number_of_events,
        "organizationID": organization_id,
        "assetTypeID": asset_type_id,
        "groupEvents": group_events,
        "subTypeID": sub_type_id,
    }
    if start_date is not None:
        params["from"] = to_api_time(start_date)
    if end_date is not None:
        params["to"] = to_api_time(end_date)

    response = self._make_request(method="GET", endpoint=endpoint, params=params)
    decoded_response: List = self._decode_response(response=response, default=[])
    return decoded_response

get_measurement_threshold(asset_id, measurement_ids)

Gets threshold values for a given measurement type and asset id, using Measurement/HealthInterval/{id} endpoint.

Parameters:

Name	Type	Description	Default
asset_id	Union[str, int]	ID of an asset.	required
measurement_ids	Union[Sequence[int], Sequence[str], int, str]	IDs of measurements.	required

Returns:

Type	Description
List[Dict]	List of measurement thresholds. If not found then the empty list is returned.

Source code in reportconnectors/api_client/smartsensor_api_client.py

def get_measurement_threshold(
    self, asset_id: Union[str, int], measurement_ids: Union[Sequence[int], Sequence[str], int, str]
) -> List[Dict]:
    """
    Gets threshold values for a given measurement type and asset id,
    using `Measurement/HealthInterval/{id}` endpoint.

    Args:
        asset_id: ID of an asset.
        measurement_ids: IDs of measurements.

    Returns:
        List of measurement thresholds. If not found then the empty list is returned.
    """

    endpoint = f"Measurement/HealthInterval/{asset_id}"
    parsed_measurement_ids = parse_measurement_ids(measurement_ids)
    if parsed_measurement_ids is None:
        return []
    params = {"measurementTypes": parsed_measurement_ids}
    response = self._make_request(method="GET", endpoint=endpoint, params=params)
    decoded_response: List = self._decode_response(response=response, default=[])
    return decoded_response

get_measurement_unit(measurement_unit_group, standard='metric')

Gets the unit data for a specific measurement id. The output comes from the get_measurement_units response.

Parameters:

Name	Type	Description	Default
measurement_unit_group	int	ID of a measurement group.	required
standard	str	Unit standard. Supported: metric and imperial. Default: `metric	'metric'

Returns:

Type	Description
Dict	Unit data. If not found then the empty dict is returned.

Source code in reportconnectors/api_client/smartsensor_api_client.py

def get_measurement_unit(self, measurement_unit_group: int, standard: str = "metric") -> Dict:
    """
    Gets the unit data for a specific measurement id. The output comes from the `get_measurement_units` response.

    Args:
        measurement_unit_group: ID of a measurement group.
        standard: Unit standard. Supported: `metric` and `imperial`. Default: `metric

    Returns:
        Unit data. If not found then the empty dict is returned.


    """
    no_unit_response = {"symbol": "", "factor": 1.0, "offset": 0.0}
    measurement_units = self.get_measurement_units()

    single_unit_dict = measurement_units.get(measurement_unit_group, {})
    # Default standard is `metric` standard
    default_standard_dict = single_unit_dict.get("metric", no_unit_response)
    standard_dict = single_unit_dict.get(standard.casefold(), default_standard_dict)
    return standard_dict

get_measurement_units(force=False)

Gets the list of units for all available measurements types, using Measurement/Unit endpoint. Due to the static nature of this data, by default it is read only once from the API. Any subsequent requests are dismissed, and the first response is returned instead. To override it, use force=True argument.

Parameters:

Name	Type	Description	Default
force	bool	If true then each method call is calling the API. Otherwise, only the first one does, and subsequent calls are serverd with the first response.	False

Returns:

Type	Description
Dict	Dictionary with measurement unit data. If not found, then the empty dict is returned.

Source code in reportconnectors/api_client/smartsensor_api_client.py

def get_measurement_units(self, force: bool = False) -> Dict:
    """
    Gets the list of units for all available measurements types, using `Measurement/Unit` endpoint.
    Due to the static nature of this data, by default it is read only once from the API. Any subsequent requests
    are dismissed, and the first response is returned instead. To override it, use `force=True` argument.

    Args:
        force: If true then each method call is calling the API. Otherwise, only the first one does, and subsequent
            calls are serverd with the first response.

    Returns:
        Dictionary with measurement unit data. If not found, then the empty dict is returned.

    """
    endpoint = "Measurement/Unit"

    if self._units and not force:
        return self._units

    response = self._make_request(method="GET", endpoint=endpoint)
    decoded_response: List = self._decode_response(response=response, default=[])
    for element in decoded_response:
        _group_id = element.get("measurementUnitGroup", {}).get("measurementUnitGroupID", None)
        _list = element.get("measurementUnits", None)
        if (_group_id is None) or (_list is None):
            continue
        _single_unit_dict = {}
        for _unit_definition in _list:
            _standard = _unit_definition.get("measurementUnitStandard", "").casefold()
            _symbol = _unit_definition.get("measurementUnitSymbol", "")
            _factor = _unit_definition.get("measurementUnitConversionRate", 1.0)
            _offset = _unit_definition.get("measurementUnitConversionOffset", 0.0)
            if _standard == "":
                continue
            _single_unit_dict[_standard] = {"symbol": _symbol, "factor": _factor, "offset": _offset}
            # MANUAL PATCH for Fahrenheits
            # Since Smart Sensor API provides `ConversionRate` only, it is impossible to convert from °C to °F
            # To solve this problem, offset field was created, and appropriate values were added manually
            if _standard == "imperial" and _group_id == 4:
                _single_unit_dict[_standard] = {"symbol": "°F", "factor": 0.5555, "offset": 32.0}
        self._units[_group_id] = _single_unit_dict
    return self._units

get_measurements_for_multiple_assets(asset_ids, measurement_ids, start_date, end_date, delta=datetime.timedelta(days=90))

Gets measurement data for multiple assets in a provided time range.

It is using POST Measurement/List/Value endpoint, that allows data retrieval for a multiple asset ids at once.

This highly increases the performance, and should be used whenever it's possible.

Parameters:

Name	Type	Description	Default
asset_ids	Union[Sequence[str], Sequence[int]]	List of asset ids to get the measurements for. Provided as a list of int or comma-separated string.	required
measurement_ids	Union[Sequence[int], Sequence[str]]	IDs of the measurements that should be received.	required
start_date	datetime	Start date of the measurements.	required
end_date	datetime	End date of the measurements.	required
delta	timedelta	Maximum time range to be requested in a single request. Due to SmartSensor API limitations only requests with time range less than 90 days are accepted. This method provides automatic support for greater time ranges, but the max delta for a single request has to be provided. Default: 90 days.	timedelta(days=90)

Returns:

Type	Description
List	List of measurement data for many assets. If not found then the empty list is returned.

Source code in reportconnectors/api_client/smartsensor_api_client.py

def get_measurements_for_multiple_assets(
    self,
    asset_ids: Union[Sequence[str], Sequence[int]],
    measurement_ids: Union[Sequence[int], Sequence[str]],
    start_date: datetime.datetime,
    end_date: datetime.datetime,
    delta: datetime.timedelta = datetime.timedelta(days=90),
) -> List:
    """
    Gets measurement data for multiple assets in a provided time range.

    It is using POST `Measurement/List/Value` endpoint, that allows data retrieval for a multiple asset ids
    at once.

    This highly increases the performance, and should be used whenever it's possible.

    Args:
        asset_ids: List of asset ids to get the measurements for. Provided as a list of int
            or comma-separated string.
        measurement_ids: IDs of the measurements that should be received.
        start_date: Start date of the measurements.
        end_date: End date of the measurements.
        delta: Maximum time range to be requested in a single request. Due to SmartSensor API limitations
            only requests with time range less than 90 days are accepted. This method provides automatic support
            for greater time ranges, but the max delta for a single request has to be provided. Default: 90 days.

    Returns:
        List of measurement data for many assets. If not found then the empty list is returned.

    """
    endpoint = "Measurement/List/Value"
    if not measurement_ids:
        return []

    sub_time_ranges = split_time_range(start_date=start_date, end_date=end_date, delta=delta)
    responses = []
    for sub_start_date, sub_end_date in sub_time_ranges:
        log.debug(f"Collecting data for time range: {sub_start_date} / {sub_end_date}")
        params = {"from": to_api_time(sub_start_date), "to": to_api_time(sub_end_date)}
        data = {
            "assetIDList": sorted([str(asset_id) for asset_id in asset_ids]),
            "measurementTypes": sorted([str(measurement_id) for measurement_id in measurement_ids]),
        }
        response = self._make_request(method="POST", endpoint=endpoint, params=params, json_data=data)
        decoded_response: List = self._decode_response(response=response, default=[])
        if not decoded_response:
            continue
        responses.append(decoded_response)

    if not responses:
        return []
    merged_response = merge_measurement_value_responses(responses)
    return merged_response

get_measurements_for_single_asset(asset_id, measurement_ids, start_date, end_date)

Gets measurement data for a single asset in a provided time range.

It is using Measurement/Value endpoint, that allows data retrieval for a single asset only.

To get the data for several assets at once use get_measurements_for_many_assets

Parameters:

Name	Type	Description	Default
asset_id	Union[str, int]	ID of an asset.	required
measurement_ids	Union[Sequence[int], Sequence[str], int, str]	IDs of the measurements that should be received.	required
start_date	datetime	Start date of the measurements.	required
end_date	datetime	End date of the measurements.	required

Returns:

Type	Description
List[Dict]	List of measurement data for single asset. If not found then the empty list is returned.

Source code in reportconnectors/api_client/smartsensor_api_client.py

def get_measurements_for_single_asset(
    self,
    asset_id: Union[str, int],
    measurement_ids: Union[Sequence[int], Sequence[str], int, str],
    start_date: datetime.datetime,
    end_date: datetime.datetime,
) -> List[Dict]:
    """
    Gets measurement data for a single asset in a provided time range.

    It is using `Measurement/Value` endpoint, that allows data retrieval for a single asset only.

    To get the data for several assets at once use `get_measurements_for_many_assets`

    Args:
        asset_id: ID of an asset.
        measurement_ids: IDs of the measurements that should be received.
        start_date: Start date of the measurements.
        end_date: End date of the measurements.

    Returns:
        List of measurement data for single asset. If not found then the empty list is returned.
    """
    endpoint = "Measurement/Value"
    parsed_measurement_ids = parse_measurement_ids(measurement_ids)
    if parsed_measurement_ids is None:
        return []
    params = {
        "assetID": int(asset_id),
        "measurementTypes": parsed_measurement_ids,
        "from": to_api_time(start_date),
        "to": to_api_time(end_date),
    }
    response = self._make_request(method="GET", endpoint=endpoint, params=params)
    decoded_response: List = self._decode_response(response=response, default=[])
    return decoded_response

get_organization_data(organization_id)

Gets the data about the organization. It only works for super admins.

Parameters:

Name	Type	Description	Default
organization_id	int	ID of an organization.	required

Returns:

Type	Description
Dict	Organization data if found. Otherwise, Empty dict.

Source code in reportconnectors/api_client/smartsensor_api_client.py

def get_organization_data(self, organization_id: int) -> Dict:
    """
    Gets the data about the organization. It only works for super admins.

    Args:
        organization_id: ID of an organization.

    Returns:
        Organization data if found. Otherwise, Empty dict.

    """
    endpoint = f"Organization/{organization_id}"
    response = self._make_request(method="GET", endpoint=endpoint)
    decoded_response: Dict = self._decode_response(response=response, default={})
    return decoded_response

get_organization_name(organization_id)

Get the working organization name based on provided organization ID

Parameters:

Name	Type	Description	Default
organization_id	int	ID of organization	required

Returns:

Type	Description
Optional[str]	Organization name if found. Otherwise, None is returned.

Source code in reportconnectors/api_client/smartsensor_api_client.py

def get_organization_name(self, organization_id: int) -> Optional[str]:
    """
    Get the working organization name based on provided organization ID

    Args:
        organization_id: ID of organization

    Returns:
        Organization name if found. Otherwise, `None` is returned.
    """

    def _search_organization_in_user_history(_id: int) -> Dict:
        organizations = self.get_organizations_list(use_organization_history=True)
        # Find the organization data by its id
        for org_data in organizations:
            if org_data.get("organizationID", None) == _id:
                return org_data
        return {}

    # super admins can use /Organization/{id} endpoint which is much faster than listing all organizations and
    # searching those organizations by id
    # regular users cannot do that, so we will be searching in user organization history it this case
    if self.is_super_admin:
        organization_data = self.get_organization_data(organization_id=organization_id)
    else:
        organization_data = _search_organization_in_user_history(_id=organization_id)

    if not organization_data:
        log.debug(f"Organization (id={organization_id}) not found")
        return None

    organization_name = organization_data.get("organizationName", None)
    return organization_name

get_organizations_list(use_organization_history=True)

Gets the list of organizations. For SuperAdmin users /Organization endpoint is used. For standard users the list of available organizations is read from User/OrganizationHistory endpoint.

Parameters:

Name	Type	Description	Default
use_organization_history	bool	If true, then the list of available organizations for standard users is read from User/OrganizationHistory endpoint. Otherwise, Organization endpoint is always used.	True

Source code in reportconnectors/api_client/smartsensor_api_client.py

def get_organizations_list(self, use_organization_history: bool = True) -> List[Dict]:
    """
    Gets the list of organizations.
    For SuperAdmin users `/Organization` endpoint is used. For standard users the list of available organizations
    is read from `User/OrganizationHistory` endpoint.

    Args:
        use_organization_history: If true, then the list of available organizations for standard users
            is read from `User/OrganizationHistory` endpoint. Otherwise, `Organization` endpoint is always used.
    """
    _endpoint = (
        "Organization" if (self.is_super_admin or not use_organization_history) else "User/OrganizationHistory"
    )

    response = self._make_request(method="GET", endpoint=_endpoint)
    decoded_response: List = self._decode_response(response=response, default=[])

    # Store the results locally for organization name look-up
    self._organizations = decoded_response
    return decoded_response

get_plant_data(plant_id)

Gets the plant data, using Plant/{id} endpoint

Parameters:

Name	Type	Description	Default
plant_id	int	ID of a Plant	required

Returns:

Type	Description
Dict	Plant data. If not found then the empty dict is returned.

Source code in reportconnectors/api_client/smartsensor_api_client.py

def get_plant_data(self, plant_id: int) -> Dict:
    """
    Gets the plant data, using `Plant/{id}` endpoint

    Args:
        plant_id: ID of a Plant

    Returns:
        Plant data. If not found then the empty dict is returned.
    """
    endpoint = f"Plant/{plant_id}"
    response = self._make_request(method="GET", endpoint=endpoint)
    decoded_response: Dict = self._decode_response(response=response, default={})
    return decoded_response

get_plants_list(organization_id=None)

Gets a list of all plants in the organization. If no organization id is provided, then the currently selected one is used.

Parameters:

Name	Type	Description	Default
organization_id	Optional[int]	ID of an organization	None

Returns:

Type	Description
List[Dict]	List of plants' data. If nothing found, the empty list is returned.

Source code in reportconnectors/api_client/smartsensor_api_client.py

def get_plants_list(self, organization_id: Optional[int] = None) -> List[Dict]:
    """
    Gets a list of all plants in the organization.
    If no organization id is provided, then the currently selected one is used.

    Args:
        organization_id: ID of an organization

    Returns:
        List of plants' data. If nothing found, the empty list is returned.

    """
    endpoint = "Plant"
    if organization_id is None:
        log.debug(f"Using default Organization(id={self.organization_id})")
        organization_id = self.organization_id

    params = {"organizationID": organization_id}
    response = self._make_request(method="GET", endpoint=endpoint, params=params)
    decoded_response: List = self._decode_response(response=response, default=[])
    # Store plants' data in the `_plants[org_id]` attribute.
    self._plants[organization_id] = decoded_response
    return decoded_response

get_sites_list(organization_id)

Gets the list of sites from the organization identified by organizationID.

Parameters:

Name	Type	Description	Default
organization_id	int	ID of an organization.	required

Returns:

Type	Description
List[Dict]	List of dictionaries containing information about Sites that belong to the organization.

Source code in reportconnectors/api_client/smartsensor_api_client.py

def get_sites_list(self, organization_id: int) -> List[Dict]:
    """
    Gets the list of sites from the organization identified by organizationID.

    Args:
        organization_id: ID of an organization.

    Returns:
        List of dictionaries containing information about Sites that belong to the organization.
    """

    endpoint = "Site"
    params = {"organizationID": organization_id}

    response = self._make_request(method="GET", endpoint=endpoint, params=params)
    decoded_response: List = self._decode_response(response=response, default=[])
    return decoded_response

get_raw_data_for_asset(asset_id, mode='csv', start_date=None, end_date=None, fetch_files=True)

Gets list of raw data objects. It accepts two modes: "csv" and "binary".

If start_date and end_date are provided, it returns all objects found between those days, otherwise only the latest raw data object is returned.

If fetch_files parameter is set to False while the start and end dates are provided then the returned list of raw data objects doesn't contain the actual contend of raw data file, but only list their names. That is helpful when you want only to list all available raw data, but you don't need to download it.

Parameters:

Name	Type	Description	Default
asset_id	int	ID of an asset.	required
mode	Literal['binary', 'csv']	Raw data mode. Default: csv	'csv'
start_date	Optional[datetime]	Start date	None
end_date	Optional[datetime]	End date	None
fetch_files	bool	If True the actual content of raw data is included in the response. Otherwise, only the file names are included.	True

Returns:

Type	Description
List[Dict]	List of "Raw Data" data for a given asset. If nothing is found, then the empty list is returned.

Source code in reportconnectors/api_client/smartsensor_api_client.py

def get_raw_data_for_asset(
    self,
    asset_id: int,
    mode: Literal["binary", "csv"] = "csv",
    start_date: Optional[datetime.datetime] = None,
    end_date: Optional[datetime.datetime] = None,
    fetch_files: bool = True,
) -> List[Dict]:
    """
    Gets list of raw data objects. It accepts two modes: "csv" and "binary".

    If `start_date` and `end_date` are provided, it returns all objects found between those days,
    otherwise only the latest raw data object is returned.

    If `fetch_files` parameter is set to False while the start and end dates are provided then the returned
    list of raw data objects doesn't contain the actual contend of raw data file, but only list their names.
    That is helpful when you want only to list all available raw data, but you don't need to download it.

    Args:
        asset_id: ID of an asset.
        mode: Raw data mode. Default: `csv`
        start_date: Start date
        end_date: End date
        fetch_files: If True the actual content of raw data is included in the response. Otherwise, only the
            file names are included.

    Returns:
        List of "Raw Data" data for a given asset. If nothing is found, then the empty list is returned.

    """
    params: Dict[str, Union[str, int]] = {"assetID": asset_id}
    _type = self.KeyNames.RAW_DATA_BINARY if mode.casefold() == "binary" else self.KeyNames.RAW_DATA_CSV

    if isinstance(start_date, datetime.datetime) and isinstance(end_date, datetime.datetime):
        endpoint = f"TimeBasedAnalytics/KPI/Values/RawData/{_type}"
        params["from"] = to_api_time(start_date)
        params["to"] = to_api_time(end_date)
        params["fetchFileContents"] = fetch_files
    else:
        endpoint = f"TimeBasedAnalytics/KPI/RawData/{_type}"

    response = self._make_request(method="GET", endpoint=endpoint, params=params)
    decoded_response: List = self._decode_response(response=response, default=[])
    return decoded_response

get_sensor_details(sensor_identifier, sensor_type_id)

Gets the sensor details using the Sensor/Details/{id} endpoint

Parameters:

Name	Type	Description	Default
sensor_identifier	str	Sensor ID	required
sensor_type_id	int	Sensor Type ID	required

Returns:

Type	Description
Dict	Dictionary with sensor details. If nothing found, the empty dict is returned.

Source code in reportconnectors/api_client/smartsensor_api_client.py

def get_sensor_details(self, sensor_identifier: str, sensor_type_id: int) -> Dict:
    """
    Gets the sensor details using the `Sensor/Details/{id}` endpoint

    Args:
        sensor_identifier: Sensor ID
        sensor_type_id: Sensor Type ID

    Returns:
        Dictionary with sensor details. If nothing found, the empty dict is returned.

    """
    endpoint = f"Sensor/Details/{sensor_identifier}"
    params = {"sensorTypeID": sensor_type_id}
    response = self._make_request(method="GET", endpoint=endpoint, params=params)
    decoded_response: Dict = self._decode_response(response=response, default={})
    return decoded_response

get_sensor_feature(sensor_identifier, feature_types, sensor_type_id, start_date=None, end_date=None)

Gets the sensor feature using Sensor/Feature/{id} endpoint

Parameters:

Name	Type	Description	Default
sensor_identifier	str	Sensor ID	required
sensor_type_id	int	Sensor Type ID	required
feature_types	str	Feature types, As a string with comma separated list of features.	required
start_date	Optional[datetime]	Start Date. Used with time based features.	None
end_date	Optional[datetime]	End Date. User with time based features.	None

Returns:

Type	Description
List[Dict]	List of features. If not found, then the empty list is returned.

Source code in reportconnectors/api_client/smartsensor_api_client.py

def get_sensor_feature(
    self,
    sensor_identifier: str,
    feature_types: str,
    sensor_type_id: int,
    start_date: Optional[datetime.datetime] = None,
    end_date: Optional[datetime.datetime] = None,
) -> List[Dict]:
    """
    Gets the sensor feature using `Sensor/Feature/{id}` endpoint

    Args:
        sensor_identifier: Sensor ID
        sensor_type_id: Sensor Type ID
        feature_types: Feature types, As a string with comma separated list of features.
        start_date: Start Date. Used with time based features.
        end_date: End Date. User with time based features.

    Returns:
        List of features. If not found, then the empty list is returned.

    """

    endpoint = f"Sensor/Feature/{sensor_identifier}"
    params = {"sensorTypeID": sensor_type_id, "featureTypes": feature_types}
    if start_date is not None:
        params["from"] = to_api_time(start_date)
    if end_date is not None:
        params["to"] = to_api_time(end_date)

    response = self._make_request(method="GET", endpoint=endpoint, params=params)
    decoded_response: List = self._decode_response(response=response, default=[])
    return decoded_response

get_sensor_latest_features(sensor_identifier, sensor_type_id, feature_types='', include_complex_objects=False)

Gets the list of the latest features using Sensor/Feature/Value/{id} endpoint

Parameters:

Name	Type	Description	Default
sensor_identifier	str	Sensor ID	required
sensor_type_id	int	Sensor Type ID	required
feature_types	str	Feature types, As a string with comma separated list of features.	''
include_complex_objects	bool	If True, then the complex objects (e.g. raw data content) will be included in the response, otherwise only the headers of those complex objects are present. Default: False	False

Returns:

Type	Description
List[Dict]	List of the latest sensor features. If not found then the empty list is returned.

Source code in reportconnectors/api_client/smartsensor_api_client.py

def get_sensor_latest_features(
    self,
    sensor_identifier: str,
    sensor_type_id: int,
    feature_types: str = "",
    include_complex_objects: bool = False,
) -> List[Dict]:
    """
    Gets the list of the latest features using `Sensor/Feature/Value/{id}` endpoint

    Args:
        sensor_identifier: Sensor ID
        sensor_type_id: Sensor Type ID
        feature_types: Feature types, As a string with comma separated list of features.
        include_complex_objects: If True, then the complex objects (e.g. raw data content) will be included
            in the response, otherwise only the headers of those complex objects are present. Default: `False`

    Returns:
        List of the latest sensor features. If not found then the empty list is returned.

    """
    endpoint = f"Sensor/Feature/Value/{sensor_identifier}"
    params = {
        "sensorTypeID": sensor_type_id,
        "featureTypes": feature_types,
        "complexObject": include_complex_objects,
    }
    response = self._make_request(method="GET", endpoint=endpoint, params=params)
    decoded_response: List = self._decode_response(response=response, default=[])
    return decoded_response

get_sensor_properties(sensor_identifier, sensor_type_id)

Gets sensor properties, as a subset of sensor details.

Args: sensor_identifier: Sensor ID sensor_type_id: Sensor Type ID

Returns:

Type	Description
List[Dict]	List with sensor properties. If nothing found, the empty list is returned.

Source code in reportconnectors/api_client/smartsensor_api_client.py

def get_sensor_properties(self, sensor_identifier: str, sensor_type_id: int) -> List[Dict]:
    """
     Gets sensor properties, as a subset of sensor details.

     Args:
        sensor_identifier: Sensor ID
        sensor_type_id: Sensor Type ID

    Returns:
        List with sensor properties. If nothing found, the empty list is returned.


    """
    sensor_details = self.get_sensor_details(sensor_identifier=sensor_identifier, sensor_type_id=sensor_type_id)
    if "properties" in sensor_details:
        return sensor_details["properties"]
    return []

get_user_data()

Gets the data of currently logged user.

Returns:

Type	Description
Dict	Dictionary with user data. Default: empty dict.

Source code in reportconnectors/api_client/smartsensor_api_client.py

def get_user_data(self) -> Dict:
    """
    Gets the data of currently logged user.

    Returns:
        Dictionary with user data. Default: empty dict.
    """
    endpoint = "User"
    response = self._make_request(method="GET", endpoint=endpoint)
    decoded_response: Dict = self._decode_response(response=response, default={})
    return decoded_response

find_asset_id(identifier)

Gets Asset ID based on provided identifier. Identifier might be either 'Asset ID' (then it is only validated) or a 'Sensor ID' (then the .get_asset_data_by_sensor_id method is used)

Parameters:

Name	Type	Description	Default
identifier	Union[int, str]	Asset ID or Sensor ID	required

Returns:

Type	Description
Optional[int]	Asset ID if asset is found. Otherwise, None is returned.

Source code in reportconnectors/api_client/smartsensor_api_client.py

def find_asset_id(self, identifier: Union[int, str]) -> Optional[int]:
    """
    Gets Asset ID based on provided identifier. Identifier might be either 'Asset ID' (then it is only validated)
    or a 'Sensor ID' (then the `.get_asset_data_by_sensor_id method` is used)

    Args:
        identifier: Asset ID or Sensor ID

    Returns:
        Asset ID if asset is found. Otherwise, `None` is returned.
    """
    # Try by Asset ID first
    response = self.get_asset_data(asset_id=identifier)
    asset_id = self._check_response_for_asset_id(response=response)
    if asset_id:
        return asset_id
    # Try by Sensor ID
    response = self.get_asset_data_by_sensor_id(sensor_id=str(identifier))
    asset_id = self._check_response_for_asset_id(response=response)
    return asset_id

send_sensor_feature(sensor_identifier, sensor_type_id, features, method='POST')

Sends the sensor feature to SmartSensor API using Sensor/Feature/{id} endpoint

Parameters:

Name	Type	Description	Default
sensor_identifier	str	Sensor ID	required
sensor_type_id	int	Sensor Type ID	required
features	List[Dict]	List of features to send.	required
method	MethodType	HTTP method to be used while sending the features. Default method: POST	'POST'

Returns:

Type	Description
bool	True if sending was successful. Otherwise, False is returned.

Source code in reportconnectors/api_client/smartsensor_api_client.py

def send_sensor_feature(
    self, sensor_identifier: str, sensor_type_id: int, features: List[Dict], method: MethodType = "POST"
) -> bool:
    """
    Sends the sensor feature to SmartSensor API using `Sensor/Feature/{id}` endpoint

    Args:
        sensor_identifier: Sensor ID
        sensor_type_id: Sensor Type ID
        features: List of features to send.
        method: HTTP method to be used while sending the features. Default method: POST

    Returns:
        True if sending was successful. Otherwise, `False` is returned.

    """

    endpoint = "Sensor/Feature"
    data = {"sensorIdentifier": sensor_identifier, "sensorTypeID": sensor_type_id, "features": features}
    response = self._make_request(method=method, endpoint=endpoint, json_data=data)
    status_code = self._get_status_code(response=response)
    _is_successful = status_code == HTTPStatus.OK
    return _is_successful

send_report_confirmation(token)

Sends a confirmation to the Report/Completed endpoint, when the report generation process is ready.

This function can be used in ReportAzureFunction to confirm when the report generation process is done.

Parameters:

Name	Type	Description	Default
token	str	Report generation process token.	required

Returns:

Type	Description
bool	Delivery status as a bool. True if the response status code was 200, False otherwise.

Source code in reportconnectors/api_client/smartsensor_api_client.py

def send_report_confirmation(self, token: str) -> bool:
    """
    Sends a confirmation to the `Report/Completed` endpoint, when the report generation process is ready.

    This function can be used in ReportAzureFunction to confirm when the report generation process is done.

    Args:
        token: Report generation process token.

    Returns:
        Delivery status as a bool. True if the response status code was 200, False otherwise.
    """

    endpoint = "Report/Completed"
    params = {"token": token}
    response = self._make_request(method="GET", endpoint=endpoint, params=params)
    status_code = self._get_status_code(response=response)
    is_status_code = status_code == HTTPStatus.OK
    return is_status_code

set_organization(organization_name)

Sets the new working organization for a current user, based on provided organization ID.

Parameters:

Name	Type	Description	Default
organization_name	str	Name of organization to be switched to.	required

Returns:

Type	Description
bool	True if the organization change was successful. Otherwise, false.

Source code in reportconnectors/api_client/smartsensor_api_client.py

def set_organization(self, organization_name: str) -> bool:
    """
    Sets the new working organization for a current user, based on provided organization ID.

    Args:
        organization_name: Name of organization to be switched to.

    Returns:
        True if the organization change was successful. Otherwise, false.
    """

    data = {"organizationName": organization_name}
    response = self._make_request(method="PUT", endpoint="User/Organization", json_data=data)
    decoded_response: Dict = self._decode_response(response=response, default={})
    status_code = self._get_status_code(response=response)
    _new_org_id = decoded_response.get("organizationID", None)
    if (status_code == HTTPStatus.OK) and _new_org_id:
        self.organization_id = _new_org_id
        return True
    else:
        return False

set_organization_id(organization_id)

Sets the new working organization for a current user, based on provided organization ID.

Parameters:

Name	Type	Description	Default
organization_id	int	ID of organization to be switched to.	required

Returns:

Type	Description
bool	True if the organization change was successful. Otherwise, false.

Source code in reportconnectors/api_client/smartsensor_api_client.py

def set_organization_id(self, organization_id: int) -> bool:
    """
    Sets the new working organization for a current user, based on provided organization ID.

    Args:
        organization_id: ID of organization to be switched to.

    Returns:
        True if the organization change was successful. Otherwise, false.
    """

    organization_name = self.get_organization_name(organization_id)
    if not organization_name:
        # Return false if there is no matching organization ID
        return False

    return self.set_organization(organization_name)