diff --git a/environment.yml b/environment.yml index 13cbe63cc..3702ae958 100644 --- a/environment.yml +++ b/environment.yml @@ -18,6 +18,7 @@ channels: - conda-forge dependencies: - python>=3.10,<3.13 + - setuptools>=68.0.0 - importlib-metadata>=7.0.0,<9.0.0 - jinja2>=3.1.6,<4.0.0 - pytest==7.4.0 @@ -25,7 +26,7 @@ dependencies: - pytest-cov==4.1.0 - pylint==2.17.4 - pip>=23.1.2 - - turbodbc==4.11.0 + - turbodbc>=4.5.0 - numpy>=1.23.4,<2.0.0 - oauthlib>=3.2.2,<4.0.0 - cryptography>=38.0.3 @@ -74,6 +75,8 @@ dependencies: - pmdarima>=2.0.4 - scikit-learn>=1.3.0,<1.6.0 - pip: + # Ensure setuptools is installed first for pkg_resources needed by turbodbc + - setuptools>=68.0.0 # protobuf installed via pip to avoid libabseil conflicts with conda libarrow - protobuf>=5.29.0,<5.30.0 # pyarrow constraint must match conda version to avoid upgrades @@ -91,4 +94,4 @@ dependencies: - web3>=7.7.0,<8.0.0 - eth-typing>=5.0.1,<6.0.0 - pandas>=2.0.1,<2.3.0 - - moto[s3]>=5.0.16,<6.0.0 + - moto[s3]>=5.0.16,<6.0.0 \ No newline at end of file diff --git a/setup.py b/setup.py index 8f1f16db1..9bed41785 100644 --- a/setup.py +++ b/setup.py @@ -28,6 +28,8 @@ long_description = (here / "PYPI-README.md").read_text() INSTALL_REQUIRES = [ + "setuptools>=68.0.0", + # "turbodbc>=4.5.0", "databricks-sql-connector>=3.6.0,<3.7.0", "pyarrow>=14.0.1,<17.0.0", "azure-core>=1.38.0,<2.0.0", diff --git a/src/api/Dockerfile b/src/api/Dockerfile index 0ddc1763b..2230ae551 100644 --- a/src/api/Dockerfile +++ b/src/api/Dockerfile @@ -22,11 +22,11 @@ ENV AzureWebJobsScriptRoot=/home/site/wwwroot \ COPY src/api/requirements.txt / RUN rm -rf /var/lib/apt/lists/partial \ - apt-get clean \ + && apt-get clean \ && apt-get update -o Acquire::CompressionTypes::Order::=gz \ && apt-get --no-install-recommends install -y odbcinst1debian2 libodbc1 odbcinst unixodbc unixodbc-dev libsasl2-dev libsasl2-modules-gssapi-mit libboost-all-dev \ && apt-get --no-install-recommends install -y ca-certificates curl python3-pip python3-dev python3-setuptools python3-wheel gcc g++ \ - && apt-get --no-install-recommends install -y zip unzip wget \ + && apt-get --no-install-recommends install -y zip unzip wget \ && wget --secure-protocol=TLSv1_2 --max-redirect=0 https://databricks-bi-artifacts.s3.us-east-2.amazonaws.com/simbaspark-drivers/odbc/2.7.7/SimbaSparkODBC-2.7.7.1016-Debian-64bit.zip -P /odbc/ \ && unzip /odbc/SimbaSparkODBC-2.7.7.1016-Debian-64bit.zip -d /odbc \ && dpkg -i /odbc/simbaspark_2.7.7.1016-2_amd64.deb \ @@ -34,9 +34,8 @@ RUN rm -rf /var/lib/apt/lists/partial \ && pip install --no-cache-dir numpy==1.26.4 \ && python -c "import pyarrow; pyarrow.create_library_symlinks()" \ && CFLAGS="-D_GLIBCXX_USE_CXX11_ABI=0" pip install --no-cache-dir -r /requirements.txt \ - && rm -rf /var/lib/apt/lists/* /var/cache/apt/archives/* \ - && rm /odbc -r - + && rm -rf /var/lib/apt/lists/* /var/cache/apt/archives/* \ + && rm -rf /odbc RUN echo '[ODBC Drivers]' > /etc/odbcinst.ini \ && echo 'Simba Spark ODBC Driver = Installed' >> /etc/odbcinst.ini \ && echo '[Simba Spark ODBC Driver]' >> /etc/odbcinst.ini \ @@ -45,5 +44,4 @@ RUN echo '[ODBC Drivers]' > /etc/odbcinst.ini \ USER app COPY src/api/ /home/site/wwwroot -COPY src /home/site/wwwroot/src - +COPY src /home/site/wwwroot/src \ No newline at end of file diff --git a/src/api/FastAPIApp/__init__.py b/src/api/FastAPIApp/__init__.py index 5e0681bc5..00d489852 100644 --- a/src/api/FastAPIApp/__init__.py +++ b/src/api/FastAPIApp/__init__.py @@ -65,9 +65,7 @@ [ReDoc](/redoc) [Real Time Data Ingestion Platform](https://www.rtdip.io/) -""".format( - os.environ.get("TENANT_ID") -) +""".format(os.environ.get("TENANT_ID")) app = FastAPI( title=TITLE, diff --git a/src/api/host.json b/src/api/host.json index 79edb2c32..9d8cbb5f3 100644 --- a/src/api/host.json +++ b/src/api/host.json @@ -15,6 +15,6 @@ }, "extensionBundle": { "id": "Microsoft.Azure.Functions.ExtensionBundle", - "version": "[2.*, 3.0.0)" + "version": "[4.*, 5.0.0)" } } \ No newline at end of file diff --git a/src/api/requirements.txt b/src/api/requirements.txt index 8f79f6a58..22d6a50bf 100644 --- a/src/api/requirements.txt +++ b/src/api/requirements.txt @@ -1,9 +1,10 @@ # Do not include azure-functions-worker as it may conflict with the Azure Functions platform +setuptools>=68.0.0 azure-functions==1.20.0 fastapi==0.121.0 starlette==0.49.1 pydantic==2.10.1 -# turbodbc==4.11.0 +# turbodbc==4.5.3 pyodbc==5.2.0 importlib_metadata>=7.0.0,<9.0.0 databricks-sql-connector==3.6.0 diff --git a/src/api/v1/batch.py b/src/api/v1/batch.py index 4b17c772e..32776a019 100755 --- a/src/api/v1/batch.py +++ b/src/api/v1/batch.py @@ -36,7 +36,6 @@ from concurrent.futures import * import pandas as pd - ROUTE_FUNCTION_MAPPING = { "/events/raw": "raw", "/events/latest": "latest", @@ -120,7 +119,7 @@ async def batch_events_get( try: # Set up connection - (connection, parameters) = common_api_setup_tasks( + connection, _ = common_api_setup_tasks( base_query_parameters=base_query_parameters, base_headers=base_headers, ) diff --git a/src/api/v1/circular_average.py b/src/api/v1/circular_average.py index 382f2a32d..e3d6f71f7 100644 --- a/src/api/v1/circular_average.py +++ b/src/api/v1/circular_average.py @@ -45,7 +45,7 @@ def circular_average_events_get( base_headers, ): try: - (connection, parameters) = common_api_setup_tasks( + connection, parameters = common_api_setup_tasks( base_query_parameters, raw_query_parameters=raw_query_parameters, tag_query_parameters=tag_query_parameters, diff --git a/src/api/v1/circular_standard_deviation.py b/src/api/v1/circular_standard_deviation.py index 836a958a6..7236a8b45 100644 --- a/src/api/v1/circular_standard_deviation.py +++ b/src/api/v1/circular_standard_deviation.py @@ -46,7 +46,7 @@ def circular_standard_deviation_events_get( base_headers, ): try: - (connection, parameters) = common_api_setup_tasks( + connection, parameters = common_api_setup_tasks( base_query_parameters, raw_query_parameters=raw_query_parameters, tag_query_parameters=tag_query_parameters, diff --git a/src/api/v1/common.py b/src/api/v1/common.py index d7d3fa177..573c43529 100644 --- a/src/api/v1/common.py +++ b/src/api/v1/common.py @@ -39,7 +39,6 @@ from src.sdk.python.rtdip_sdk.queries.time_series import batch - if importlib.util.find_spec("turbodbc") != None: from src.sdk.python.rtdip_sdk.connectors import TURBODBCSQLConnection from src.api.auth import azuread diff --git a/src/api/v1/interpolate.py b/src/api/v1/interpolate.py index 0a14feac2..98bfdbbcf 100644 --- a/src/api/v1/interpolate.py +++ b/src/api/v1/interpolate.py @@ -45,7 +45,7 @@ def interpolate_events_get( base_headers, ): try: - (connection, parameters) = common_api_setup_tasks( + connection, parameters = common_api_setup_tasks( base_query_parameters, raw_query_parameters=raw_query_parameters, tag_query_parameters=tag_query_parameters, diff --git a/src/api/v1/interpolation_at_time.py b/src/api/v1/interpolation_at_time.py index cc812bc25..5e6ce0dfe 100644 --- a/src/api/v1/interpolation_at_time.py +++ b/src/api/v1/interpolation_at_time.py @@ -42,7 +42,7 @@ def interpolation_at_time_events_get( base_headers, ): try: - (connection, parameters) = common_api_setup_tasks( + connection, parameters = common_api_setup_tasks( base_query_parameters, tag_query_parameters=tag_query_parameters, interpolation_at_time_query_parameters=interpolation_at_time_query_parameters, diff --git a/src/api/v1/latest.py b/src/api/v1/latest.py index e39bb4ed7..9d36058b1 100644 --- a/src/api/v1/latest.py +++ b/src/api/v1/latest.py @@ -35,7 +35,7 @@ def latest_retrieval_get( query_parameters, metadata_query_parameters, limit_offset_parameters, base_headers ): try: - (connection, parameters) = common_api_setup_tasks( + connection, parameters = common_api_setup_tasks( query_parameters, metadata_query_parameters=metadata_query_parameters, limit_offset_query_parameters=limit_offset_parameters, diff --git a/src/api/v1/metadata.py b/src/api/v1/metadata.py index 4470e8dca..6f0f7d057 100644 --- a/src/api/v1/metadata.py +++ b/src/api/v1/metadata.py @@ -33,7 +33,7 @@ def metadata_retrieval_get( query_parameters, metadata_query_parameters, limit_offset_parameters, base_headers ): try: - (connection, parameters) = common_api_setup_tasks( + connection, parameters = common_api_setup_tasks( query_parameters, metadata_query_parameters=metadata_query_parameters, limit_offset_query_parameters=limit_offset_parameters, diff --git a/src/api/v1/models.py b/src/api/v1/models.py index 000a517a5..5ea934d63 100644 --- a/src/api/v1/models.py +++ b/src/api/v1/models.py @@ -30,7 +30,6 @@ from src.api.auth.azuread import oauth2_scheme from typing import Generic, TypeVar, Optional - EXAMPLE_DATE = "2022-01-01" EXAMPLE_DATETIME = "2022-01-01T15:00:00" EXAMPLE_DATETIME_TIMEZOME = "2022-01-01T15:00:00+00:00" @@ -231,10 +230,10 @@ def __init__( class BaseQueryParams: def __init__( self, - business_unit: str = Query(None, description="Business Unit Name"), + business_unit: str = Query(..., description="Business Unit Name"), region: str = Query(..., description="Region"), - asset: str = Query(None, description="Asset"), - data_security_level: str = Query(None, description="Data Security Level"), + asset: str = Query(..., description="Asset"), + data_security_level: str = Query(..., description="Data Security Level"), authorization: str = Depends(oauth2_scheme), ): # Additional validation when mapping endpoint not provided - ensure validation error for missing params @@ -300,7 +299,7 @@ class RawQueryParams: def __init__( self, data_type: str = Query( - None, + ..., description="Data Type can be one of the following options: float, double, integer, string", examples=["float", "double", "integer", "string"], ), @@ -447,7 +446,7 @@ def __init__( time_interval_rate: str = DuplicatedQueryParameters.time_interval_rate, time_interval_unit: str = DuplicatedQueryParameters.time_interval_unit, window_length: int = Query( - ..., description="Window Length in days", examples=[1] + default=1, description="Window Length in days", examples=[1] ), step: str = Query( default="metadata", diff --git a/src/api/v1/plot.py b/src/api/v1/plot.py index 63378914b..7fddf840b 100644 --- a/src/api/v1/plot.py +++ b/src/api/v1/plot.py @@ -43,7 +43,7 @@ def plot_events_get( base_headers, ): try: - (connection, parameters) = common_api_setup_tasks( + connection, parameters = common_api_setup_tasks( base_query_parameters, raw_query_parameters=raw_query_parameters, tag_query_parameters=tag_query_parameters, diff --git a/src/api/v1/raw.py b/src/api/v1/raw.py index 2267a4151..97a08fa06 100644 --- a/src/api/v1/raw.py +++ b/src/api/v1/raw.py @@ -39,7 +39,7 @@ def raw_events_get( base_headers, ): try: - (connection, parameters) = common_api_setup_tasks( + connection, parameters = common_api_setup_tasks( base_query_parameters, raw_query_parameters=raw_query_parameters, tag_query_parameters=tag_query_parameters, diff --git a/src/api/v1/resample.py b/src/api/v1/resample.py index d3789a72a..6c08f07cc 100644 --- a/src/api/v1/resample.py +++ b/src/api/v1/resample.py @@ -45,7 +45,7 @@ def resample_events_get( base_headers, ): try: - (connection, parameters) = common_api_setup_tasks( + connection, parameters = common_api_setup_tasks( base_query_parameters, raw_query_parameters=raw_query_parameters, tag_query_parameters=tag_query_parameters, diff --git a/src/api/v1/sql.py b/src/api/v1/sql.py index 7b9c36ceb..307a051df 100644 --- a/src/api/v1/sql.py +++ b/src/api/v1/sql.py @@ -37,7 +37,7 @@ def sql_get( base_headers, ): try: - (connection, parameters) = common_api_setup_tasks( + connection, parameters = common_api_setup_tasks( base_query_parameters, sql_query_parameters=sql_query_parameters, limit_offset_query_parameters=limit_offset_parameters, diff --git a/src/api/v1/summary.py b/src/api/v1/summary.py index ce8400e63..68e46f7fc 100644 --- a/src/api/v1/summary.py +++ b/src/api/v1/summary.py @@ -39,7 +39,7 @@ def summary_events_get( base_headers, ): try: - (connection, parameters) = common_api_setup_tasks( + connection, parameters = common_api_setup_tasks( base_query_parameters, summary_query_parameters=summary_query_parameters, tag_query_parameters=tag_query_parameters, diff --git a/src/api/v1/time_weighted_average.py b/src/api/v1/time_weighted_average.py index dac0759cc..22a12fb2c 100644 --- a/src/api/v1/time_weighted_average.py +++ b/src/api/v1/time_weighted_average.py @@ -43,7 +43,7 @@ def time_weighted_average_events_get( base_headers, ): try: - (connection, parameters) = common_api_setup_tasks( + connection, parameters = common_api_setup_tasks( base_query_parameters, raw_query_parameters=raw_query_parameters, tag_query_parameters=tag_query_parameters, diff --git a/src/sdk/python/rtdip_sdk/data_models/storage_objects/storage_objects_utils.py b/src/sdk/python/rtdip_sdk/data_models/storage_objects/storage_objects_utils.py index 2bc653395..b0bd1d11b 100644 --- a/src/sdk/python/rtdip_sdk/data_models/storage_objects/storage_objects_utils.py +++ b/src/sdk/python/rtdip_sdk/data_models/storage_objects/storage_objects_utils.py @@ -49,10 +49,8 @@ def validate_uri(uri: str): return parsed_uri.scheme, parsed_uri.hostname, parsed_uri.path except Exception as ex: logging.error(ex) - raise SystemError( - f"Could not convert to valid tuple \ - or scheme not supported: {uri} {parsed_uri.scheme}" - ) + raise SystemError(f"Could not convert to valid tuple \ + or scheme not supported: {uri} {parsed_uri.scheme}") def get_supported_schema() -> list: diff --git a/src/sdk/python/rtdip_sdk/data_models/transformers/london_smart_meter_transformer_2_usage.py b/src/sdk/python/rtdip_sdk/data_models/transformers/london_smart_meter_transformer_2_usage.py index 63f3ab75f..e5279b599 100644 --- a/src/sdk/python/rtdip_sdk/data_models/transformers/london_smart_meter_transformer_2_usage.py +++ b/src/sdk/python/rtdip_sdk/data_models/transformers/london_smart_meter_transformer_2_usage.py @@ -18,7 +18,6 @@ import hashlib import time - series_id_str = "usage_series_id_001" output_header_str: str = "Uid,SeriesId,Timestamp,IntervalTimestamp,Value" diff --git a/src/sdk/python/rtdip_sdk/data_models/utils/timeseries_utils.py b/src/sdk/python/rtdip_sdk/data_models/utils/timeseries_utils.py index 767f64914..93cb2f325 100644 --- a/src/sdk/python/rtdip_sdk/data_models/utils/timeseries_utils.py +++ b/src/sdk/python/rtdip_sdk/data_models/utils/timeseries_utils.py @@ -22,7 +22,6 @@ import random import logging - type_checks = [ # (Type, Test) (int, int), diff --git a/src/sdk/python/rtdip_sdk/pipelines/_pipeline_utils/weather_ecmwf.py b/src/sdk/python/rtdip_sdk/pipelines/_pipeline_utils/weather_ecmwf.py index edab618e1..b4c17eb64 100755 --- a/src/sdk/python/rtdip_sdk/pipelines/_pipeline_utils/weather_ecmwf.py +++ b/src/sdk/python/rtdip_sdk/pipelines/_pipeline_utils/weather_ecmwf.py @@ -15,7 +15,6 @@ BooleanType, ) - RTDIP_FLOAT_WEATHER_DATA_MODEL = StructType( [ StructField("TagName", StringType(), False), diff --git a/src/sdk/python/rtdip_sdk/pipelines/deploy/databricks.py b/src/sdk/python/rtdip_sdk/pipelines/deploy/databricks.py index fb3f2617f..dd43bff81 100644 --- a/src/sdk/python/rtdip_sdk/pipelines/deploy/databricks.py +++ b/src/sdk/python/rtdip_sdk/pipelines/deploy/databricks.py @@ -395,7 +395,7 @@ def deploy(self) -> Union[bool, ValueError]: module = self._load_module( task.task_key + "file_upload", task.notebook_task.notebook_path ) - (task_libraries, spark_configuration) = PipelineComponentsGetUtility( + task_libraries, spark_configuration = PipelineComponentsGetUtility( module.__name__ ).execute() workspace_client.workspace.mkdirs(path=self.workspace_directory) @@ -415,7 +415,7 @@ def deploy(self) -> Union[bool, ValueError]: module = self._load_module( task.task_key + "file_upload", task.spark_python_task.python_file ) - (task_libraries, spark_configuration) = PipelineComponentsGetUtility( + task_libraries, spark_configuration = PipelineComponentsGetUtility( module ).execute() workspace_client.workspace.mkdirs(path=self.workspace_directory) diff --git a/src/sdk/python/rtdip_sdk/pipelines/transformers/spark/mirico_json_to_metadata.py b/src/sdk/python/rtdip_sdk/pipelines/transformers/spark/mirico_json_to_metadata.py index 6d7ef0158..28e89e20c 100644 --- a/src/sdk/python/rtdip_sdk/pipelines/transformers/spark/mirico_json_to_metadata.py +++ b/src/sdk/python/rtdip_sdk/pipelines/transformers/spark/mirico_json_to_metadata.py @@ -105,15 +105,13 @@ def transform(self) -> DataFrame: tag_name_expr.alias("TagName"), lit("").alias("Description"), lit("").alias("UoM"), - expr( - """struct( + expr("""struct( body.retroAltitude, body.retroLongitude, body.retroLatitude, body.sensorAltitude, body.sensorLongitude, - body.sensorLatitude)""" - ).alias("Properties"), + body.sensorLatitude)""").alias("Properties"), ).dropDuplicates(["TagName"]) return df.select("TagName", "Description", "UoM", "Properties") diff --git a/src/sdk/python/rtdip_sdk/pipelines/utilities/spark/session.py b/src/sdk/python/rtdip_sdk/pipelines/utilities/spark/session.py index a9bfa98e7..a4c9e5f63 100644 --- a/src/sdk/python/rtdip_sdk/pipelines/utilities/spark/session.py +++ b/src/sdk/python/rtdip_sdk/pipelines/utilities/spark/session.py @@ -86,7 +86,7 @@ def settings() -> dict: def execute(self) -> SparkSession: """To execute""" try: - (task_libraries, spark_configuration) = PipelineComponentsGetUtility( + task_libraries, spark_configuration = PipelineComponentsGetUtility( self.module, self.config ).execute() self.spark = SparkClient( diff --git a/src/sdk/python/rtdip_sdk/queries/time_series/_time_series_query_builder.py b/src/sdk/python/rtdip_sdk/queries/time_series/_time_series_query_builder.py index 3797e5877..7ee635fea 100644 --- a/src/sdk/python/rtdip_sdk/queries/time_series/_time_series_query_builder.py +++ b/src/sdk/python/rtdip_sdk/queries/time_series/_time_series_query_builder.py @@ -180,6 +180,134 @@ def _build_raw_query( return raw_query_sql +def _build_raw_query_for_interpolate( + sql_query_name, + timestamp_column, + tagname_column, + status_column, + value_column, + start_date, + end_date, + time_zone, + time_interval_unit, + deduplicate=None, + source=None, + business_unit=None, + asset=None, + data_security_level=None, + data_type=None, + tag_names=None, + include_status=None, + include_bad_data=None, + case_insensitivity_tag_search=None, + sort=True, +): + """Build raw query for interpolation - same as _build_raw_query but with time gradient buffering. + + Time gradient buffering expands the query range to ensure accurate interpolation at boundaries. + For each time_interval_unit, a corresponding buffer is applied: + - minute → ±5 minutes + - second → ±60 seconds + - hour → ±1 hour + - day → ±1 day + + This ensures we have data points just outside the requested range for proper linear interpolation + at the boundary timestamps. Only data within the original requested range is returned to the user. + """ + # Determine time gradient based on time_interval_unit + time_gradient_map = { + "minute": (5, "minute"), + "second": (60, "second"), + "hour": (1, "hour"), + "day": (1, "day"), + } + + gradient_value, gradient_unit = time_gradient_map.get( + time_interval_unit, (5, "minute") + ) + + # Select + raw_query_sql = f"{sql_query_name} AS (SELECT" + if deduplicate == True: + raw_query_sql = " ".join([raw_query_sql, "DISTINCT"]) + + # Event Time + raw_query_sql = " ".join( + [ + raw_query_sql, + f"from_utc_timestamp(date_trunc('millisecond',`{timestamp_column}`), '{time_zone}') AS `{timestamp_column}`,", + ] + ) + + # Tag Name + raw_query_sql = " ".join([raw_query_sql, f"`{tagname_column}`,"]) + + # Status + if include_status == True: + raw_query_sql = " ".join([raw_query_sql, f"`{status_column}`,"]) + else: + raw_query_sql = " ".join([raw_query_sql, "'Good' AS `Status`,"]) + + # Value + raw_query_sql = " ".join([raw_query_sql, f"`{value_column}` FROM"]) + + if source is not None: + raw_query_sql = " ".join([raw_query_sql, f"`{source.lower()}`"]) + else: + raw_query_sql = " ".join( + [ + raw_query_sql, + f"`{business_unit.lower()}`.`sensors`.`{asset.lower()}_{data_security_level.lower()}_events_{data_type.lower()}`", + ] + ) + + # Where - time gradient buffering for accurate boundary interpolation + # Expands query range by ±gradient_value in gradient_unit to fetch data outside the requested range + # This ensures linear interpolation at boundary timestamps has sufficient context + # Example: for minute intervals, query range is expanded by ±5 minutes = ±300 seconds + raw_query_sql = " ".join( + [ + raw_query_sql, + f"WHERE `{timestamp_column}` BETWEEN", + f"timestampadd({gradient_unit}, -{gradient_value}, to_timestamp('{start_date}'))", + f"AND", + f"timestampadd({gradient_unit}, {gradient_value}, to_timestamp('{end_date}'))", + "AND", + ] + ) + + if case_insensitivity_tag_search == True: + quoted_tag_names = "', '".join([tag.upper() for tag in tag_names]) + raw_query_sql = " ".join( + [ + raw_query_sql, + f"UPPER(`{tagname_column}`) IN ('{quoted_tag_names}')", + ] + ) + else: + quoted_tag_names = "', '".join(tag_names) + raw_query_sql = " ".join( + [ + raw_query_sql, + f"`{tagname_column}` IN ('{quoted_tag_names}')", + ] + ) + + if include_status == True and include_bad_data == False: + raw_query_sql = " ".join([raw_query_sql, f"AND `{status_column}` <> 'Bad'"]) + + if sort == True: + raw_query_sql = " ".join( + [ + raw_query_sql, + f"ORDER BY `{tagname_column}`, `{timestamp_column}`", + ] + ) + raw_query_sql += ")" + + return raw_query_sql + + def _build_resample_query( sql_query_list, sql_query_name, @@ -266,39 +394,174 @@ def _build_fill_intervals_query( return intervals_query + fill_intervals_query + from_sql + ")" +def _build_interpolate_intervals_cte( + timestamp_column, + tagname_column, + tag_names, + start_date, + end_date, + time_zone, + time_interval_rate, + time_interval_unit, + case_insensitivity_tag_search, +): + """Build intervals CTE with time sequence and tag explosion.""" + quoted_tag_names = ( + "', '".join([tag.upper() for tag in tag_names]) + if case_insensitivity_tag_search == True + else "', '".join(tag_names) + ) + return f"intervals AS (SELECT DISTINCT explode(sequence(from_utc_timestamp(to_timestamp('{start_date}'), '{time_zone}'), from_utc_timestamp(to_timestamp('{end_date}'), '{time_zone}'), INTERVAL '{time_interval_rate} {time_interval_unit}')) AS `{timestamp_column}`, explode(array('{quoted_tag_names}')) AS `{tagname_column}`)" + + +def _build_interpolate_fill_intervals_cte( + timestamp_column, + tagname_column, + value_column, +): + """Build fill_intervals CTE that unions raw data with intervals.""" + return f"fill_intervals AS (SELECT `{timestamp_column}`, `{tagname_column}`, `{value_column}`, 0 AS `SortKey` FROM raw UNION ALL SELECT `{timestamp_column}`, `{tagname_column}`, CAST(NULL AS DOUBLE), 1 AS `SortKey` FROM intervals)" + + +def _build_interpolate_calculate_cte( + timestamp_column, + tagname_column, + value_column, +): + """Build interpolate_calculate CTE that applies LAG/LEAD to capture previous/next values.""" + return f"interpolate_calculate AS (SELECT `{timestamp_column}`, `{tagname_column}`, `SortKey`, `{value_column}`, LAG(CASE WHEN `SortKey` = 0 THEN struct(`{timestamp_column}` AS `{timestamp_column}`, `{value_column}` AS `{value_column}`) END) IGNORE NULLS OVER (PARTITION BY `{tagname_column}` ORDER BY `{timestamp_column}`, `SortKey`) AS `Prev`, LEAD(CASE WHEN `SortKey` = 0 THEN struct(`{timestamp_column}` AS `{timestamp_column}`, `{value_column}` AS `{value_column}`) END) IGNORE NULLS OVER (PARTITION BY `{tagname_column}` ORDER BY `{timestamp_column}`, `SortKey`) AS `Next` FROM fill_intervals)" + + +def _build_interpolate_interpolate_cte( + timestamp_column, + tagname_column, + value_column, +): + """Build interpolate CTE that performs linear interpolation using unix_millis.""" + return f"interpolate AS (SELECT `{timestamp_column}`, `{tagname_column}`, CASE WHEN `Prev` IS NOT NULL AND `Next` IS NOT NULL THEN `Prev`.`{value_column}` + ((`Next`.`{value_column}` - `Prev`.`{value_column}`) * (unix_millis(`{timestamp_column}`) - unix_millis(`Prev`.`{timestamp_column}`)) / (unix_millis(`Next`.`{timestamp_column}`) - unix_millis(`Prev`.`{timestamp_column}`))) WHEN `Prev` IS NOT NULL THEN `Prev`.`{value_column}` ELSE NULL END AS `{value_column}` FROM interpolate_calculate WHERE `SortKey` = 1)" + + +def _build_interpolate_uom_cte( + sql_query_name, + timestamp_column, + tagname_column, + value_column, + metadata_source, + business_unit, + asset, + data_security_level, + metadata_tagname_column, + metadata_uom_column, +): + """Build final uom CTE that joins with metadata for UoM.""" + uom_query_sql = f"{sql_query_name} AS (SELECT i.`{timestamp_column}`, i.`{tagname_column}`, i.`{value_column}`, m.`{metadata_uom_column}` FROM interpolate i LEFT OUTER JOIN" + + if metadata_source: + uom_query_sql = " ".join([uom_query_sql, f"{metadata_source}"]) + else: + uom_query_sql = " ".join( + [ + uom_query_sql, + f"`{business_unit.lower()}`.`sensors`.`{asset.lower()}_{data_security_level.lower()}_metadata`", + ] + ) + + uom_query_sql = " ".join( + [ + uom_query_sql, + f"AS m ON i.`{tagname_column}` = m.`{metadata_tagname_column}`)", + ] + ) + + return uom_query_sql + + def _build_interpolate_query( sql_query_list, - sql_query_name, tagname_column, timestamp_column, value_column, - sort=True, + tag_names, + start_date, + end_date, + time_zone, + time_interval_rate, + time_interval_unit, + metadata_source, + business_unit, + asset, + data_security_level, + metadata_tagname_column, + metadata_uom_column, + case_insensitivity_tag_search, + display_uom=False, ): - parent_sql_query_name = sql_query_list[-1]["query_name"] + """Build the complete interpolate query with all CTEs. + + Returns tuple of (query_sql, final_cte_name): + - query_sql: The CTE string to be added to sql_query_list + - final_cte_name: The name of the final CTE ('interpolate' or 'uom') + """ + # Build individual CTEs using dedicated functions + intervals_query_sql = _build_interpolate_intervals_cte( + timestamp_column=timestamp_column, + tagname_column=tagname_column, + tag_names=tag_names, + start_date=start_date, + end_date=end_date, + time_zone=time_zone, + time_interval_rate=time_interval_rate, + time_interval_unit=time_interval_unit, + case_insensitivity_tag_search=case_insensitivity_tag_search, + ) - interpolate_calc_query_sql = f"{sql_query_name}_calculate AS (SELECT `Original{timestamp_column}`, `{timestamp_column}`, `{tagname_column}`, " - lag_value_query_sql = f"CASE WHEN `{value_column}` IS NOT NULL THEN NULL ELSE LAG(`{timestamp_column}_{value_column}`) IGNORE NULLS OVER (PARTITION BY `{tagname_column}` ORDER BY `{timestamp_column}`) END AS Prev{timestamp_column}{value_column}, " - lead_value_query_sql = f"CASE WHEN `{value_column}` IS NOT NULL THEN NULL ELSE LEAD(`{timestamp_column}_{value_column}`) IGNORE NULLS OVER (PARTITION BY `{tagname_column}` ORDER BY `{timestamp_column}`) END AS Next{timestamp_column}{value_column}, " - value_query_sql = f"CASE WHEN `Original{timestamp_column}` = `{timestamp_column}` THEN `{value_column}` WHEN `Prev{timestamp_column}{value_column}` IS NOT NULL AND `Next{timestamp_column}{value_column}` IS NOT NULL THEN `Prev{timestamp_column}{value_column}`.`{value_column}` + ((`Next{timestamp_column}{value_column}`.`{value_column}` - `Prev{timestamp_column}{value_column}`.`{value_column}`) * (unix_timestamp(`{timestamp_column}`) - unix_timestamp(`Prev{timestamp_column}{value_column}`.`{timestamp_column}`)) / (unix_timestamp(`Next{timestamp_column}{value_column}`.`{timestamp_column}`) - unix_timestamp(`Prev{timestamp_column}{value_column}`.`{timestamp_column}`))) WHEN `Prev{timestamp_column}{value_column}` IS NOT NULL THEN `Prev{timestamp_column}{value_column}`.`{value_column}` ELSE NULL END as `{value_column}` FROM {parent_sql_query_name} " - interpolate_project_query_sql = f"), {sql_query_name} AS (SELECT `{timestamp_column}`, `{tagname_column}`, `{value_column}` FROM {sql_query_name}_calculate WHERE `Original{timestamp_column}` IS NULL OR `Original{timestamp_column}` = `{timestamp_column}` " - - interpolate_query_sql = ( - interpolate_calc_query_sql - + lag_value_query_sql - + lead_value_query_sql - + value_query_sql - + interpolate_project_query_sql + fill_intervals_query_sql = _build_interpolate_fill_intervals_cte( + timestamp_column=timestamp_column, + tagname_column=tagname_column, + value_column=value_column, ) - if sort == True: - interpolate_query_sql = " ".join( - [ - interpolate_query_sql, - f"ORDER BY `{tagname_column}`, `{timestamp_column}`", - ] + interpolate_calculate_query_sql = _build_interpolate_calculate_cte( + timestamp_column=timestamp_column, + tagname_column=tagname_column, + value_column=value_column, + ) + + interpolate_query_sql_cte = _build_interpolate_interpolate_cte( + timestamp_column=timestamp_column, + tagname_column=tagname_column, + value_column=value_column, + ) + + # Combine CTEs - conditionally include UOM + cte_list = [ + intervals_query_sql, + fill_intervals_query_sql, + interpolate_calculate_query_sql, + interpolate_query_sql_cte, + ] + + final_cte_name = "interpolate" + + if display_uom: + uom_query_sql = _build_interpolate_uom_cte( + sql_query_name="uom", + timestamp_column=timestamp_column, + tagname_column=tagname_column, + value_column=value_column, + metadata_source=metadata_source, + business_unit=business_unit, + asset=asset, + data_security_level=data_security_level, + metadata_tagname_column=metadata_tagname_column, + metadata_uom_column=metadata_uom_column, ) + cte_list.append(uom_query_sql) + final_cte_name = "uom" + + interpolate_query_sql = ", ".join(cte_list) - return interpolate_query_sql + ")" + return interpolate_query_sql, final_cte_name def _build_summary_query( @@ -462,7 +725,9 @@ def _build_uom_query( return uom_sql_query + ")" -def _build_output_query(sql_query_list, to_json, limit, offset): +def _build_output_query( + sql_query_list, to_json, limit, offset, columns=None, order_by_clause=None +): parent_sql_query_name = sql_query_list[-1]["query_name"] output_sql_query = f"SELECT" @@ -476,11 +741,16 @@ def _build_output_query(sql_query_list, to_json, limit, offset): ")) AS Value", ] ) + elif columns is not None: + output_sql_query = " ".join([output_sql_query, ", ".join(columns)]) else: output_sql_query = " ".join([output_sql_query, "*"]) output_sql_query = " ".join([output_sql_query, f"FROM {parent_sql_query_name}"]) + if order_by_clause is not None: + output_sql_query = " ".join([output_sql_query, order_by_clause]) + if limit is not None: output_sql_query = " ".join([output_sql_query, f"LIMIT {limit}"]) @@ -631,7 +901,7 @@ def _sample_query_parameters(parameters_dict: dict) -> dict: "include_bad_data": parameters_dict["include_bad_data"], "time_interval_rate": parameters_dict["time_interval_rate"], "time_interval_unit": parameters_dict["time_interval_unit"], - "agg_method": parameters_dict["agg_method"], + "agg_method": parameters_dict.get("agg_method", None), "fill": parameters_dict.get("fill", False), "time_zone": parameters_dict["time_zone"], "pivot": parameters_dict.get("pivot", None), @@ -907,13 +1177,11 @@ def _plot_query_parameters(parameters_dict: dict) -> dict: def _interpolation_query(parameters_dict: dict) -> str: - parameters_dict["agg_method"] = None - interpolate_parameters = _sample_query_parameters(parameters_dict) sql_query_list = [] - raw_query = _build_raw_query( + raw_query = _build_raw_query_for_interpolate( sql_query_name="raw", timestamp_column=interpolate_parameters["timestamp_column"], tagname_column=interpolate_parameters["tagname_column"], @@ -921,10 +1189,9 @@ def _interpolation_query(parameters_dict: dict) -> str: value_column=interpolate_parameters["value_column"], start_date=interpolate_parameters["start_date"], end_date=interpolate_parameters["end_date"], - time_interval_rate=interpolate_parameters["time_interval_rate"], + time_zone=interpolate_parameters.get("time_zone", "+0000"), time_interval_unit=interpolate_parameters["time_interval_unit"], - agg_method=None, - time_zone=interpolate_parameters["time_zone"], + deduplicate=True, source=interpolate_parameters["source"], business_unit=interpolate_parameters["business_unit"], asset=interpolate_parameters["asset"], @@ -932,6 +1199,7 @@ def _interpolation_query(parameters_dict: dict) -> str: data_type=interpolate_parameters["data_type"], tag_names=interpolate_parameters["tag_names"], include_status=interpolate_parameters["include_status"], + include_bad_data=interpolate_parameters["include_bad_data"], case_insensitivity_tag_search=interpolate_parameters[ "case_insensitivity_tag_search" ], @@ -940,32 +1208,10 @@ def _interpolation_query(parameters_dict: dict) -> str: sql_query_list.append({"query_name": "raw", "sql_query": raw_query}) - # resample_query = _build_resample_query( - # sql_query_list=sql_query_list, - # sql_query_name="resample", - # timestamp_column=interpolate_parameters["timestamp_column"], - # tagname_column=interpolate_parameters["tagname_column"], - # value_column=interpolate_parameters["value_column"], - # tag_names=interpolate_parameters["tag_names"], - # start_date=interpolate_parameters["start_date"], - # end_date=interpolate_parameters["end_date"], - # time_zone=interpolate_parameters["time_zone"], - # time_interval_rate=interpolate_parameters["time_interval_rate"], - # time_interval_unit=interpolate_parameters["time_interval_unit"], - # agg_method=interpolate_parameters["agg_method"], - # case_insensitivity_tag_search=interpolate_parameters[ - # "case_insensitivity_tag_search" - # ], - # fill=True, - # sort=False, - # ) - - # sql_query_list.append({"query_name": "resample", "sql_query": resample_query}) - fill_intervals_query = _build_fill_intervals_query( + interpolate_query, final_cte_name = _build_interpolate_query( sql_query_list=sql_query_list, - sql_query_name="fill_intervals", - timestamp_column=interpolate_parameters["timestamp_column"], tagname_column=interpolate_parameters["tagname_column"], + timestamp_column=interpolate_parameters["timestamp_column"], value_column=interpolate_parameters["value_column"], tag_names=interpolate_parameters["tag_names"], start_date=interpolate_parameters["start_date"], @@ -973,30 +1219,22 @@ def _interpolation_query(parameters_dict: dict) -> str: time_zone=interpolate_parameters["time_zone"], time_interval_rate=interpolate_parameters["time_interval_rate"], time_interval_unit=interpolate_parameters["time_interval_unit"], + metadata_source=interpolate_parameters["metadata_source"], + business_unit=interpolate_parameters["business_unit"], + asset=interpolate_parameters["asset"], + data_security_level=interpolate_parameters["data_security_level"], + metadata_tagname_column=interpolate_parameters["metadata_tagname_column"], + metadata_uom_column=interpolate_parameters["metadata_uom_column"], case_insensitivity_tag_search=interpolate_parameters[ "case_insensitivity_tag_search" ], + display_uom=interpolate_parameters["display_uom"], ) sql_query_list.append( - {"query_name": "fill_intervals", "sql_query": fill_intervals_query} - ) - - interpolate_query = _build_interpolate_query( - sql_query_list=sql_query_list, - sql_query_name="interpolate", - timestamp_column=interpolate_parameters["timestamp_column"], - tagname_column=interpolate_parameters["tagname_column"], - value_column=interpolate_parameters["value_column"], - sort=( - interpolate_parameters["sort"] - if interpolate_parameters["pivot"] == False - else False - ), + {"query_name": final_cte_name, "sql_query": interpolate_query} ) - sql_query_list.append({"query_name": "interpolate", "sql_query": interpolate_query}) - if interpolate_parameters["pivot"] == True: pivot_query = _build_pivot_query( sql_query_list=sql_query_list, @@ -1013,26 +1251,32 @@ def _interpolation_query(parameters_dict: dict) -> str: sql_query_list.append({"query_name": "pivot", "sql_query": pivot_query}) - if interpolate_parameters["display_uom"] == True: - uom_query = _build_uom_query( - sql_query_list=sql_query_list, - sql_query_name="uom", - metadata_source=interpolate_parameters["metadata_source"], - business_unit=interpolate_parameters["business_unit"], - asset=interpolate_parameters["asset"], - data_security_level=interpolate_parameters["data_security_level"], - tagname_column=interpolate_parameters["tagname_column"], - metadata_tagname_column=interpolate_parameters["metadata_tagname_column"], - metadata_uom_column=interpolate_parameters["metadata_uom_column"], - ) - - sql_query_list.append({"query_name": "uom", "sql_query": uom_query}) + # Build order by clause if needed + order_by_clause = None + if interpolate_parameters["sort"] and interpolate_parameters["pivot"] == False: + order_by_clause = f"ORDER BY `{interpolate_parameters['tagname_column']}`, `{interpolate_parameters['timestamp_column']}`" + + # Build columns list for explicit selection only when final CTE is interpolate + # (not uom or pivot, which already have their specific columns) + columns = None + if ( + final_cte_name == "interpolate" + and interpolate_parameters["display_uom"] == False + and interpolate_parameters["pivot"] == False + ): + columns = [ + f"`{interpolate_parameters['timestamp_column']}`", + f"`{interpolate_parameters['tagname_column']}`", + f"`{interpolate_parameters['value_column']}`", + ] output_query = _build_output_query( sql_query_list=sql_query_list, to_json=interpolate_parameters["to_json_resample"], limit=interpolate_parameters["limit"], offset=interpolate_parameters["offset"], + columns=columns, + order_by_clause=order_by_clause, ) sql_query_list.append({"query_name": "output", "sql_query": output_query}) diff --git a/src/sdk/python/rtdip_sdk/queries/time_series/interpolate.py b/src/sdk/python/rtdip_sdk/queries/time_series/interpolate.py index eecf42dfc..76dc77324 100644 --- a/src/sdk/python/rtdip_sdk/queries/time_series/interpolate.py +++ b/src/sdk/python/rtdip_sdk/queries/time_series/interpolate.py @@ -51,6 +51,20 @@ def get(connection: object, parameters_dict: dict) -> pd.DataFrame: Returns: DataFrame: A resampled and interpolated dataframe. + !!! warning + **Time Gradient Buffering** + + The interpolation query automatically expands the requested date range with a time-based buffer to ensure accurate interpolation at the boundaries. The buffer size depends on the `time_interval_unit`: + + - **minute**: ±5 minutes + - **second**: ±60 seconds + - **hour**: ±1 hour + - **day**: ±1 day + + **Example**: If you request data from 2024-01-01 00:00:00 to 2024-01-01 23:59:59 with minute intervals, the query will fetch data from 2023-12-31 23:55:00 to 2024-01-02 00:04:59. Only data within your requested range will be returned, but the buffer ensures interpolated values at your boundaries are accurate. + + **Important**: In cases where data collection is infrequent (e.g., sparse sensor data), the default buffer boundaries may not cover sufficient historical/future data points for accurate interpolation. If you encounter interpolation gaps or inaccurate boundary values, consider expanding your query window to include more data around your requested date range. This ensures sufficient data points exist within the buffer for computing accurate interpolated values. + !!! warning Setting `case_insensitivity_tag_search` to True will result in a longer query time. diff --git a/src/sdk/python/rtdip_sdk/queries/time_series/time_weighted_average.py b/src/sdk/python/rtdip_sdk/queries/time_series/time_weighted_average.py index be1083e8d..9b176a114 100644 --- a/src/sdk/python/rtdip_sdk/queries/time_series/time_weighted_average.py +++ b/src/sdk/python/rtdip_sdk/queries/time_series/time_weighted_average.py @@ -39,7 +39,7 @@ def get(connection: object, parameters_dict: dict) -> pd.DataFrame: end_date (str): End date (Either a utc date in the format YYYY-MM-DD or a utc datetime in the format YYYY-MM-DDTHH:MM:SS or specify the timezone offset in the format YYYY-MM-DDTHH:MM:SS+zz:zz) time_interval_rate (str): The time interval rate (numeric input) time_interval_unit (str): The time interval unit (second, minute, day, hour) - window_length (int): Add longer window time in days for the start or end of specified date to cater for edge cases. + window_length (int): Add longer window time in days for the start or end of specified date to cater for edge cases. Optional and defaults to 1 day. include_bad_data (bool): Include "Bad" data points with True or remove "Bad" data points with False step (str): data points with step "enabled" or "disabled". The options for step are "true", "false" or "metadata". "metadata" will retrieve the step value from the metadata table. display_uom (optional bool): Display the unit of measure with True or False. Does not apply to pivoted tables. Defaults to False diff --git a/tests/api/v1/api_test_objects.py b/tests/api/v1/api_test_objects.py index 2599a298c..5ed9f4779 100644 --- a/tests/api/v1/api_test_objects.py +++ b/tests/api/v1/api_test_objects.py @@ -154,11 +154,11 @@ TIME_WEIGHTED_AVERAGE_MOCKED_PARAMETER_DICT["time_interval_rate"] = "15" TIME_WEIGHTED_AVERAGE_MOCKED_PARAMETER_DICT["time_interval_unit"] = "minute" -TIME_WEIGHTED_AVERAGE_MOCKED_PARAMETER_DICT["window_length"] = 10 +TIME_WEIGHTED_AVERAGE_MOCKED_PARAMETER_DICT["window_length"] = 1 TIME_WEIGHTED_AVERAGE_MOCKED_PARAMETER_DICT["step"] = "metadata" TIME_WEIGHTED_AVERAGE_MOCKED_PARAMETER_ERROR_DICT["time_interval_rate"] = "15" TIME_WEIGHTED_AVERAGE_MOCKED_PARAMETER_ERROR_DICT["time_interval_unit"] = "minute" -TIME_WEIGHTED_AVERAGE_MOCKED_PARAMETER_ERROR_DICT["window_length"] = 10 +TIME_WEIGHTED_AVERAGE_MOCKED_PARAMETER_ERROR_DICT["window_length"] = 1 TIME_WEIGHTED_AVERAGE_MOCKED_PARAMETER_ERROR_DICT["step"] = "metadata" TIME_WEIGHTED_AVERAGE_POST_MOCKED_PARAMETER_DICT = ( diff --git a/tests/api/v1/test_api_circular_average.py b/tests/api/v1/test_api_circular_average.py index a31b727e9..88163c935 100644 --- a/tests/api/v1/test_api_circular_average.py +++ b/tests/api/v1/test_api_circular_average.py @@ -150,140 +150,3 @@ async def test_api_circular_average_post_error(mocker: MockerFixture, api_test_d assert response.status_code == 400 assert actual == '{"detail":"Error Connecting to Database"}' - - -async def test_api_circular_average_get_lookup_success(mocker: MockerFixture): - """ - Case when no business_unit, asset etc supplied so instead invokes tag lookup - """ - - test_data = pd.DataFrame( - { - "EventTime": [datetime.now(timezone.utc)], - "TagName": ["Tagname1"], - "Value": [1.5], - } - ) - - # Mock the batch method, which outputs test data in the form of an array of dfs - mock_method = "src.sdk.python.rtdip_sdk.queries.time_series.batch.get" - mock_method_return_data = [test_data] - mocker = mocker_setup( - mocker, - mock_method, - mock_method_return_data, - tag_mapping_data=MOCK_TAG_MAPPING_SINGLE, - ) - mocker.patch.dict( - os.environ, {"DATABRICKS_SERVING_ENDPOINT": MOCK_MAPPING_ENDPOINT_URL} - ) - - # Remove parameters so that runs lookup - modified_param_dict = CIRCULAR_AVERAGE_MOCKED_PARAMETER_DICT.copy() - del modified_param_dict["business_unit"] - - async with AsyncClient(transport=ASGITransport(app=app), base_url=BASE_URL) as ac: - actual = await ac.get( - MOCK_API_NAME, headers=TEST_HEADERS, params=modified_param_dict - ) - - expected = test_data.to_json(orient="table", index=False, date_unit="ns") - expected = ( - expected.replace(',"tz":"UTC"', "").rstrip("}") - + ',"pagination":{"limit":null,"offset":null,"next":null}}' - ) - - assert actual.text == expected - assert actual.status_code == 200 - - -async def test_api_circular_average_post_lookup_success(mocker: MockerFixture): - """ - Case when no business_unit, asset etc supplied so instead invokes tag lookup - """ - - test_data = pd.DataFrame( - { - "EventTime": [CIRCULAR_AVERAGE_MOCKED_PARAMETER_DICT["start_date"]], - "TagName": ["Tagname1"], - "Status": ["Good"], - "Value": [1.01], - } - ) - - # Mock the batch method, which outputs test data in the form of an array of dfs - mock_method = "src.sdk.python.rtdip_sdk.queries.time_series.batch.get" - mock_method_return_data = [test_data] - mocker = mocker_setup( - mocker, - mock_method, - mock_method_return_data, - tag_mapping_data=MOCK_TAG_MAPPING_SINGLE, - ) - mocker.patch.dict( - os.environ, {"DATABRICKS_SERVING_ENDPOINT": MOCK_MAPPING_ENDPOINT_URL} - ) - - # Remove parameters so that runs lookup - modified_param_dict = CIRCULAR_AVERAGE_MOCKED_PARAMETER_DICT.copy() - del modified_param_dict["business_unit"] - - async with AsyncClient(transport=ASGITransport(app=app), base_url=BASE_URL) as ac: - actual = await ac.post( - MOCK_API_NAME, - headers=TEST_HEADERS, - params=modified_param_dict, - json=CIRCULAR_AVERAGE_POST_BODY_MOCKED_PARAMETER_DICT, - ) - - expected = test_data.to_json(orient="table", index=False, date_unit="ns") - expected = ( - expected.replace(',"tz":"UTC"', "").rstrip("}") - + ',"pagination":{"limit":null,"offset":null,"next":null}}' - ) - - assert actual.text == expected - assert actual.status_code == 200 - - -async def test_api_circular_average_get_lookup_no_tag_map_error(mocker: MockerFixture): - """ - Case when no business_unit, asset etc supplied so instead invokes tag lookup - """ - - test_data = pd.DataFrame( - { - "EventTime": [CIRCULAR_AVERAGE_MOCKED_PARAMETER_DICT["start_date"]], - "TagName": ["Tagname1"], - "Status": ["Good"], - "Value": [1.01], - } - ) - - # Mock the batch method, which outputs test data in the form of an array of dfs - mock_method = "src.sdk.python.rtdip_sdk.queries.time_series.batch.get" - mock_method_return_data = [test_data] - mocker = mocker_setup( - mocker, - mock_method, - mock_method_return_data, - tag_mapping_data=MOCK_TAG_MAPPING_EMPTY, - ) - mocker.patch.dict( - os.environ, {"DATABRICKS_SERVING_ENDPOINT": MOCK_MAPPING_ENDPOINT_URL} - ) - - # Remove parameters so that runs lookup - modified_param_dict = CIRCULAR_AVERAGE_MOCKED_PARAMETER_DICT.copy() - modified_param_dict["tagname"] = ["NonExistentTag"] - del modified_param_dict["business_unit"] - - async with AsyncClient(transport=ASGITransport(app=app), base_url=BASE_URL) as ac: - actual = await ac.get( - MOCK_API_NAME, headers=TEST_HEADERS, params=modified_param_dict - ) - - expected = '{"detail":"One or more tags do not have tables associated with them, the data belongs to a confidential table, or you do not have access. If the tag belongs to a confidential table and you do have access, please supply the business_unit, asset, data_security_level and data_type"}' - - assert actual.text == expected - assert actual.status_code == 400 diff --git a/tests/api/v1/test_api_latest.py b/tests/api/v1/test_api_latest.py index 7c2e941db..69c7d9246 100644 --- a/tests/api/v1/test_api_latest.py +++ b/tests/api/v1/test_api_latest.py @@ -201,153 +201,3 @@ async def test_api_raw_post_error(mocker: MockerFixture, api_test_data): assert response.status_code == 400 assert actual == '{"detail":"Error Connecting to Database"}' - - -async def test_api_latest_get_lookup_success(mocker: MockerFixture): - """ - Case when no business_unit, asset etc supplied so instead invokes tag lookup - """ - - test_data = pd.DataFrame( - { - "TagName": ["TestTag"], - "EventTime": [datetime.now(timezone.utc)], - "Status": ["Good"], - "Value": ["1.01"], - "ValueType": ["string"], - "GoodEventTime": [datetime.now(timezone.utc)], - "GoodValue": ["1.01"], - "GoodValueType": ["string"], - } - ) - - # Mock the batch method, which outputs test data in the form of an array of dfs - mock_method = "src.sdk.python.rtdip_sdk.queries.time_series.batch.get" - mock_method_return_data = [test_data] - mocker = mocker_setup( - mocker, - mock_method, - mock_method_return_data, - tag_mapping_data=MOCK_TAG_MAPPING_SINGLE, - ) - mocker.patch.dict( - os.environ, {"DATABRICKS_SERVING_ENDPOINT": MOCK_MAPPING_ENDPOINT_URL} - ) - - # Remove parameters so that runs lookup - modified_param_dict = METADATA_MOCKED_PARAMETER_DICT.copy() - del modified_param_dict["business_unit"] - - async with AsyncClient(transport=ASGITransport(app=app), base_url=BASE_URL) as ac: - actual = await ac.get( - MOCK_API_NAME, headers=TEST_HEADERS, params=modified_param_dict - ) - - expected = test_data.to_json(orient="table", index=False, date_unit="ns") - expected = ( - expected.replace(',"tz":"UTC"', "").rstrip("}") - + ',"pagination":{"limit":null,"offset":null,"next":null}}' - ) - - assert actual.text == expected - assert actual.status_code == 200 - - -async def test_api_latest_post_lookup_success(mocker: MockerFixture): - """ - Case when no business_unit, asset etc supplied so instead invokes tag lookup - """ - - test_data = pd.DataFrame( - { - "TagName": ["TestTag"], - "EventTime": [datetime.now(timezone.utc)], - "Status": ["Good"], - "Value": ["1.01"], - "ValueType": ["string"], - "GoodEventTime": [datetime.now(timezone.utc)], - "GoodValue": ["1.01"], - "GoodValueType": ["string"], - } - ) - - # Mock the batch method, which outputs test data in the form of an array of dfs - mock_method = "src.sdk.python.rtdip_sdk.queries.time_series.batch.get" - mock_method_return_data = [test_data] - mocker = mocker_setup( - mocker, - mock_method, - mock_method_return_data, - tag_mapping_data=MOCK_TAG_MAPPING_SINGLE, - ) - mocker.patch.dict( - os.environ, {"DATABRICKS_SERVING_ENDPOINT": MOCK_MAPPING_ENDPOINT_URL} - ) - - # Remove parameters so that runs lookup - modified_param_dict = METADATA_MOCKED_PARAMETER_DICT.copy() - del modified_param_dict["business_unit"] - - async with AsyncClient(transport=ASGITransport(app=app), base_url=BASE_URL) as ac: - actual = await ac.post( - MOCK_API_NAME, - headers=TEST_HEADERS, - params=modified_param_dict, - json=METADATA_POST_BODY_MOCKED_PARAMETER_DICT, - ) - - expected = test_data.to_json(orient="table", index=False, date_unit="ns") - expected = ( - expected.replace(',"tz":"UTC"', "").rstrip("}") - + ',"pagination":{"limit":null,"offset":null,"next":null}}' - ) - - assert actual.text == expected - assert actual.status_code == 200 - - -async def test_api_latest_get_lookup_no_tag_map_error(mocker: MockerFixture): - """ - Case when no business_unit, asset etc supplied so instead invokes tag lookup - """ - - test_data = pd.DataFrame( - { - "TagName": ["TestTag"], - "EventTime": [datetime.now(timezone.utc)], - "Status": ["Good"], - "Value": ["1.01"], - "ValueType": ["string"], - "GoodEventTime": [datetime.now(timezone.utc)], - "GoodValue": ["1.01"], - "GoodValueType": ["string"], - } - ) - - # Mock the batch method, which outputs test data in the form of an array of dfs - mock_method = "src.sdk.python.rtdip_sdk.queries.time_series.batch.get" - mock_method_return_data = [test_data] - mocker = mocker_setup( - mocker, - mock_method, - mock_method_return_data, - tag_mapping_data=MOCK_TAG_MAPPING_EMPTY, - ) - mocker.patch.dict( - os.environ, {"DATABRICKS_SERVING_ENDPOINT": MOCK_MAPPING_ENDPOINT_URL} - ) - - # Remove parameters so that runs lookup - modified_param_dict = METADATA_MOCKED_PARAMETER_DICT.copy() - modified_param_dict["tagname"] = ["NonExistentTag"] - del modified_param_dict["business_unit"] - - async with AsyncClient(transport=ASGITransport(app=app), base_url=BASE_URL) as ac: - actual = await ac.get( - MOCK_API_NAME, headers=TEST_HEADERS, params=modified_param_dict - ) - - expected = '{"detail":"One or more tags do not have tables associated with them, the data belongs to a confidential table, or you do not have access. If the tag belongs to a confidential table and you do have access, please supply the business_unit, asset, data_security_level and data_type"}' - - assert actual.text == expected - assert actual.status_code == 400 diff --git a/tests/api/v1/test_api_metadata.py b/tests/api/v1/test_api_metadata.py index 585b41267..b8abadf43 100644 --- a/tests/api/v1/test_api_metadata.py +++ b/tests/api/v1/test_api_metadata.py @@ -188,114 +188,3 @@ async def test_api_metadata_post_error(mocker: MockerFixture, api_test_data): assert response.status_code == 400 assert actual == '{"detail":"Error Connecting to Database"}' - - -async def test_api_metadata_get_lookup_success(mocker: MockerFixture): - """ - Case when no business_unit, asset etc supplied so instead invokes tag lookup - """ - - # Mock the batch method, which outputs test data in the form of an array of dfs - mock_method = "src.sdk.python.rtdip_sdk.queries.time_series.batch.get" - mock_method_return_data = [TEST_DATA] - mocker = mocker_setup( - mocker, - mock_method, - mock_method_return_data, - tag_mapping_data=MOCK_TAG_MAPPING_SINGLE, - ) - mocker.patch.dict( - os.environ, {"DATABRICKS_SERVING_ENDPOINT": MOCK_MAPPING_ENDPOINT_URL} - ) - - # Remove parameters so that runs lookup - modified_param_dict = METADATA_MOCKED_PARAMETER_DICT.copy() - del modified_param_dict["business_unit"] - - async with AsyncClient(transport=ASGITransport(app=app), base_url=BASE_URL) as ac: - actual = await ac.get( - MOCK_API_NAME, headers=TEST_HEADERS, params=modified_param_dict - ) - - expected = TEST_DATA.to_json(orient="table", index=False, date_unit="ns") - expected = ( - expected.replace(',"tz":"UTC"', "").rstrip("}") - + ',"pagination":{"limit":null,"offset":null,"next":null}}' - ) - - assert actual.text == expected - assert actual.status_code == 200 - - -async def test_api_metadata_post_lookup_success(mocker: MockerFixture): - """ - Case when no business_unit, asset etc supplied so instead invokes tag lookup - """ - - # Mock the batch method, which outputs test data in the form of an array of dfs - mock_method = "src.sdk.python.rtdip_sdk.queries.time_series.batch.get" - mock_method_return_data = [TEST_DATA] - mocker = mocker_setup( - mocker, - mock_method, - mock_method_return_data, - tag_mapping_data=MOCK_TAG_MAPPING_SINGLE, - ) - mocker.patch.dict( - os.environ, {"DATABRICKS_SERVING_ENDPOINT": MOCK_MAPPING_ENDPOINT_URL} - ) - - # Remove parameters so that runs lookup - modified_param_dict = METADATA_MOCKED_PARAMETER_DICT.copy() - del modified_param_dict["business_unit"] - - async with AsyncClient(transport=ASGITransport(app=app), base_url=BASE_URL) as ac: - actual = await ac.post( - MOCK_API_NAME, - headers=TEST_HEADERS, - params=modified_param_dict, - json=METADATA_POST_BODY_MOCKED_PARAMETER_DICT, - ) - - expected = TEST_DATA.to_json(orient="table", index=False, date_unit="ns") - expected = ( - expected.replace(',"tz":"UTC"', "").rstrip("}") - + ',"pagination":{"limit":null,"offset":null,"next":null}}' - ) - - assert actual.text == expected - assert actual.status_code == 200 - - -async def test_api_metadata_get_lookup_no_tag_map_error(mocker: MockerFixture): - """ - Case when no business_unit, asset etc supplied so instead invokes tag lookup - """ - - # Mock the batch method, which outputs test data in the form of an array of dfs - mock_method = "src.sdk.python.rtdip_sdk.queries.time_series.batch.get" - mock_method_return_data = [TEST_DATA] - mocker = mocker_setup( - mocker, - mock_method, - mock_method_return_data, - tag_mapping_data=MOCK_TAG_MAPPING_EMPTY, - ) - mocker.patch.dict( - os.environ, {"DATABRICKS_SERVING_ENDPOINT": MOCK_MAPPING_ENDPOINT_URL} - ) - - # Remove parameters so that runs lookup - modified_param_dict = METADATA_MOCKED_PARAMETER_DICT.copy() - modified_param_dict["tagname"] = ["NonExistentTag"] - del modified_param_dict["business_unit"] - - async with AsyncClient(transport=ASGITransport(app=app), base_url=BASE_URL) as ac: - actual = await ac.get( - MOCK_API_NAME, headers=TEST_HEADERS, params=modified_param_dict - ) - - expected = '{"detail":"One or more tags do not have tables associated with them, the data belongs to a confidential table, or you do not have access. If the tag belongs to a confidential table and you do have access, please supply the business_unit, asset, data_security_level and data_type"}' - - assert actual.text == expected - assert actual.status_code == 400 diff --git a/tests/api/v1/test_api_raw.py b/tests/api/v1/test_api_raw.py index 93d681b66..b7501c8c1 100644 --- a/tests/api/v1/test_api_raw.py +++ b/tests/api/v1/test_api_raw.py @@ -139,140 +139,3 @@ async def test_api_raw_post_error(mocker: MockerFixture, api_test_data): assert response.status_code == 400 assert actual == '{"detail":"Error Connecting to Database"}' - - -async def test_api_raw_get_lookup_success(mocker: MockerFixture, api_test_data): - """ - Case when no business_unit, asset etc supplied so instead invokes tag lookup - """ - - test_data = pd.DataFrame( - { - "EventTime": [RAW_MOCKED_PARAMETER_DICT["start_date"]], - "TagName": ["Tagname1"], - "Status": ["Good"], - "Value": [1.01], - } - ) - - # Mock the batch method, which outputs test data in the form of an array of dfs - mock_method = "src.sdk.python.rtdip_sdk.queries.time_series.batch.get" - mock_method_return_data = [test_data] - mocker = mocker_setup( - mocker, - mock_method, - mock_method_return_data, - tag_mapping_data=MOCK_TAG_MAPPING_SINGLE, - ) - mocker.patch.dict( - os.environ, {"DATABRICKS_SERVING_ENDPOINT": MOCK_MAPPING_ENDPOINT_URL} - ) - - # Remove parameters so that runs lookup - modified_param_dict = RAW_MOCKED_PARAMETER_DICT.copy() - del modified_param_dict["business_unit"] - - async with AsyncClient(transport=ASGITransport(app=app), base_url=BASE_URL) as ac: - actual = await ac.get( - MOCK_API_NAME, headers=TEST_HEADERS, params=modified_param_dict - ) - - expected = test_data.to_json(orient="table", index=False, date_unit="ns") - expected = ( - expected.rstrip("}") + ',"pagination":{"limit":null,"offset":null,"next":null}}' - ) - - assert actual.text == expected - assert actual.status_code == 200 - - -async def test_api_raw_post_lookup_success(mocker: MockerFixture): - """ - Case when no business_unit, asset etc supplied so instead invokes tag lookup - """ - - test_data = pd.DataFrame( - { - "EventTime": [RAW_MOCKED_PARAMETER_DICT["start_date"]], - "TagName": ["Tagname1"], - "Status": ["Good"], - "Value": [1.01], - } - ) - - # Mock the batch method, which outputs test data in the form of an array of dfs - mock_method = "src.sdk.python.rtdip_sdk.queries.time_series.batch.get" - mock_method_return_data = [test_data] - mocker = mocker_setup( - mocker, - mock_method, - mock_method_return_data, - tag_mapping_data=MOCK_TAG_MAPPING_SINGLE, - ) - mocker.patch.dict( - os.environ, {"DATABRICKS_SERVING_ENDPOINT": MOCK_MAPPING_ENDPOINT_URL} - ) - - # Remove parameters so that runs lookup - modified_param_dict = RAW_POST_MOCKED_PARAMETER_DICT.copy() - del modified_param_dict["business_unit"] - - async with AsyncClient(transport=ASGITransport(app=app), base_url=BASE_URL) as ac: - actual = await ac.post( - MOCK_API_NAME, - headers=TEST_HEADERS, - params=modified_param_dict, - json=RAW_POST_BODY_MOCKED_PARAMETER_DICT, - ) - - expected = test_data.to_json(orient="table", index=False, date_unit="ns") - expected = ( - expected.rstrip("}") + ',"pagination":{"limit":null,"offset":null,"next":null}}' - ) - - assert actual.text == expected - assert actual.status_code == 200 - - -async def test_api_raw_get_lookup_no_tag_map_error(mocker: MockerFixture): - """ - Case when no business_unit, asset etc supplied so instead invokes tag lookup - AND there is no table associated with the tag which results in error. - """ - - test_data = pd.DataFrame( - { - "EventTime": [RAW_MOCKED_PARAMETER_DICT["start_date"]], - "TagName": ["Tagname1"], - "Status": ["Good"], - "Value": [1.01], - } - ) - - # Mock the batch method, which outputs test data in the form of an array of dfs - mock_method = "src.sdk.python.rtdip_sdk.queries.time_series.batch.get" - mock_method_return_data = [test_data] - mocker = mocker_setup( - mocker, - mock_method, - mock_method_return_data, - tag_mapping_data=MOCK_TAG_MAPPING_EMPTY, - ) - mocker.patch.dict( - os.environ, {"DATABRICKS_SERVING_ENDPOINT": MOCK_MAPPING_ENDPOINT_URL} - ) - - # Remove parameters so that runs lookup, and add tag that does not exist - modified_param_dict = RAW_MOCKED_PARAMETER_DICT.copy() - modified_param_dict["tagname"] = ["NonExistentTag"] - del modified_param_dict["business_unit"] - - async with AsyncClient(transport=ASGITransport(app=app), base_url=BASE_URL) as ac: - actual = await ac.get( - MOCK_API_NAME, headers=TEST_HEADERS, params=modified_param_dict - ) - - expected = '{"detail":"One or more tags do not have tables associated with them, the data belongs to a confidential table, or you do not have access. If the tag belongs to a confidential table and you do have access, please supply the business_unit, asset, data_security_level and data_type"}' - - assert actual.text == expected - assert actual.status_code == 400 diff --git a/tests/sdk/python/rtdip_sdk/pipelines/destinations/spark/test_kafka_eventhub.py b/tests/sdk/python/rtdip_sdk/pipelines/destinations/spark/test_kafka_eventhub.py index 5e59d5a65..976ffdc82 100644 --- a/tests/sdk/python/rtdip_sdk/pipelines/destinations/spark/test_kafka_eventhub.py +++ b/tests/sdk/python/rtdip_sdk/pipelines/destinations/spark/test_kafka_eventhub.py @@ -36,7 +36,6 @@ ArrayType, ) - kafka_configuration_dict = {"failOnDataLoss": "true", "startingOffsets": "earliest"} eventhub_connection_string = "Endpoint=sb://test.servicebus.windows.net/;SharedAccessKeyName=test;SharedAccessKey=test_key;EntityPath=test_eventhub" diff --git a/tests/sdk/python/rtdip_sdk/pipelines/forecasting/spark/test_k_nearest_neighbors.py b/tests/sdk/python/rtdip_sdk/pipelines/forecasting/spark/test_k_nearest_neighbors.py index 95d91c4bf..660ee272b 100644 --- a/tests/sdk/python/rtdip_sdk/pipelines/forecasting/spark/test_k_nearest_neighbors.py +++ b/tests/sdk/python/rtdip_sdk/pipelines/forecasting/spark/test_k_nearest_neighbors.py @@ -74,6 +74,42 @@ def sample_data(spark): "Bad", 0.55, ), + ( + "TAG1", + datetime.strptime("2024-01-02 20:07:46.000", "%Y-%m-%d %H:%M:%S.%f"), + "Good", + 0.36, + ), + ( + "TAG2", + datetime.strptime("2024-01-02 20:08:46.000", "%Y-%m-%d %H:%M:%S.%f"), + "Good", + 0.46, + ), + ( + "TAG1", + datetime.strptime("2024-01-02 20:09:46.000", "%Y-%m-%d %H:%M:%S.%f"), + "Bad", + 0.37, + ), + ( + "TAG2", + datetime.strptime("2024-01-02 20:10:46.000", "%Y-%m-%d %H:%M:%S.%f"), + "Good", + 0.56, + ), + ( + "TAG1", + datetime.strptime("2024-01-02 20:11:46.000", "%Y-%m-%d %H:%M:%S.%f"), + "Good", + 0.38, + ), + ( + "TAG2", + datetime.strptime("2024-01-02 20:12:46.000", "%Y-%m-%d %H:%M:%S.%f"), + "Bad", + 0.57, + ), ] return spark.createDataFrame(data, schema=SCHEMA) diff --git a/tests/sdk/python/rtdip_sdk/pipelines/transformers/spark/ecmwf/test_nc_extractbase_to_weather_data_model.py b/tests/sdk/python/rtdip_sdk/pipelines/transformers/spark/ecmwf/test_nc_extractbase_to_weather_data_model.py index 9c101f5ca..22b94a60b 100644 --- a/tests/sdk/python/rtdip_sdk/pipelines/transformers/spark/ecmwf/test_nc_extractbase_to_weather_data_model.py +++ b/tests/sdk/python/rtdip_sdk/pipelines/transformers/spark/ecmwf/test_nc_extractbase_to_weather_data_model.py @@ -7,7 +7,6 @@ ECMWFExtractBaseToWeatherDataModel, ) - # Sample test data load_path = os.path.join(os.path.dirname(os.path.realpath(__file__)), "test_file") date_start = "2021-01-01 00:00:00" diff --git a/tests/sdk/python/rtdip_sdk/pipelines/transformers/spark/iso/test_miso_to_mdm.py b/tests/sdk/python/rtdip_sdk/pipelines/transformers/spark/iso/test_miso_to_mdm.py index 5e2217dc9..3818a816a 100644 --- a/tests/sdk/python/rtdip_sdk/pipelines/transformers/spark/iso/test_miso_to_mdm.py +++ b/tests/sdk/python/rtdip_sdk/pipelines/transformers/spark/iso/test_miso_to_mdm.py @@ -32,7 +32,6 @@ ) from pyspark.sql import SparkSession, DataFrame - parent_base_path: str = os.path.join( os.path.dirname(os.path.realpath(__file__)), "test_data" ) diff --git a/tests/sdk/python/rtdip_sdk/pipelines/utilities/aws/test_s3_copy_utility.py b/tests/sdk/python/rtdip_sdk/pipelines/utilities/aws/test_s3_copy_utility.py index c733542eb..a0d7afb5b 100644 --- a/tests/sdk/python/rtdip_sdk/pipelines/utilities/aws/test_s3_copy_utility.py +++ b/tests/sdk/python/rtdip_sdk/pipelines/utilities/aws/test_s3_copy_utility.py @@ -21,7 +21,6 @@ import boto3 from moto import mock_aws - sys.path.insert(0, ".") from src.sdk.python.rtdip_sdk.pipelines.utilities.aws.s3_copy_utility import ( diff --git a/tests/sdk/python/rtdip_sdk/queries/_test_utils/sdk_test_objects.py b/tests/sdk/python/rtdip_sdk/queries/_test_utils/sdk_test_objects.py index 6459c2e16..62b0bf625 100644 --- a/tests/sdk/python/rtdip_sdk/queries/_test_utils/sdk_test_objects.py +++ b/tests/sdk/python/rtdip_sdk/queries/_test_utils/sdk_test_objects.py @@ -50,10 +50,10 @@ PLOT_MOCKED_QUERY_UOM = "WITH raw_events AS (SELECT DISTINCT from_utc_timestamp(date_trunc('millisecond',`EventTime`), '+0000') AS `EventTime`, `TagName`, `Status`, `Value` FROM `mocked-buiness-unit`.`sensors`.`mocked-asset_mocked-data-security-level_events_mocked-data-type` WHERE `EventTime` BETWEEN to_timestamp('2011-01-01T00:00:00+00:00') AND to_timestamp('2011-01-02T23:59:59+00:00') AND `TagName` IN ('mocked-TAGNAME')), date_array AS (SELECT explode(sequence(from_utc_timestamp(to_timestamp('2011-01-01T00:00:00+00:00'), '+0000'), from_utc_timestamp(to_timestamp('2011-01-02T23:59:59+00:00'), '+0000'), INTERVAL '15 minute')) AS timestamp_array), window_buckets AS (SELECT timestamp_array AS window_start, timestampadd(minute, 15, timestamp_array) AS window_end FROM date_array), plot AS (SELECT /*+ RANGE_JOIN(d, 900) */ d.window_start, d.window_end, e.`TagName`, min(CASE WHEN `Status` = 'Bad' THEN null ELSE struct(e.`Value`, e.`EventTime`) END) OVER (PARTITION BY e.`TagName`, d.window_start ORDER BY e.`EventTime` ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) AS `min_Value`, max(CASE WHEN `Status` = 'Bad' THEN null ELSE struct(e.`Value`, e.`EventTime`) END) OVER (PARTITION BY e.`TagName`, d.window_start ORDER BY e.`EventTime` ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) AS `max_Value`, first(CASE WHEN `Status` = 'Bad' THEN null ELSE struct(e.`Value`, e.`EventTime`) END, True) OVER (PARTITION BY e.`TagName`, d.window_start ORDER BY e.`EventTime` ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) AS `first_Value`, last(CASE WHEN `Status` = 'Bad' THEN null ELSE struct(e.`Value`, e.`EventTime`) END, True) OVER (PARTITION BY e.`TagName`, d.window_start ORDER BY e.`EventTime` ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) AS `last_Value`, first(CASE WHEN `Status` = 'Bad' THEN struct(e.`Value`, e.`EventTime`) ELSE null END, True) OVER (PARTITION BY e.`TagName`, d.window_start ORDER BY e.`EventTime` ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) AS `excp_Value` FROM window_buckets d INNER JOIN raw_events e ON d.window_start <= e.`EventTime` AND d.window_end > e.`EventTime`), deduplicate AS (SELECT window_start AS `EventTime`, `TagName`, `min_Value` as `Min`, `max_Value` as `Max`, `first_Value` as `First`, `last_Value` as `Last`, `excp_Value` as `Exception` FROM plot GROUP BY window_start, `TagName`, `min_Value`, `max_Value`, `first_Value`, `last_Value`, `excp_Value`), project AS (SELECT distinct Values.EventTime, `TagName`, Values.Value FROM (SELECT * FROM deduplicate UNPIVOT (`Values` for `Aggregation` IN (`Min`, `Max`, `First`, `Last`, `Exception`))) ORDER BY `TagName`, `EventTime`), uom AS (SELECT project.*, metadata.`UoM` FROM project LEFT OUTER JOIN `mocked-buiness-unit`.`sensors`.`mocked-asset_mocked-data-security-level_metadata` AS metadata ON project.`TagName` = metadata.`TagName`) SELECT * FROM uom" # Interpolate -INTERPOLATE_MOCKED_QUERY = "WITH raw AS (SELECT from_utc_timestamp(date_trunc('millisecond',`EventTime`), '+0000') AS `EventTime`, window(from_utc_timestamp(date_trunc('millisecond',`EventTime`), '+0000'), '15 minute', '15 minute', '0 second') AS `window`, `TagName`, `Status`, `Value` FROM `mocked-buiness-unit`.`sensors`.`mocked-asset_mocked-data-security-level_events_mocked-data-type` WHERE `EventTime` BETWEEN to_timestamp('2011-01-01T00:00:00+00:00') AND timestampadd(minute, 15, to_timestamp('2011-01-02T23:59:59+00:00')) AND `TagName` IN ('mocked-TAGNAME')), intervals AS (SELECT DISTINCT explode(sequence(from_utc_timestamp(to_timestamp('2011-01-01T00:00:00+00:00'), '+0000'), from_utc_timestamp(to_timestamp('2011-01-02T23:59:59+00:00'), '+0000'), INTERVAL '15 minute')) AS `EventTime`, explode(array('mocked-TAGNAME')) AS `TagName`), fill_intervals as (SELECT intervals.`TagName`, intervals.`EventTime` as `EventTime`, raw. `EventTime` as `OriginalEventTime`, raw.`Value`, CASE WHEN raw.`Value` IS NULL THEN NULL ELSE struct(raw.`EventTime`, raw.`Value`) END AS `EventTime_Value` FROM intervals LEFT OUTER JOIN raw ON intervals.`EventTime` = raw.`window`.start AND intervals.`TagName` = raw.`TagName`), interpolate_calculate AS (SELECT `OriginalEventTime`, `EventTime`, `TagName`, CASE WHEN `Value` IS NOT NULL THEN NULL ELSE LAG(`EventTime_Value`) IGNORE NULLS OVER (PARTITION BY `TagName` ORDER BY `EventTime`) END AS PrevEventTimeValue, CASE WHEN `Value` IS NOT NULL THEN NULL ELSE LEAD(`EventTime_Value`) IGNORE NULLS OVER (PARTITION BY `TagName` ORDER BY `EventTime`) END AS NextEventTimeValue, CASE WHEN `OriginalEventTime` = `EventTime` THEN `Value` WHEN `PrevEventTimeValue` IS NOT NULL AND `NextEventTimeValue` IS NOT NULL THEN `PrevEventTimeValue`.`Value` + ((`NextEventTimeValue`.`Value` - `PrevEventTimeValue`.`Value`) * (unix_timestamp(`EventTime`) - unix_timestamp(`PrevEventTimeValue`.`EventTime`)) / (unix_timestamp(`NextEventTimeValue`.`EventTime`) - unix_timestamp(`PrevEventTimeValue`.`EventTime`))) WHEN `PrevEventTimeValue` IS NOT NULL THEN `PrevEventTimeValue`.`Value` ELSE NULL END as `Value` FROM fill_intervals ), interpolate AS (SELECT `EventTime`, `TagName`, `Value` FROM interpolate_calculate WHERE `OriginalEventTime` IS NULL OR `OriginalEventTime` = `EventTime` ORDER BY `TagName`, `EventTime`) SELECT * FROM interpolate" -INTERPOLATE_MOCKED_QUERY_CHECK_TAGS = "WITH raw AS (SELECT from_utc_timestamp(date_trunc('millisecond',`EventTime`), '+0000') AS `EventTime`, window(from_utc_timestamp(date_trunc('millisecond',`EventTime`), '+0000'), '15 minute', '15 minute', '0 second') AS `window`, `TagName`, `Status`, `Value` FROM `mocked-buiness-unit`.`sensors`.`mocked-asset_mocked-data-security-level_events_mocked-data-type` WHERE `EventTime` BETWEEN to_timestamp('2011-01-01T00:00:00+00:00') AND timestampadd(minute, 15, to_timestamp('2011-01-02T23:59:59+00:00')) AND UPPER(`TagName`) IN ('MOCKED-TAGNAME')), intervals AS (SELECT DISTINCT explode(sequence(from_utc_timestamp(to_timestamp('2011-01-01T00:00:00+00:00'), '+0000'), from_utc_timestamp(to_timestamp('2011-01-02T23:59:59+00:00'), '+0000'), INTERVAL '15 minute')) AS `EventTime`, explode(array('MOCKED-TAGNAME')) AS `TagName`), fill_intervals as (SELECT intervals.`TagName`, intervals.`EventTime` as `EventTime`, raw. `EventTime` as `OriginalEventTime`, raw.`Value`, CASE WHEN raw.`Value` IS NULL THEN NULL ELSE struct(raw.`EventTime`, raw.`Value`) END AS `EventTime_Value` FROM intervals LEFT OUTER JOIN raw ON intervals.`EventTime` = raw.`window`.start AND intervals.`TagName` = raw.`TagName`), interpolate_calculate AS (SELECT `OriginalEventTime`, `EventTime`, `TagName`, CASE WHEN `Value` IS NOT NULL THEN NULL ELSE LAG(`EventTime_Value`) IGNORE NULLS OVER (PARTITION BY `TagName` ORDER BY `EventTime`) END AS PrevEventTimeValue, CASE WHEN `Value` IS NOT NULL THEN NULL ELSE LEAD(`EventTime_Value`) IGNORE NULLS OVER (PARTITION BY `TagName` ORDER BY `EventTime`) END AS NextEventTimeValue, CASE WHEN `OriginalEventTime` = `EventTime` THEN `Value` WHEN `PrevEventTimeValue` IS NOT NULL AND `NextEventTimeValue` IS NOT NULL THEN `PrevEventTimeValue`.`Value` + ((`NextEventTimeValue`.`Value` - `PrevEventTimeValue`.`Value`) * (unix_timestamp(`EventTime`) - unix_timestamp(`PrevEventTimeValue`.`EventTime`)) / (unix_timestamp(`NextEventTimeValue`.`EventTime`) - unix_timestamp(`PrevEventTimeValue`.`EventTime`))) WHEN `PrevEventTimeValue` IS NOT NULL THEN `PrevEventTimeValue`.`Value` ELSE NULL END as `Value` FROM fill_intervals ), interpolate AS (SELECT `EventTime`, `TagName`, `Value` FROM interpolate_calculate WHERE `OriginalEventTime` IS NULL OR `OriginalEventTime` = `EventTime` ORDER BY `TagName`, `EventTime`) SELECT * FROM interpolate" -INTERPOLATE_MOCKED_QUERY_PIVOT = "WITH raw AS (SELECT from_utc_timestamp(date_trunc('millisecond',`EventTime`), '+0000') AS `EventTime`, window(from_utc_timestamp(date_trunc('millisecond',`EventTime`), '+0000'), '15 minute', '15 minute', '0 second') AS `window`, `TagName`, `Status`, `Value` FROM `mocked-buiness-unit`.`sensors`.`mocked-asset_mocked-data-security-level_events_mocked-data-type` WHERE `EventTime` BETWEEN to_timestamp('2011-01-01T00:00:00+00:00') AND timestampadd(minute, 15, to_timestamp('2011-01-02T23:59:59+00:00')) AND `TagName` IN ('mocked-TAGNAME')), intervals AS (SELECT DISTINCT explode(sequence(from_utc_timestamp(to_timestamp('2011-01-01T00:00:00+00:00'), '+0000'), from_utc_timestamp(to_timestamp('2011-01-02T23:59:59+00:00'), '+0000'), INTERVAL '15 minute')) AS `EventTime`, explode(array('mocked-TAGNAME')) AS `TagName`), fill_intervals as (SELECT intervals.`TagName`, intervals.`EventTime` as `EventTime`, raw. `EventTime` as `OriginalEventTime`, raw.`Value`, CASE WHEN raw.`Value` IS NULL THEN NULL ELSE struct(raw.`EventTime`, raw.`Value`) END AS `EventTime_Value` FROM intervals LEFT OUTER JOIN raw ON intervals.`EventTime` = raw.`window`.start AND intervals.`TagName` = raw.`TagName`), interpolate_calculate AS (SELECT `OriginalEventTime`, `EventTime`, `TagName`, CASE WHEN `Value` IS NOT NULL THEN NULL ELSE LAG(`EventTime_Value`) IGNORE NULLS OVER (PARTITION BY `TagName` ORDER BY `EventTime`) END AS PrevEventTimeValue, CASE WHEN `Value` IS NOT NULL THEN NULL ELSE LEAD(`EventTime_Value`) IGNORE NULLS OVER (PARTITION BY `TagName` ORDER BY `EventTime`) END AS NextEventTimeValue, CASE WHEN `OriginalEventTime` = `EventTime` THEN `Value` WHEN `PrevEventTimeValue` IS NOT NULL AND `NextEventTimeValue` IS NOT NULL THEN `PrevEventTimeValue`.`Value` + ((`NextEventTimeValue`.`Value` - `PrevEventTimeValue`.`Value`) * (unix_timestamp(`EventTime`) - unix_timestamp(`PrevEventTimeValue`.`EventTime`)) / (unix_timestamp(`NextEventTimeValue`.`EventTime`) - unix_timestamp(`PrevEventTimeValue`.`EventTime`))) WHEN `PrevEventTimeValue` IS NOT NULL THEN `PrevEventTimeValue`.`Value` ELSE NULL END as `Value` FROM fill_intervals ), interpolate AS (SELECT `EventTime`, `TagName`, `Value` FROM interpolate_calculate WHERE `OriginalEventTime` IS NULL OR `OriginalEventTime` = `EventTime` ), pivot AS (SELECT * FROM (SELECT `EventTime`, `Value`, `TagName` FROM interpolate) PIVOT (FIRST(`Value`) FOR `TagName` IN ('mocked-TAGNAME' AS `mocked-TAGNAME`)) ORDER BY `EventTime`) SELECT * FROM pivot" -INTERPOLATE_MOCKED_QUERY_UOM = "WITH raw AS (SELECT from_utc_timestamp(date_trunc('millisecond',`EventTime`), '+0000') AS `EventTime`, window(from_utc_timestamp(date_trunc('millisecond',`EventTime`), '+0000'), '15 minute', '15 minute', '0 second') AS `window`, `TagName`, `Status`, `Value` FROM `mocked-buiness-unit`.`sensors`.`mocked-asset_mocked-data-security-level_events_mocked-data-type` WHERE `EventTime` BETWEEN to_timestamp('2011-01-01T00:00:00+00:00') AND timestampadd(minute, 15, to_timestamp('2011-01-02T23:59:59+00:00')) AND `TagName` IN ('mocked-TAGNAME')), intervals AS (SELECT DISTINCT explode(sequence(from_utc_timestamp(to_timestamp('2011-01-01T00:00:00+00:00'), '+0000'), from_utc_timestamp(to_timestamp('2011-01-02T23:59:59+00:00'), '+0000'), INTERVAL '15 minute')) AS `EventTime`, explode(array('mocked-TAGNAME')) AS `TagName`), fill_intervals as (SELECT intervals.`TagName`, intervals.`EventTime` as `EventTime`, raw. `EventTime` as `OriginalEventTime`, raw.`Value`, CASE WHEN raw.`Value` IS NULL THEN NULL ELSE struct(raw.`EventTime`, raw.`Value`) END AS `EventTime_Value` FROM intervals LEFT OUTER JOIN raw ON intervals.`EventTime` = raw.`window`.start AND intervals.`TagName` = raw.`TagName`), interpolate_calculate AS (SELECT `OriginalEventTime`, `EventTime`, `TagName`, CASE WHEN `Value` IS NOT NULL THEN NULL ELSE LAG(`EventTime_Value`) IGNORE NULLS OVER (PARTITION BY `TagName` ORDER BY `EventTime`) END AS PrevEventTimeValue, CASE WHEN `Value` IS NOT NULL THEN NULL ELSE LEAD(`EventTime_Value`) IGNORE NULLS OVER (PARTITION BY `TagName` ORDER BY `EventTime`) END AS NextEventTimeValue, CASE WHEN `OriginalEventTime` = `EventTime` THEN `Value` WHEN `PrevEventTimeValue` IS NOT NULL AND `NextEventTimeValue` IS NOT NULL THEN `PrevEventTimeValue`.`Value` + ((`NextEventTimeValue`.`Value` - `PrevEventTimeValue`.`Value`) * (unix_timestamp(`EventTime`) - unix_timestamp(`PrevEventTimeValue`.`EventTime`)) / (unix_timestamp(`NextEventTimeValue`.`EventTime`) - unix_timestamp(`PrevEventTimeValue`.`EventTime`))) WHEN `PrevEventTimeValue` IS NOT NULL THEN `PrevEventTimeValue`.`Value` ELSE NULL END as `Value` FROM fill_intervals ), interpolate AS (SELECT `EventTime`, `TagName`, `Value` FROM interpolate_calculate WHERE `OriginalEventTime` IS NULL OR `OriginalEventTime` = `EventTime` ORDER BY `TagName`, `EventTime`), uom AS (SELECT interpolate.*, metadata.`UoM` FROM interpolate LEFT OUTER JOIN `mocked-buiness-unit`.`sensors`.`mocked-asset_mocked-data-security-level_metadata` AS metadata ON interpolate.`TagName` = metadata.`TagName`) SELECT * FROM uom" +INTERPOLATE_MOCKED_QUERY = "WITH raw AS (SELECT DISTINCT from_utc_timestamp(date_trunc('millisecond',`EventTime`), '+0000') AS `EventTime`, `TagName`, `Status`, `Value` FROM `mocked-buiness-unit`.`sensors`.`mocked-asset_mocked-data-security-level_events_mocked-data-type` WHERE `EventTime` BETWEEN timestampadd(minute, -5, to_timestamp('2011-01-01T00:00:00+00:00')) AND timestampadd(minute, 5, to_timestamp('2011-01-02T23:59:59+00:00')) AND `TagName` IN ('mocked-TAGNAME')), intervals AS (SELECT DISTINCT explode(sequence(from_utc_timestamp(to_timestamp('2011-01-01T00:00:00+00:00'), '+0000'), from_utc_timestamp(to_timestamp('2011-01-02T23:59:59+00:00'), '+0000'), INTERVAL '15 minute')) AS `EventTime`, explode(array('mocked-TAGNAME')) AS `TagName`), fill_intervals AS (SELECT `EventTime`, `TagName`, `Value`, 0 AS `SortKey` FROM raw UNION ALL SELECT `EventTime`, `TagName`, CAST(NULL AS DOUBLE), 1 AS `SortKey` FROM intervals), interpolate_calculate AS (SELECT `EventTime`, `TagName`, `SortKey`, `Value`, LAG(CASE WHEN `SortKey` = 0 THEN struct(`EventTime` AS `EventTime`, `Value` AS `Value`) END) IGNORE NULLS OVER (PARTITION BY `TagName` ORDER BY `EventTime`, `SortKey`) AS `Prev`, LEAD(CASE WHEN `SortKey` = 0 THEN struct(`EventTime` AS `EventTime`, `Value` AS `Value`) END) IGNORE NULLS OVER (PARTITION BY `TagName` ORDER BY `EventTime`, `SortKey`) AS `Next` FROM fill_intervals), interpolate AS (SELECT `EventTime`, `TagName`, CASE WHEN `Prev` IS NOT NULL AND `Next` IS NOT NULL THEN `Prev`.`Value` + ((`Next`.`Value` - `Prev`.`Value`) * (unix_millis(`EventTime`) - unix_millis(`Prev`.`EventTime`)) / (unix_millis(`Next`.`EventTime`) - unix_millis(`Prev`.`EventTime`))) WHEN `Prev` IS NOT NULL THEN `Prev`.`Value` ELSE NULL END AS `Value` FROM interpolate_calculate WHERE `SortKey` = 1) SELECT `EventTime`, `TagName`, `Value` FROM interpolate ORDER BY `TagName`, `EventTime`" +INTERPOLATE_MOCKED_QUERY_CHECK_TAGS = "WITH raw AS (SELECT DISTINCT from_utc_timestamp(date_trunc('millisecond',`EventTime`), '+0000') AS `EventTime`, `TagName`, `Status`, `Value` FROM `mocked-buiness-unit`.`sensors`.`mocked-asset_mocked-data-security-level_events_mocked-data-type` WHERE `EventTime` BETWEEN timestampadd(minute, -5, to_timestamp('2011-01-01T00:00:00+00:00')) AND timestampadd(minute, 5, to_timestamp('2011-01-02T23:59:59+00:00')) AND UPPER(`TagName`) IN ('MOCKED-TAGNAME')), intervals AS (SELECT DISTINCT explode(sequence(from_utc_timestamp(to_timestamp('2011-01-01T00:00:00+00:00'), '+0000'), from_utc_timestamp(to_timestamp('2011-01-02T23:59:59+00:00'), '+0000'), INTERVAL '15 minute')) AS `EventTime`, explode(array('MOCKED-TAGNAME')) AS `TagName`), fill_intervals AS (SELECT `EventTime`, `TagName`, `Value`, 0 AS `SortKey` FROM raw UNION ALL SELECT `EventTime`, `TagName`, CAST(NULL AS DOUBLE), 1 AS `SortKey` FROM intervals), interpolate_calculate AS (SELECT `EventTime`, `TagName`, `SortKey`, `Value`, LAG(CASE WHEN `SortKey` = 0 THEN struct(`EventTime` AS `EventTime`, `Value` AS `Value`) END) IGNORE NULLS OVER (PARTITION BY `TagName` ORDER BY `EventTime`, `SortKey`) AS `Prev`, LEAD(CASE WHEN `SortKey` = 0 THEN struct(`EventTime` AS `EventTime`, `Value` AS `Value`) END) IGNORE NULLS OVER (PARTITION BY `TagName` ORDER BY `EventTime`, `SortKey`) AS `Next` FROM fill_intervals), interpolate AS (SELECT `EventTime`, `TagName`, CASE WHEN `Prev` IS NOT NULL AND `Next` IS NOT NULL THEN `Prev`.`Value` + ((`Next`.`Value` - `Prev`.`Value`) * (unix_millis(`EventTime`) - unix_millis(`Prev`.`EventTime`)) / (unix_millis(`Next`.`EventTime`) - unix_millis(`Prev`.`EventTime`))) WHEN `Prev` IS NOT NULL THEN `Prev`.`Value` ELSE NULL END AS `Value` FROM interpolate_calculate WHERE `SortKey` = 1) SELECT `EventTime`, `TagName`, `Value` FROM interpolate ORDER BY `TagName`, `EventTime`" +INTERPOLATE_MOCKED_QUERY_PIVOT = "WITH raw AS (SELECT DISTINCT from_utc_timestamp(date_trunc('millisecond',`EventTime`), '+0000') AS `EventTime`, `TagName`, `Status`, `Value` FROM `mocked-buiness-unit`.`sensors`.`mocked-asset_mocked-data-security-level_events_mocked-data-type` WHERE `EventTime` BETWEEN timestampadd(minute, -5, to_timestamp('2011-01-01T00:00:00+00:00')) AND timestampadd(minute, 5, to_timestamp('2011-01-02T23:59:59+00:00')) AND `TagName` IN ('mocked-TAGNAME')), intervals AS (SELECT DISTINCT explode(sequence(from_utc_timestamp(to_timestamp('2011-01-01T00:00:00+00:00'), '+0000'), from_utc_timestamp(to_timestamp('2011-01-02T23:59:59+00:00'), '+0000'), INTERVAL '15 minute')) AS `EventTime`, explode(array('mocked-TAGNAME')) AS `TagName`), fill_intervals AS (SELECT `EventTime`, `TagName`, `Value`, 0 AS `SortKey` FROM raw UNION ALL SELECT `EventTime`, `TagName`, CAST(NULL AS DOUBLE), 1 AS `SortKey` FROM intervals), interpolate_calculate AS (SELECT `EventTime`, `TagName`, `SortKey`, `Value`, LAG(CASE WHEN `SortKey` = 0 THEN struct(`EventTime` AS `EventTime`, `Value` AS `Value`) END) IGNORE NULLS OVER (PARTITION BY `TagName` ORDER BY `EventTime`, `SortKey`) AS `Prev`, LEAD(CASE WHEN `SortKey` = 0 THEN struct(`EventTime` AS `EventTime`, `Value` AS `Value`) END) IGNORE NULLS OVER (PARTITION BY `TagName` ORDER BY `EventTime`, `SortKey`) AS `Next` FROM fill_intervals), interpolate AS (SELECT `EventTime`, `TagName`, CASE WHEN `Prev` IS NOT NULL AND `Next` IS NOT NULL THEN `Prev`.`Value` + ((`Next`.`Value` - `Prev`.`Value`) * (unix_millis(`EventTime`) - unix_millis(`Prev`.`EventTime`)) / (unix_millis(`Next`.`EventTime`) - unix_millis(`Prev`.`EventTime`))) WHEN `Prev` IS NOT NULL THEN `Prev`.`Value` ELSE NULL END AS `Value` FROM interpolate_calculate WHERE `SortKey` = 1), pivot AS (SELECT * FROM (SELECT `EventTime`, `Value`, `TagName` FROM interpolate) PIVOT (FIRST(`Value`) FOR `TagName` IN ('mocked-TAGNAME' AS `mocked-TAGNAME`)) ORDER BY `EventTime`) SELECT * FROM pivot" +INTERPOLATE_MOCKED_QUERY_UOM = "WITH raw AS (SELECT DISTINCT from_utc_timestamp(date_trunc('millisecond',`EventTime`), '+0000') AS `EventTime`, `TagName`, `Status`, `Value` FROM `mocked-buiness-unit`.`sensors`.`mocked-asset_mocked-data-security-level_events_mocked-data-type` WHERE `EventTime` BETWEEN timestampadd(minute, -5, to_timestamp('2011-01-01T00:00:00+00:00')) AND timestampadd(minute, 5, to_timestamp('2011-01-02T23:59:59+00:00')) AND `TagName` IN ('mocked-TAGNAME')), intervals AS (SELECT DISTINCT explode(sequence(from_utc_timestamp(to_timestamp('2011-01-01T00:00:00+00:00'), '+0000'), from_utc_timestamp(to_timestamp('2011-01-02T23:59:59+00:00'), '+0000'), INTERVAL '15 minute')) AS `EventTime`, explode(array('mocked-TAGNAME')) AS `TagName`), fill_intervals AS (SELECT `EventTime`, `TagName`, `Value`, 0 AS `SortKey` FROM raw UNION ALL SELECT `EventTime`, `TagName`, CAST(NULL AS DOUBLE), 1 AS `SortKey` FROM intervals), interpolate_calculate AS (SELECT `EventTime`, `TagName`, `SortKey`, `Value`, LAG(CASE WHEN `SortKey` = 0 THEN struct(`EventTime` AS `EventTime`, `Value` AS `Value`) END) IGNORE NULLS OVER (PARTITION BY `TagName` ORDER BY `EventTime`, `SortKey`) AS `Prev`, LEAD(CASE WHEN `SortKey` = 0 THEN struct(`EventTime` AS `EventTime`, `Value` AS `Value`) END) IGNORE NULLS OVER (PARTITION BY `TagName` ORDER BY `EventTime`, `SortKey`) AS `Next` FROM fill_intervals), interpolate AS (SELECT `EventTime`, `TagName`, CASE WHEN `Prev` IS NOT NULL AND `Next` IS NOT NULL THEN `Prev`.`Value` + ((`Next`.`Value` - `Prev`.`Value`) * (unix_millis(`EventTime`) - unix_millis(`Prev`.`EventTime`)) / (unix_millis(`Next`.`EventTime`) - unix_millis(`Prev`.`EventTime`))) WHEN `Prev` IS NOT NULL THEN `Prev`.`Value` ELSE NULL END AS `Value` FROM interpolate_calculate WHERE `SortKey` = 1), uom AS (SELECT i.`EventTime`, i.`TagName`, i.`Value`, m.`UoM` FROM interpolate i LEFT OUTER JOIN `mocked-buiness-unit`.`sensors`.`mocked-asset_mocked-data-security-level_metadata` AS m ON i.`TagName` = m.`TagName`) SELECT * FROM uom ORDER BY `TagName`, `EventTime`" # Time Weighted Average TWA_MOCKED_QUERY = 'WITH raw_events AS (SELECT DISTINCT `TagName`, from_utc_timestamp(date_trunc("millisecond",`EventTime`), "+0000") AS `EventTime`, `Status`, `Value` FROM `mocked-buiness-unit`.`sensors`.`mocked-asset_mocked-data-security-level_events_mocked-data-type` WHERE to_date(`EventTime`) BETWEEN date_sub(to_date(to_timestamp("2011-01-01T00:00:00+00:00")), 1) AND date_add(to_date(to_timestamp("2011-01-02T23:59:59+00:00")), 1) AND `TagName` IN (\'mocked-TAGNAME\') ) ,date_array AS (SELECT explode(sequence(from_utc_timestamp(to_timestamp("2011-01-01T00:00:00+00:00"), "+0000"), from_utc_timestamp(to_timestamp("2011-01-02T23:59:59+00:00"), "+0000"), INTERVAL \'15 minute\')) AS `EventTime`, explode(array(\'mocked-TAGNAME\')) AS `TagName`) ,boundary_events AS (SELECT coalesce(a.`TagName`, b.`TagName`) AS `TagName`, coalesce(a.`EventTime`, b.`EventTime`) AS `EventTime`, b.`Status`, b.`Value` FROM date_array a FULL OUTER JOIN raw_events b ON a.`EventTime` = b.`EventTime` AND a.`TagName` = b.`TagName`) ,window_buckets AS (SELECT `EventTime` AS window_start, LEAD(`EventTime`) OVER (ORDER BY `EventTime`) AS window_end FROM (SELECT distinct `EventTime` FROM date_array) ) ,window_events AS (SELECT /*+ RANGE_JOIN(b, 900 ) */ b.`TagName`, b.`EventTime`, a.window_start AS `WindowEventTime`, b.`Status`, b.`Value` FROM boundary_events b LEFT OUTER JOIN window_buckets a ON a.window_start <= b.`EventTime` AND a.window_end > b.`EventTime`) ,fill_status AS (SELECT *, last_value(`Status`, true) OVER (PARTITION BY `TagName` ORDER BY `EventTime` ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS `Fill_Status`, CASE WHEN `Fill_Status` <> "Bad" THEN `Value` ELSE null END AS `Good_Value` FROM window_events) ,fill_value AS (SELECT *, last_value(`Good_Value`, true) OVER (PARTITION BY `TagName` ORDER BY `EventTime` ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS `Fill_Value` FROM fill_status) ,fill_step AS (SELECT *, false AS Step FROM fill_value) ,interpolate AS (SELECT *, CASE WHEN `Step` = false AND `Status` IS NULL AND `Value` IS NULL THEN lag(`EventTime`) OVER ( PARTITION BY `TagName` ORDER BY `EventTime` ) ELSE NULL END AS `Previous_EventTime`, CASE WHEN `Step` = false AND `Status` IS NULL AND `Value` IS NULL THEN lag(`Fill_Value`) OVER ( PARTITION BY `TagName` ORDER BY `EventTime` ) ELSE NULL END AS `Previous_Fill_Value`, lead(`EventTime`) OVER ( PARTITION BY `TagName` ORDER BY `EventTime` ) AS `Next_EventTime`, CASE WHEN `Step` = false AND `Status` IS NULL AND `Value` IS NULL THEN lead(`Fill_Value`) OVER ( PARTITION BY `TagName` ORDER BY `EventTime` ) ELSE NULL END AS `Next_Fill_Value`, CASE WHEN `Step` = false AND `Status` IS NULL AND `Value` IS NULL THEN `Previous_Fill_Value` + ( (`Next_Fill_Value` - `Previous_Fill_Value`) * ( ( unix_timestamp(`EventTime`) - unix_timestamp(`Previous_EventTime`) ) / ( unix_timestamp(`Next_EventTime`) - unix_timestamp(`Previous_EventTime`) ) ) ) ELSE NULL END AS `Interpolated_Value`, coalesce(`Interpolated_Value`, `Fill_Value`) as `Event_Value` FROM fill_step ),twa_calculations AS (SELECT `TagName`, `EventTime`, `WindowEventTime`, `Step`, `Status`, `Value`, `Previous_EventTime`, `Previous_Fill_Value`, `Next_EventTime`, `Next_Fill_Value`, `Interpolated_Value`, `Fill_Status`, `Fill_Value`, `Event_Value`, lead(`Fill_Status`) OVER (PARTITION BY `TagName` ORDER BY `EventTime`) AS `Next_Status` , CASE WHEN `Next_Status` <> "Bad" OR (`Fill_Status` <> "Bad" AND `Next_Status` = "Bad") THEN lead(`Event_Value`) OVER (PARTITION BY `TagName` ORDER BY `EventTime`) ELSE `Value` END AS `Next_Value_For_Status` , CASE WHEN `Fill_Status` <> "Bad" THEN `Next_Value_For_Status` ELSE 0 END AS `Next_Value` , CASE WHEN `Fill_Status` <> "Bad" AND `Next_Status` <> "Bad" THEN ((cast(`Next_EventTime` AS double) - cast(`EventTime` AS double)) / 60) WHEN `Fill_Status` <> "Bad" AND `Next_Status` = "Bad" THEN ((cast(`Next_EventTime` AS integer) - cast(`EventTime` AS double)) / 60) ELSE 0 END AS good_minutes , CASE WHEN Step == false THEN ((`Event_Value` + `Next_Value`) * 0.5) * good_minutes ELSE (`Event_Value` * good_minutes) END AS twa_value FROM interpolate) ,twa AS (SELECT `TagName`, `WindowEventTime` AS `EventTime`, sum(twa_value) / sum(good_minutes) AS `Value` from twa_calculations GROUP BY `TagName`, `WindowEventTime`) ,project AS (SELECT * FROM twa WHERE `EventTime` BETWEEN to_timestamp("2011-01-01T00:00:00") AND to_timestamp("2011-01-02T23:59:59")) SELECT * FROM project ORDER BY `TagName`, `EventTime` '