oee changes for calculating batch oee

conf/application.conf

@@ -51,4 +51,11 @@ topic=$KAFKA_TOPIC
 enable_sites_partition=$ENABLE_KAFKA_PARTITION
 split_key=$KAFKA_PARTITION_KEY
 round_robin_enable=$ROUND_ROBIN_PARTITION
-partition_db=13
\ No newline at end of file
+partition_db=13
+
+[KAIROS_DB]
+uri = $KAIROS_URI
+default_full_tag_level = c3
+default_tag_level = c5
+default_site_tag_level = c1
+kairos_cache_time=$KAIROS_CACHE_TIME
\ No newline at end of file

requirements.txt

@@ -25,4 +25,6 @@ paho-mqtt~=1.5.0
 pendulum~=2.1.2
 requests~=2.27.1
 aiohttp~=3.8.1
-numpy==1.22.3
\ No newline at end of file
+numpy==1.22.3
+python-dateutil~=2.8.2
+configparser~=3.5.3
\ No newline at end of file

scripts/config/__init__.py

@@ -52,6 +52,11 @@ class DBConf:
     if not MONGO_URI:
         print("Error, environment variable MONGO_URI not set")
         sys.exit(1)
+    KAIROS_URL = config.get("KAIROS_DB", "uri")
+    KAIROS_DEFAULT_FULL_TAG = config.get("KAIROS_DB", "default_full_tag_level")
+    KAIROS_DEFAULT_SITE_TAG = config.get("KAIROS_DB", "default_site_tag_level")
+    KAIROS_DEFAULT_TAG = config.get("KAIROS_DB", "default_tag_level")
+    KAIROS_CACHE_TIME = config.getint("KAIROS_DB", "kairos_cache_time")
 
 
 class PathToStorage(object):

scripts/constants/__init__.py

+import os
 from collections import namedtuple
 
 
@@ -28,7 +29,9 @@ class Endpoints:
 
 
 class StatusCodes:
-    SUCCESS = [200, 201, 204]
+    SUCCESS = "success"
+    FAILED = "failed"
+    SUCCESS_CODES = [200, 201, 204]
 
 
 class ResponseCodes:
@@ -67,3 +70,66 @@ class DBConstants:
     db_ilens_assistant = "ilens_assistant"
 
     collection_constants = "constants"
+    collection_tag_hierarchy = "tag_hierarchy"
+
+
+class EndpointConstants:
+    hierarchy_downtime = "/downtime_log/get/overall_downtime"
+
+
+class TagCategoryConstants:
+    TOTAL_UNITS_CATEGORY = os.environ.get("TOTAL_UNITS_CATEGORY", default="OEE - Total Produced Units")
+    REJECT_UNITS_CATEGORY = os.environ.get("TOTAL_UNITS_CATEGORY", default="OEE - Reject Units")
+
+
+class CommonConstants:
+    ui = 'ui_datetime_format'
+    utc = 'utc_datetime_format'
+    nsc = 'no_special_chars_datetime_format'
+    USER_META_TIME_FORMAT = "%Y-%m-%d %H:%M:%S"
+    umtf = 'user_meta_time_format'
+    __temporary_format__ = '%Y-%m-%dT%H:%M:%S+0530'
+    __iso_format__ = '%Y-%m-%dT%H:%M:%S%z'
+    __form_iso_format__ = '%Y-%m-%dT%H:%M:%S.%f%z'
+    __utc_datetime_format__ = '%Y-%m-%dT%H:%M:%S'
+    __ui_datetime_format__ = '%Y-%m-%d %H:%M:%S'
+    __no_special_chars_datetime_format__ = '%Y%m%d%H%M%S'
+    __user_meta_time_format__ = "%d-%m-%Y %H:%M:%S"
+    __user_meta_time_format_ws__ = "%d-%m-%Y %H:%M"
+    __ui_display_datetime_format__ = "%d %b %Y %H:%M"
+    __ui_display_date_format__ = "%d %b %Y"
+    __time_format__ = "%H:%M:%S"
+
+
+class CommonStatusCode:
+    SUCCESS_CODES = (
+        200,
+        201,
+        204,
+    )
+
+
+class TagHierarchyKeys:
+    KEY_ID = "id"
+    KEY_SITE_ID = "site_id"
+    KEY_SITE_NAME = "site_name"
+    KEY_DEPT_ID = "dept_id"
+    KEY_DEPT_NAME = "dept_name"
+    KEY_LINE_ID = "line_id"
+    KEY_LINE_NAME = "line_name"
+    KEY_EQUIPMENT_ID = "equipment_id"
+    KEY_EQUIPMENT_NAME = "equipment_name"
+    KEY_PARAMETER_ID = "parameter_id"
+    KEY_PROJECT_ID = "project_id"
+    KEY_LOWER_LIMIT = "lower_limit"
+    KEY_UPPER_LIMIT = "upper_limit"
+    KEY_DETAILS_LEVEL = "kpi_details"
+    KEY_KPI_LEVEL = "kpi_level_list"
+    KEY_KPI_PARENT = "kpi_parent_list"
+    KEY_TAG_CATEGORY_ID = "tag_category_id"
+    KEY_TAG_CODE = "tag_code"
+    KEY_TAG_REGISTER = "tag_register"
+    KEY_STARRED = "starred"
+    KEY_ASSET_MODEL_ID = "asset_model_id"
+    KEY_ASSET_VERSION = "asset_version"
+    KEY_additional_fields = "additional_fields"

