Interpolation at Time Function

get(connection, parameters_dict)

An RTDIP interpolation at time function which works out the linear interpolation at a specific time based on the points before and after.

This function requires the user to input a dictionary of parameters. (See Attributes table below.)

Parameters:

Name	Type	Description	Default
connection	object	Connection chosen by the user (Databricks SQL Connect, PYODBC SQL Connect, TURBODBC SQL Connect)	required
parameters_dict	dict	A dictionary of parameters (see Attributes table below)	required

Attributes:

Name	Type	Description
business_unit	str	Business unit of the data
region	str	Region
asset	str	Asset
data_security_level	str	Level of data security
data_type	str	Type of the data (float, integer, double, string)
tag_names	str	Name of the tag
timestamps	list	List of timestamp or timestamps in the format YYY-MM-DDTHH:MM:SS or YYY-MM-DDTHH:MM:SS+zz:zz where %z is the timezone. (Example +00:00 is the UTC timezone)
window_length	int	Add longer window time in days for the start or end of specified date to cater for edge cases.
include_bad_data	bool	Include "Bad" data points with True or remove "Bad" data points with False
pivot	bool	Pivot the data on timestamp column with True or do not pivot the data with False
limit	optional int	The number of rows to be returned
offset	optional int	The number of rows to skip before returning rows
case_insensitivity_tag_search	optional bool	Search for tags using case insensitivity with True or case sensitivity with False

Returns:

Name	Type	Description
DataFrame	DataFrame	A interpolated at time dataframe.

Warning

Setting case_insensitivity_tag_search to True will result in a longer query time.

Source code in src/sdk/python/rtdip_sdk/queries/time_series/interpolation_at_time.py

def get(connection: object, parameters_dict: dict) -> pd.DataFrame:
    """
    An RTDIP interpolation at time function which works out the linear interpolation at a specific time based on the points before and after.

    This function requires the user to input a dictionary of parameters. (See Attributes table below.)

    Args:
        connection: Connection chosen by the user (Databricks SQL Connect, PYODBC SQL Connect, TURBODBC SQL Connect)
        parameters_dict: A dictionary of parameters (see Attributes table below)

    Attributes:
        business_unit (str): Business unit of the data
        region (str): Region
        asset (str):  Asset
        data_security_level (str): Level of data security
        data_type (str): Type of the data (float, integer, double, string)
        tag_names (str): Name of the tag
        timestamps (list): List of timestamp or timestamps in the format YYY-MM-DDTHH:MM:SS or YYY-MM-DDTHH:MM:SS+zz:zz where %z is the timezone. (Example +00:00 is the UTC timezone)
        window_length (int): Add longer window time in days for the start or end of specified date to cater for edge cases.
        include_bad_data (bool): Include "Bad" data points with True or remove "Bad" data points with False
        pivot (bool): Pivot the data on timestamp column with True or do not pivot the data with False
        limit (optional int): The number of rows to be returned
        offset (optional int): The number of rows to skip before returning rows
        case_insensitivity_tag_search (optional bool): Search for tags using case insensitivity with True or case sensitivity with False

    Returns:
        DataFrame: A interpolated at time dataframe.

    !!! warning
        Setting `case_insensitivity_tag_search` to True will result in a longer query time.
    """
    if isinstance(parameters_dict["tag_names"], list) is False:
        raise ValueError("tag_names must be a list")

    if isinstance(parameters_dict["timestamps"], list) is False:
        raise ValueError("timestamps must be a list")

    try:
        query = _query_builder(parameters_dict, "interpolation_at_time")

        try:
            cursor = connection.cursor()
            cursor.execute(query)
            df = cursor.fetch_all()
            cursor.close()
            connection.close()
            return df
        except Exception as e:
            logging.exception("error returning dataframe")
            raise e

    except Exception as e:
        logging.exception("error with interpolation at time function")
        raise e

Example

from rtdip_sdk.authentication.azure import DefaultAuth
from rtdip_sdk.connectors import DatabricksSQLConnection
from rtdip_sdk.queries import interpolation_at_time

auth = DefaultAuth().authenticate()
token = auth.get_token("2ff814a6-3304-4ab8-85cb-cd0e6f879c1d/.default").token
connection = DatabricksSQLConnection("{server_hostname}", "{http_path}", token)

parameters = {
    "business_unit": "{business_unit}",
    "region": "{region}", 
    "asset": "{asset_name}", 
    "data_security_level": "{security_level}", 
    "data_type": "float",
    "tag_names": ["{tag_name_1}", "{tag_name_2}"],
    "timestamps": ["2023-01-01", "2023-01-02"], 
    "window_length": 1,
}
x = interpolation_at_time.get(connection, parameters)
print(x)

This example is using DefaultAuth() and DatabricksSQLConnection() to authenticate and connect. You can find other ways to authenticate here. The alternative built in connection methods are either by PYODBCSQLConnection(), TURBODBCSQLConnection() or SparkConnection().

Note

See Samples Repository for full list of examples.

Note

server_hostname and http_path can be found on the SQL Warehouses Page.