Merge pull request #517 from rtdip/develop

v0.8.0

docs/sdk/code-reference/query/latest.md

@@ -0,0 +1,2 @@
+# Raw Function
+::: src.sdk.python.rtdip_sdk.queries.time_series.latest

docs/sdk/queries/functions.md

@@ -8,6 +8,11 @@ The RTDIP SDK enables users to perform complex queries, including aggregation on
 
 [Raw](../code-reference/query/raw.md) facilitates performing raw extracts of time series data, typically filtered by a Tag Name or Device Name and an event time.
 
+### Latest
+
+[Latest](../code-reference/query/latest.md) queries provides the latest event values. The RTDIP SDK requires the following parameters to retrieve the latest event values:
+- TagNames - A list of tag names
+
 ### Resample
 
 [Resample](../code-reference/query/resample.md) enables changing the frequency of time series observations. This is achieved by providing the following parameters:
@@ -69,6 +74,7 @@ The RTDIP SDK enables users to perform complex queries, including aggregation on
 ### Metadata
 [Metadata](../code-reference/query/metadata.md) queries provide contextual information for time series measurements and include information such as names, descriptions and units of measure.
 
+
 !!! note "Note"
     </b>RTDIP are continuously adding more to this list so check back regularly.<br />
 

environment.yml

@@ -54,7 +54,7 @@ dependencies:
     - pygments==2.16.1
     - pymdown-extensions==10.1.0
     - databricks-sql-connector==2.9.3
-    - databricks-sdk==0.6.0
+    - databricks-sdk==0.9.0
     - semver==3.0.0
     - xlrd==2.0.1
     - pygithub==1.59.0

mkdocs.yml

@@ -239,10 +239,11 @@ nav:
           - Queries:
               - Query Builder: sdk/code-reference/query/query_builder.md
               - Functions:
+                  - Raw: sdk/code-reference/query/raw.md
+                  - Latest: sdk/code-reference/query/latest.md                                
                   - Resample: sdk/code-reference/query/resample.md
                   - Interpolate: sdk/code-reference/query/interpolate.md
                   - Interpolation at Time: sdk/code-reference/query/interpolation-at-time.md
-                  - Raw: sdk/code-reference/query/raw.md
                   - Time Weighted Average: sdk/code-reference/query/time-weighted-average.md
                   - Circular Average: sdk/code-reference/query/circular-average.md
                   - Circular Standard Deviation: sdk/code-reference/query/circular-standard-deviation.md

setup.py