scripts/core/engine/oee_calculator.py

 from datetime import datetime
 
-from scripts.constants import UOM
+from scripts.constants import UOM, TagCategoryConstants
 from scripts.errors import ErrorCodes
 from scripts.logging import logger
 from scripts.schemas.batch_oee import OEEDataSaveRequest, BatchOEEData
@@ -87,6 +87,22 @@ class OEELossesCalculator:
         return 100 - availability_loss - quality_loss - oee_percentage
 
 
+class OEETagFinder:
+    @staticmethod
+    def get_total_units_tag_id(input_data: dict, category_name=TagCategoryConstants.TOTAL_UNITS_CATEGORY):
+        if not input_data.get(category_name):
+            logger.error(ErrorCodes.ERR006)
+            raise ValueError(ErrorCodes.ERR006)
+        return input_data.get(category_name)
+
+    @staticmethod
+    def get_reject_units_tag_id(input_data: dict, category_name=TagCategoryConstants.REJECT_UNITS_CATEGORY):
+        if not input_data.get(category_name):
+            logger.error(ErrorCodes.ERR007)
+            raise ValueError(ErrorCodes.ERR007)
+        return input_data.get(category_name)
+
+
 class OEEEngine:
     def __init__(self):
         self.oee_calc = OEECalculator()
@@ -103,28 +119,35 @@ class OEEEngine:
             # Start and End time should be in milliseconds since epoch.
             if request_data.uom == UOM.minutes:
                 divisor = UOM.time_divs.minutes
+                cal_type = "minutes"
             elif request_data.uom == UOM.seconds:
                 divisor = UOM.time_divs.seconds
+                cal_type = "seconds"
             elif request_data.uom == UOM.hours:
                 divisor = UOM.time_divs.hours
+                cal_type = "hours"
             elif request_data.uom == UOM.millis:
                 divisor = UOM.time_divs.millis
+                cal_type = "microseconds"
             else:
                 divisor = UOM.time_divs.minutes
+                cal_type = "minutes"
 
-            planned_production_time = self.common_util.get_duration(tz=request_data.tz, meta=request_data.dict(),
-                                                                    difference=True) / divisor
-
-            operating_time = planned_production_time - request_data.downtime
+            duration = self.common_util.get_duration(tz=request_data.tz, meta=request_data.dict(),
+                                                     difference=True)
+            planned_production_time = self.get_updated_planned_production_time(
+                input_time=duration, return_type=cal_type)
+            # operating time is production time
+            production_time = planned_production_time - request_data.downtime
 
             availability = self.oee_calc.calculate_availability(
-                operating_time=operating_time,
+                operating_time=production_time,
                 planned_prod_time=planned_production_time,
             )
 
             performance = self.oee_calc.calculate_performance(
                 units_produced=request_data.total_units,
-                operating_time=operating_time,
+                operating_time=production_time,
                 cycle_time=request_data.cycle_time,
             )
 
@@ -192,3 +215,16 @@ class OEEEngine:
             return batch_oee
         except Exception:
             raise
+
+    @staticmethod
+    def get_updated_planned_production_time(input_time, return_type):
+        if return_type == "minutes":
+            return input_time.minutes
+        elif return_type == "seconds":
+            return input_time.seconds
+        elif return_type == "hours":
+            return input_time.hours
+        elif return_type == "microseconds":
+            return input_time.microseconds
+        else:
+            return input_time.seconds

scripts/core/handlers/oee_handlers.py

 import traceback
 
+from scripts.constants.db_connections import mongo_client
+from scripts.db.mongo.ilens_configuration.collections import collection_constants
 from scripts.logging import logger
-from scripts.utils.mongo_util import MongoCollectionBaseClass
 from scripts.schemas.oee_config_schema import Oee_Tag_Mapping_List, Get_Oee_Tag, Update_Oee_Tags, Get_Project_Id
-from scripts.constants import CommonKeys
-from scripts.db.mongo.ilens_configuration import database
-from scripts.db.mongo.ilens_configuration.collections import collection_constants
-from scripts.utils.mongo_util import MongoCollectionBaseClass
-from scripts.schemas import mongo_schema
-
-from scripts.constants.db_connections import mongo_client
 
 
 class Oee_Services:

scripts/core/handlers/tag_handler.py

+from scripts.constants.db_connections import mongo_client
 from scripts.db.mongo.ilens_configuration.aggregations.tag_hierarchy import TagHierarchyAggregate
 from scripts.db.mongo.ilens_configuration.collections.tag_hierarchy import TagHierarchy
 from scripts.db.mongo.schema.tag_hierarchy import GetTagsLists
-from scripts.constants.db_connections import mongo_client
 
 
 class TagHierarchyHandler:
@@ -11,7 +11,7 @@ class TagHierarchyHandler:
     def get_tags_list_by_hierarchy(self, input_data: GetTagsLists):
         try:
             aggregate_query = TagHierarchyAggregate.tag_aggregate(project_id=input_data.project_id,
-                                                                  hierarchy_id=input_data.hierarchy_id)
+                                                                  hierarchy_id=input_data.hierarchy)
             tag_lists = self.tag_hierarchy_conn.get_tag_hierarchy_by_aggregate(query=aggregate_query)
             tag_lists = tag_lists[0] if tag_lists else {}
             return tag_lists

scripts/db/mongo/schema/tag_hierarchy.py

-from typing import Optional
-
 from pydantic import BaseModel
 
 
 class GetTagsLists(BaseModel):
-    project_id : str
-    hierarchy_id : str
+    project_id: str
+    hierarchy: str

scripts/errors/__init__.py

@@ -2,6 +2,12 @@ class ErrorMessages:
     UNKNOWN = "Unknown Error occurred"
     ERR001 = "Configurations not available, please verify the database."
     ERR002 = "Data Not Found"
+    K_ERROR1 = "Data Not Found in Time series Database"
+    K_ERROR2 = "Time series Database returned with an error"
+    K_ERROR3 = "Communication Error with Time series Database"
+    DF_ERROR1 = "Error occurred while forming Dataframe"
+    DF_ERROR2 = "Given group-by parameters are invalid"
+    META_ERROR1 = "Tags not Found in Meta"
 
 
 class ErrorCodes:
@@ -10,6 +16,8 @@ class ErrorCodes:
     ERR003 = "ERR003 - Operating Time is less than Productive Time"
     ERR004 = "ERR004 - Rejected Units is greater than Total Units"
     ERR005 = "ERR005 - Batch Start time not supplied"
+    ERR006 = "ERR006 - Total Units Tag not found!!"
+    ERR007 = "ERR007 - Reject Units Tag not found!!"
 
 
 class UnknownError(Exception):
@@ -64,3 +72,11 @@ class PathNotExistsException(Exception):
 
 class ImplementationError(Exception):
     pass
+
+
+class ILensErrors(Exception):
+    """Generic iLens Error"""
+
+
+class DataFrameFormationError(ILensErrors):
+    """Raise when there is an error during dataframe formation"""

scripts/schemas/batch_oee.py

@@ -3,9 +3,9 @@ from typing import Optional, Union, List
 from pydantic import BaseModel, validator
 
 
-# from scripts.utils.common_utils import CommonUtils
-#
-# common_utils = CommonUtils()
+from scripts.utils.common_utils import CommonUtils
+
+common_utils = CommonUtils()
 
 
 class GetProducts(BaseModel):