@@ -50,7 +50,7 @@
 
 PIPELINE_PACKAGES = [
     "dependency-injector==4.41.0",
-    "databricks-sdk==0.2.1",
+    "databricks-sdk==0.9.0",
     "pydantic==1.10.12",
     "azure-storage-file-datalake==12.12.0",
     "boto3==1.28.2",

src/api/v1/__init__.py

@@ -18,6 +18,7 @@
 from src.api.v1 import (
     metadata,
     raw,
+    latest,
     resample,
     interpolate,
     interpolation_at_time,

src/api/v1/latest.py

@@ -0,0 +1,126 @@
+# Copyright 2022 RTDIP
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+import logging
+import numpy as np
+from pandas.io.json import build_table_schema
+from fastapi import Query, HTTPException, Depends, Body
+import nest_asyncio
+from src.sdk.python.rtdip_sdk.queries import latest
+from src.api.v1.models import (
+    BaseQueryParams,
+    BaseHeaders,
+    MetadataQueryParams,
+    TagsBodyParams,
+    LatestResponse,
+    LimitOffsetQueryParams,
+    HTTPError,
+)
+from src.api.auth.azuread import oauth2_scheme
+from src.api.FastAPIApp import api_v1_router
+from src.api.v1 import common
+
+nest_asyncio.apply()
+
+
+def latest_retrieval_get(
+    query_parameters, metadata_query_parameters, limit_offset_parameters, base_headers
+):
+    try:
+        (connection, parameters) = common.common_api_setup_tasks(
+            query_parameters,
+            metadata_query_parameters=metadata_query_parameters,
+            limit_offset_query_parameters=limit_offset_parameters,
+            base_headers=base_headers,
+        )
+
+        data = latest.get(connection, parameters)
+        return LatestResponse(
+            schema=build_table_schema(data, index=False, primary_key=False),
+            data=data.replace({np.nan: None}).to_dict(orient="records"),
+        )
+    except Exception as e:
+        logging.error(str(e))
+        raise HTTPException(status_code=400, detail=str(e))
+
+
+get_description = """
+## Latest 
+
+Retrieval of latest event values for a given tag name or list of tag names.
+"""
+
+
+@api_v1_router.get(
+    path="/events/latest",
+    name="Latest GET",
+    description=get_description,
+    tags=["Events"],
+    dependencies=[Depends(oauth2_scheme)],
+    responses={200: {"model": LatestResponse}, 400: {"model": HTTPError}},
+    openapi_extra={
+        "externalDocs": {
+            "description": "RTDIP Latest Query Documentation",
+            "url": "https://www.rtdip.io/sdk/code-reference/query/latest/",
+        }
+    },
+)
+async def latest_get(
+    query_parameters: BaseQueryParams = Depends(),
+    metadata_query_parameters: MetadataQueryParams = Depends(),
+    limit_offset_parameters: LimitOffsetQueryParams = Depends(),
+    base_headers: BaseHeaders = Depends(),
+):
+    return latest_retrieval_get(
+        query_parameters,
+        metadata_query_parameters,
+        limit_offset_parameters,
+        base_headers,
+    )
+
+
+post_description = """
+## Metadata 
+
+Retrieval of latest event values for a given tag name or list of tag names via a POST method to enable providing a list of tag names that can exceed url length restrictions via GET Query Parameters.
+"""
+
+
+@api_v1_router.post(
+    path="/events/latest",
+    name="Latest POST",
+    description=post_description,
+    tags=["Events"],
+    dependencies=[Depends(oauth2_scheme)],
+    responses={200: {"model": LatestResponse}, 400: {"model": HTTPError}},
+    openapi_extra={
+        "externalDocs": {
+            "description": "RTDIP Latest Query Documentation",
+            "url": "https://www.rtdip.io/sdk/code-reference/query/latest/",
+        }
+    },
+)
+async def latest_post(
+    query_parameters: BaseQueryParams = Depends(),
+    metadata_query_parameters: TagsBodyParams = Body(default=...),
+    limit_offset_parameters: LimitOffsetQueryParams = Depends(),
+    base_headers: BaseHeaders = Depends(),
+):
+    return latest_retrieval_get(
+        query_parameters,
+        metadata_query_parameters,
+        limit_offset_parameters,
+        base_headers,
+    )

src/api/v1/models.py

@@ -53,6 +53,17 @@ class Config:
         extra = Extra.allow
 
 
+class LatestRow(BaseModel):
+    TagName: str
+    EventTime: datetime
+    Status: str
+    Value: str
+    ValueType: str
+    GoodEventTime: datetime
+    GoodValue: str
+    GoodValueType: str
+
+
 class RawRow(BaseModel):
     EventTime: datetime
     TagName: str
@@ -73,6 +84,11 @@ class MetadataResponse(BaseModel):
     data: List[MetadataRow]
 
 
+class LatestResponse(BaseModel):
+    field_schema: FieldSchema = Field(None, alias="schema")
+    data: List[LatestRow]
+
+
 class RawResponse(BaseModel):
     field_schema: FieldSchema = Field(None, alias="schema")
     data: List[RawRow]

src/sdk/python/rtdip_sdk/functions/latest.py

@@ -0,0 +1,21 @@
+# Copyright 2022 RTDIP
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import logging
+
+logging.warning(
+    "Module rtdip_sdk.functions is deprecated and will be removed in v1.0.0. Please import rtdip_sdk.queries instead."
+)
+
+from ..queries.metadata import *  # NOSONAR

src/sdk/python/rtdip_sdk/queries/__init__.py

@@ -14,6 +14,7 @@
 
 from .time_series import (
     raw,
+    latest,
     resample,
     interpolate,
     interpolation_at_time,

src/sdk/python/rtdip_sdk/queries/query_builder.py

@@ -22,6 +22,7 @@
     time_weighted_average,
     circular_average,
     circular_standard_deviation,
+    latest,
 )
 from . import metadata
 from pandas import DataFrame
@@ -352,6 +353,33 @@ def metadata(
 
         return metadata.get(self.connection, metadata_parameters)
 
+    def latest(
+        self,
+        tagname_filter: [str],
+        limit: int = None,
+        offset: int = None,
+    ) -> DataFrame:
+        """
+        A query to retrieve latest event_values
+
+        Args:
+            tagname_filter (list str): List of tagnames to filter on the source
+            limit (optional int): The number of rows to be returned
+            offset (optional int): The number of rows to skip before returning rows
+
+        Returns:
+            DataFrame: A dataframe of events latest_values
+        """
+        latest_parameters = {
+            "source": self.data_source,
+            "tag_names": tagname_filter,
+            "tagname_column": self.tagname_column,
+            "limit": limit,
+            "offset": offset,
+        }
+
+        return latest.get(self.connection, latest_parameters)
+
     def circular_average(
         self,
         tagname_filter: [str],

src/sdk/python/rtdip_sdk/queries/time_series/_query_builder.py

@@ -379,6 +379,42 @@ def _metadata_query(parameters_dict: dict) -> str:
     return sql_template.render(metadata_parameters)
 
 
+def _latest_query(parameters_dict: dict) -> str:
+    latest_query = (
+        "SELECT * FROM "
+        "{% if source is defined and source is not none %}"
+        "`{{ source|lower }}` "
+        "{% else %}"
+        "`{{ business_unit|lower }}`.`sensors`.`{{ asset|lower }}_{{ data_security_level|lower }}_events_latest` "
+        "{% endif %}"
+        "{% if tag_names is defined and tag_names|length > 0 %} "
+        "WHERE `{{ tagname_column }}` IN ('{{ tag_names | join('\\', \\'') }}') "
+        "{% endif %}"
+        "ORDER BY `{{ tagname_column }}` "
+        "{% if limit is defined and limit is not none %}"
+        "LIMIT {{ limit }} "
+        "{% endif %}"
+        "{% if offset is defined and offset is not none %}"
+        "OFFSET {{ offset }} "
+        "{% endif %}"
+    )
+
+    latest_parameters = {
+        "source": parameters_dict.get("source", None),
+        "business_unit": parameters_dict.get("business_unit"),
+        "region": parameters_dict.get("region"),
+        "asset": parameters_dict.get("asset"),
+        "data_security_level": parameters_dict.get("data_security_level"),
+        "tag_names": list(dict.fromkeys(parameters_dict["tag_names"])),
+        "limit": parameters_dict.get("limit", None),
+        "offset": parameters_dict.get("offset", None),
+        "tagname_column": parameters_dict.get("tagname_column", "TagName"),
+    }
+
+    sql_template = Template(latest_query)
+    return sql_template.render(latest_parameters)
+
+
 def _time_weighted_average_query(parameters_dict: dict) -> str:
     parameters_dict["start_datetime"] = datetime.strptime(
         parameters_dict["start_date"], TIMESTAMP_FORMAT
@@ -577,6 +613,9 @@ def _query_builder(parameters_dict: dict, query_type: str) -> str:
     if query_type == "metadata":
         return _metadata_query(parameters_dict)
 
+    if query_type == "latest":
+        return _latest_query(parameters_dict)
+
     parameters_dict = _parse_dates(parameters_dict)
 
     if query_type == "interpolation_at_time":