@@ -96,12 +96,12 @@ class ProductInfo(BaseModel):
 
 
 class OEEDataInsertRequest(BaseModel):
-    prod_start_time: Optional[str]
-    prod_end_time: Optional[str]
+    prod_start_time: str
+    prod_end_time: str
     prod_status: Optional[str] = "running"
     downtime: Optional[Union[float, int]]
-    hierarchy: Optional[str]
-    reference_id: Optional[str]
+    hierarchy: str
+    reference_id: str
     setup_time: Optional[Union[float, int]]
     cycle_time: Optional[Union[float, int]]
     total_units: Optional[Union[float, int]]
@@ -109,7 +109,7 @@ class OEEDataInsertRequest(BaseModel):
     tz: Optional[str] = 'Asia/Kolkata'
     project_id: str
     trigger_by: Optional[str]
-    uom: Optional[str]
+    uom: Optional[str] = "mins"
 
     class Config:
         schema_extra = {
@@ -120,29 +120,29 @@ class OEEDataInsertRequest(BaseModel):
                 "tz": "Asia/Kolkata",
                 "project_id": "project_099",
                 "reference_id": "reference_id",
-                "downtime": "",
-                "setup_time": "",
-                "cycle_time": "",
-                "total_units": "",
-                "reject_units": "",
-                "uom": "'"
+                "downtime": float(0),
+                "setup_time": float(0),
+                "cycle_time": float(5),
+                "total_units": float(0),
+                "reject_units": float(0),
+                "uom": "mins"
             }
         }
 
-    # @validator('prod_start_time')
-    # def date_format_validator_start_date(cls, v):
-    #     common_utils.check_date_format(v, "%Y-%m-%d %H:%M:%S")
-    #     return v
-    #
-    # @validator('prod_end_time')
-    # def date_format_validator_end_date(cls, v):
-    #     common_utils.check_date_format(v, "%Y-%m-%d %H:%M:%S")
-    #     return v
+    @validator('prod_start_time')
+    def date_format_validator_start_date(cls, v):
+        common_utils.check_date_format(v, "%Y-%m-%d %H:%M:%S")
+        return v
+
+    @validator('prod_end_time')
+    def date_format_validator_end_date(cls, v):
+        common_utils.check_date_format(v, "%Y-%m-%d %H:%M:%S")
+        return v
 
 
 class OEEDataSaveRequest(BaseModel):
-    prod_start_time: Optional[str]
-    prod_end_time: Optional[str]
+    prod_start_time: str
+    prod_end_time: str
     downtime: Optional[Union[float, int]] = 0
     hierarchy: Optional[str]
     reference_id: Optional[str]

scripts/utils/common_utils.py

 from datetime import datetime
 
 import pendulum
+import pytz
+import shortuuid
 
 from scripts.config import PathToServices
-from scripts.constants import Secrets
+from scripts.constants import Secrets, EndpointConstants
+from scripts.db.redis_connections import project_details_db
 from scripts.logging import logger
 from scripts.utils.auth_util import ILensRequest, AuthenticationError
+from scripts.utils.db_name_util import get_db_name
 from scripts.utils.security_utils.apply_encrytion_util import create_token
 
 
@@ -45,7 +49,7 @@ class CommonUtils:
                                       project_id=project_id)
         try:
             cookies = {'login-token': self.create_token(user_id=user_id), "user_id": user_id}
-            downtime_response = connection_obj.post(path="/downtime_log/get/overall_downtime",
+            downtime_response = connection_obj.post(path=EndpointConstants.hierarchy_downtime,
                                                     json={"project_id": project_id,
                                                           "hierarchy": hierarchy})
             response = downtime_response.json()
@@ -76,3 +80,36 @@ class CommonUtils:
         except Exception as e:
             logger.exception(str(e))
             raise
+
+    @staticmethod
+    def ts_from_format_as_timezone(_date, _format, tz):
+        date = datetime.strptime(_date, _format) \
+                   .astimezone(pytz.timezone(tz)).timestamp() * 1000
+        return date
+
+    @staticmethod
+    def convert_to_timestamp(x, date_format=None) -> int:
+        if not isinstance(x, int):
+            if isinstance(x, str):
+                x = int(datetime.strptime(x, date_format).timestamp() * 1000)
+            else:
+                x = int(x.timestamp() * 1000)
+        return x
+
+    @staticmethod
+    def datetime_from_str(date_str, date_format, tz):
+        localized_tz = pytz.timezone(tz)
+        datetime_with_tz = datetime.strptime(f"{date_str}", date_format)
+        localized_dt = localized_tz.localize(datetime_with_tz)
+        return localized_dt
+
+    @staticmethod
+    def get_next_id(_=None) -> str:
+        return shortuuid.uuid()
+
+    @staticmethod
+    def metric_name_by_project(metric, project_id):
+        """For DB splitting enabled projects"""
+        db_splitting_enabled = get_db_name(project_id=project_id,
+                                           database=metric, redis_client=project_details_db)
+        return db_splitting_enabled

scripts/utils/kairos_db_util/__init__.py

+from copy import deepcopy
+from dataclasses import dataclass
+from typing import List
+
+from scripts.config import DBConf
+from scripts.utils.common_utils import CommonUtils
+
+
+@dataclass
+class KairosConstants:
+    kairos_std_metric_name: str = 'ilens.live_data.raw'
+    kairos_complete_tag_key = "complete_tag"
+    kairos_column_names: tuple = ("timestamp", "value",)
+    kairos_time_unit: str = "ms"
+    kairos_version_api: str = "/api/v1/version"
+    kairos_health_check: str = "/api/v1/health/status"
+    kairos_query_api: str = "/api/v1/datapoints/query"
+    kairos_write_data_api: str = "/api_v1/v1/datapoints"
+    kairos_time_keys = ("start_absolute", "end_absolute",)
+    kairos_empty_result = {
+        "queries": [
+            {
+                "sample_size": 0, "results":
+                [
+                    {
+                        "name": kairos_std_metric_name, "tags": {}, "values": []
+                    }
+                ]
+            }
+        ]
+    }
+    kairos_query_levels = [{
+        "label": "Parameter",
+        "value": DBConf.KAIROS_DEFAULT_TAG
+    }, {
+        "label": "Complete Tag",
+        "value": DBConf.KAIROS_DEFAULT_FULL_TAG
+    }]
+    default_selection = [
+        {
+            "label": "Live Data",
+            "metric": "ilens.live_data.raw"
+        },
+        {
+            "label": "Error",
+            "metric": "ilens.live_data.error"
+        }
+    ]
+
+    # note: this doesn't work perfectly for months (31 days) or years (365 days)
+    SECONDS_IN_UNIT = {
+        'milliseconds': 0.001,
+        'seconds': 1,
+        'minutes': 60,
+        'hours': 3600,
+        'days': 86400,
+        'weeks': 604800,
+        'months': 2678400,
+        'years': 31536000
+    }
+
+    # constants used in get_range_needed
+    FETCH_BEFORE = 'prepend'
+    FETCH_AFTER = 'append'
+    FETCH_ALL = 'overwrite'
+
+
+class BaseQuery:
+    def __init__(self):
+        self.common_util = CommonUtils()
+
+    metric_query = {
+        "tags": {
+            f"{DBConf.KAIROS_DEFAULT_FULL_TAG}": [
+                "site$dept$equipment$tag"
+            ]
+        },
+        "name": KairosConstants.kairos_std_metric_name,
+        "group_by": [
+            {
+                "name": "tag",
+                "tags": [
+                    f"{DBConf.KAIROS_DEFAULT_FULL_TAG}"
+                ]
+            }
+        ],
+        "aggregators": [
+            {
+                "name": "last",
+                "sampling": {
+                    "value": "1",
+                    "unit": "hours"
+                },
+                "align_start_time": True
+            }
+        ]
+    }
+
+    kairos_query = {
+        "metrics": [
+
+        ],
+        "plugins": [
+
+        ],
+        "cache_time": DBConf.KAIROS_CACHE_TIME,
+        "start_absolute": 0,
+        "end_absolute": 0
+    }
+
+    static_kairos_query = {"metrics": [
+        {
+            "tags": {
+                "c2": [
+                    "dept_1238"
+                ]
+            },
+            "name": KairosConstants.kairos_std_metric_name,
+            "group_by": [
+                {
+                    "name": "tag",
+                    "tags": [
+                        "c2"
+                    ]
+                }
+            ]
+        }
+    ],
+        "plugins": [],
+        "cache_time": 0
+    }
+
+    site_status_query = {
+        "metrics": [
+            {
+                "tags": {
+                    "c1": "sitelist"  # parameter
+                },
+                "name": KairosConstants.kairos_std_metric_name,
+                "group_by": [
+                    {
+                        "name": "tag",
+                        "tags": [
+                            "c1"
+                        ]
+                    }
+                ],
+                "aggregators": "aggregators"  # parameter
+            }
+        ],
+        "plugins": [],
+        "cache_time": 0,
+        "start_relative": "start_relative"  # parameter
+    }
+
+    def form_generic_query(self, tags_list: List, start_epoch: int, end_epoch: int, project_id: str,
+                           metric_name=KairosConstants.kairos_std_metric_name, category=DBConf.KAIROS_DEFAULT_FULL_TAG):
+        metric_name = self.common_util.metric_name_by_project(metric=metric_name, project_id=project_id)
+        group_by_tags = deepcopy(tags_list)
+        group_by_tags.append(category)
+        query = {
+            "metrics": [
+                {
+                    "tags": {
+                        category: tags_list
+                    },
+                    "name": metric_name,
+                    "group_by": [
+                        {
+                            "name": "tag",
+                            "tags": group_by_tags
+                        }
+                    ]
+                }
+            ],
+            "plugins": [],
+            "cache_time": 0,
+            "start_absolute": start_epoch,
+            "end_absolute": end_epoch
+        }
+        return query

scripts/utils/kairos_db_util/df_formation_util.py

+import logging
+from typing import List
+
+import pandas as pd
+
+from scripts.config import DBConf
+from scripts.errors import DataFrameFormationError, ErrorMessages
+from . import KairosConstants
+
+
+def get_group_by(group_by_dict, group_by_tag_list, grouping_order):
+    hierarchy_id = []
+    if DBConf.KAIROS_DEFAULT_FULL_TAG in group_by_tag_list:
+        return default_group_by(group_by_dict)
+    for tag in grouping_order:
+        for each_group in group_by_dict:
+            if each_group.get("name") != "tag":  # Only group by tag is supported;
+                continue
+            all_groups = each_group.get("group", {})
+            if tag in group_by_tag_list and tag in all_groups:
+                hierarchy_id.append(all_groups[tag])
+    if not hierarchy_id:
+        raise DataFrameFormationError(ErrorMessages.DF_ERROR2)
+    return "$".join(hierarchy_id)
+
+
+def default_group_by(group_by_dict):
+    for each_group in group_by_dict:
+        if each_group.get("name") != "tag":  # Only group by tag is supported;
+            continue
+        all_groups = each_group.get("group", {})
+        if DBConf.KAIROS_DEFAULT_FULL_TAG in all_groups:
+            return all_groups[DBConf.KAIROS_DEFAULT_FULL_TAG]
+    raise DataFrameFormationError(ErrorMessages.DF_ERROR2)
+
+
+def form_df(metric, tags, values, df_index, tz, group_by_tags, grouping_order, group_by):
+    """Helper function for parsing Kairos data and forming dataframe for each results"""
+    df = pd.DataFrame(columns=["timestamp", "value"])
+    try:
+        selected_hierarchy = get_group_by(group_by, group_by_tags, grouping_order)
+        if not tags or not values:
+            new_column_name = selected_hierarchy
+            df.rename(columns={"value": new_column_name}, inplace=True)
+            df.set_index(df_index, inplace=True)
+            return df
+        df = pd.DataFrame(values, columns=KairosConstants().kairos_column_names)
+        if KairosConstants().kairos_std_metric_name in metric:
+            new_column_name = selected_hierarchy
+            df.rename(columns={"value": new_column_name}, inplace=True)
+        timezone = tz
+        df[df_index] = pd.to_datetime(df[df_index], unit=KairosConstants.kairos_time_unit)
+        df[df_index] = df[df_index].dt.tz_localize('UTC').dt.tz_convert(timezone)
+        df.set_index(df_index, inplace=True)
+        df.sort_index(inplace=True)
+        df[f'{selected_hierarchy}_diff'] = df[selected_hierarchy].diff()
+    except Exception as e:
+        logging.exception(e)
+    return df
+
+
+def get_statistics(df, metadata):
+    try:
+        if metadata.enable_stats:
+            metadata.data_statistics = {
+                "min": df.min().round(2).to_dict(),
+                "max": df.max().round(2).to_dict(),
+                "avg": df.mean().round(2).to_dict()
+            }
+    except Exception as e:
+        logging.exception(e)
+
+
+def create_kairos_df(response_data, tags_list: List, group_by_tags: List, master_df=None, df_index='timestamp',
+                     tz='Asia/kolkata'):
+    """
+    Definition for creating kairos DataFrame
+    """
+    if master_df is None:
+        master_df = pd.DataFrame()
+
+    try:
+        for each_response in response_data.get("queries"):
+            if each_response["sample_size"] == 0:
+                continue
+            for each_result in each_response["results"]:
+                df = form_df(
+                    metric=each_result.get("name"),
+                    tags=each_result.get("tags"),
+                    values=each_result.get("values"),
+                    group_by=each_result.get("group_by"),
+                    df_index=df_index,
+                    tz=tz,
+                    group_by_tags=group_by_tags,
+                    grouping_order=tags_list
+                )
+                df = df[~df.index.duplicated()]
+                master_df = master_df.combine_first(df)
+
+        master_df = master_df.fillna(0)
+        return master_df
+    except Exception as e:
+        raise DataFrameFormationError(e)

scripts/utils/kairos_db_util/query_kairos.py

+import asyncio
+import logging
+import math
+from copy import deepcopy
+from datetime import datetime, timedelta
+from pprint import pformat
+from typing import Dict, Tuple
+
+import httpx
+
+from scripts.constants import StatusCodes
+from scripts.errors import KairosDBError, ErrorMessages
+from scripts.utils.common_utils import CommonUtils
+from . import KairosConstants
+
+kairos_const = KairosConstants()
+httpx_timeout = httpx.Timeout(30, read=None)
+
+
+class KairosQuery:
+    def __init__(self, url):
+        self.url = url
+        self.common_util = CommonUtils()
+
+    def query(self, query_data: dict) -> dict:
+        """
+        An API Interface to query Kairos REST API with a standard query given as input
+        :param query_data: query json/dictionary as given in
+        documentation: https://kairosdb.github.io/docs/restapi/QueryMetrics.html#id3
+        :return: Dictionary with Data, if available
+        """
+        try:
+            with httpx.Client() as client:
+                logging.debug(pformat(query_data))
+                r = client.post(
+                    self.url + kairos_const.kairos_query_api,
+                    json=query_data,
+                    timeout=httpx_timeout
+                )
+                if r.status_code not in StatusCodes.SUCCESS_CODES:
+                    logging.error(f"KAIROS RETURNED {r.status_code}")
+                    raise KairosDBError
+                logging.debug(f"status code: {r.status_code}")
+                self.validate_results(r.json())
+                return r.json()
+        except Exception as e:
+            logging.exception(e)
+            raise
+
+    async def query_async(self, query_data: dict, client=None) -> dict:
+        """
+        An API Interface to query Kairos REST API with a standard query given as input
+        :param client: httpx.AsyncClient
+        :param query_data: query json/dictionary as given in
+        documentation: https://kairosdb.github.io/docs/restapi/QueryMetrics.html#id3
+        :return: Dictionary with Data, if available
+        """
+        try:
+            if client:
+                return await self.perform_query(query_data, client)
+            else:
+                async with httpx.AsyncClient() as client:
+                    return await self.perform_query(query_data, client)
+
+        except Exception as e:
+            logging.exception(e)
+            return kairos_const.kairos_empty_result
+
+    async def perform_query(self, query_data: dict, client):
+        r = client.post(self.url + kairos_const.kairos_query_api, json=query_data,
+                        timeout=httpx_timeout)
+        logging.debug(f"status code: {r.status_code}, elapsed time: {r.elapsed}")
+        self.validate_results(r.json())
+        return r.json()
+
+    @staticmethod
+    def validate_results(response) -> None:
+        """
+        Validates the response from Kairos. If errors are found, it raises an Error.
+        """
+        if not isinstance(response, dict):
+            raise KairosDBError(ErrorMessages.K_ERROR2)
+        if "errors" in response:
+            logging.error(f"Kairos returned with an error: {response.get('errors')}")
+            raise KairosDBError(ErrorMessages.K_ERROR2)
+
+    @staticmethod
+    def get_timedelta(value):
+        """ input has keys value, unit. common inputs noted start_relative, end_relative """
+        seconds = int(value['value']) * kairos_const.SECONDS_IN_UNIT[value['unit']]
+        return timedelta(seconds=seconds)
+
+    def get_needed_absolute_time_range(self, time_range, now=None):
+        """
+        Create date-times from Kairos timestamp data.
+        :param time_range: dict, containing Kairos timestamp data: keys {start,end}_{relative,absolute}.
+        :param now: datetime.datetime, optional. set to remove drift in time during execution.
+        :return: 2-tuple (start, end), both datetime.datetime. end may be NoneType.
+        """
+        if not now:
+            now = datetime.now()
+        if time_range.get('start_absolute'):
+            start = datetime.fromtimestamp(int(time_range['start_absolute']) / 1000)
+        else:
+            td = self.get_timedelta(time_range.get('start_relative'))
+            start = now - td
+
+        if time_range.get('end_absolute'):
+            end = datetime.fromtimestamp(int(time_range['end_absolute']) / 1000)
+        elif time_range.get('end_relative'):
+            td = self.get_timedelta(time_range.get('end_relative'))
+            end = now - td
+        else:
+            end = None
+        return start, end
+
+    def get_chunked_time_ranges(self, time_range):
+        """
+        Given a long kairos range, return N timestamp pairs so we can parallelize COLD calls (new->old).
+        This implements up to second precision.
+        :param time_range: dict, generated by populate_time_range containing kairos-formatted keys.
+        :return: list of 2-tuples of datetime.datetime
+        """
+        chunk_length = 3600  # 1 hour default
+        num_chunks = 10
+        now = datetime.now()
+
+        start_time, end_time = self.get_needed_absolute_time_range(time_range, now)
+        if not end_time:
+            end_time = now
+        start_time = start_time.replace(microsecond=0)
+        end_time = end_time.replace(microsecond=0)
+
+        elapsed_secs = (end_time - start_time).total_seconds()
+        if elapsed_secs <= chunk_length:
+            return (self.common_util.convert_to_timestamp(start_time), self.common_util.convert_to_timestamp(end_time)),
+        else:
+            # need to increase chunk length to fit into max chunks
+            if elapsed_secs > chunk_length * num_chunks:
+                chunk_length = int(elapsed_secs / num_chunks)
+            # need to downsize max chunks because we can use them of size chunk_length
+            else:
+                num_chunks = int(math.ceil(elapsed_secs / chunk_length))
+
+            chunks = []
+            length_td = timedelta(seconds=chunk_length)
+            # end_time is mutated below here
+            for _ in range(num_chunks):
+                start = end_time - length_td
+
+                # Sanity check: make sure we limit the earliest chunk if it's partial
+                if start < start_time:
+                    start = start_time
+                # Add an offset so the chunks don't overlap.
+                start += timedelta(seconds=1)
+                chunks.append(
+                    (self.common_util.convert_to_timestamp(start), self.common_util.convert_to_timestamp(end_time)))
+                end_time -= length_td
+            return chunks
+
+    @staticmethod
+    async def form_query(query: Dict, start_time: int, end_time: int):
+        """
+        Helper function for forming a query after chunking
+        """
+        __new_query__ = deepcopy(query)
+        __new_query__.update(
+            {
+                kairos_const.kairos_time_keys[0]: start_time,
+                kairos_const.kairos_time_keys[1]: end_time
+            }
+        )
+        return __new_query__
+
+    async def parallel_query(self, query) -> Tuple:
+        """
+        A helper function that has built-in algorithm to split queries into a given number of chucks.
+        The number of chunks are configurable; some aggregations may not work well when using this.
+
+        The primary purpose of this function is to query large amount of data in parallel to reduce
+        query time and load.
+        """
+        logging.debug(f"kairos query : , {query}")
+        time_range = {key: query[key] for key in kairos_const.kairos_time_keys}
+        time_chunks = self.get_chunked_time_ranges(time_range)
+
+        queries = asyncio.gather(*[
+            self.form_query(
+                query, start_time=i[0],
+                end_time=i[1]) for i in time_chunks
+        ])
+        async with httpx.AsyncClient() as client:
+            results = await asyncio.gather(*[self.query_async(i, client) for i in queries])
+        return results