task completed

.env

+START_TIME = 2022-11-08 00:00:00.000
+END_TIME = 2023-02-07 00:00:00.000
+
+kairos_api = http://192.168.0.220:8080/api/v1/datapoints/query
+MONGO_URI = 192.168.0.220:2717
+
+tags = c3
+name = ilens.live_data.raw
+
+USER = admin
+PASSWORD = iLensDevMongo783
+HIERARCHY_ID = site_114
+PROJECT_ID = project_099

.idea/.gitignore

+# Default ignored files
+/shelf/
+/workspace.xml

.idea/KairosDB-task.iml

+<?xml version="1.0" encoding="UTF-8"?>
+<module type="PYTHON_MODULE" version="4">
+  <component name="NewModuleRootManager">
+    <content url="file://$MODULE_DIR$">
+      <excludeFolder url="file://$MODULE_DIR$/venv" />
+    </content>
+    <orderEntry type="jdk" jdkName="Python 3.9 (KairosDB-task)" jdkType="Python SDK" />
+    <orderEntry type="sourceFolder" forTests="false" />
+  </component>
+</module>
\ No newline at end of file

.idea/inspectionProfiles/Project_Default.xml

+<component name="InspectionProjectProfileManager">
+  <profile version="1.0">
+    <option name="myName" value="Project Default" />
+    <inspection_tool class="PyPep8NamingInspection" enabled="true" level="WEAK WARNING" enabled_by_default="true">
+      <option name="ignoredErrors">
+        <list>
+          <option value="N801" />
+        </list>
+      </option>
+    </inspection_tool>
+  </profile>
+</component>
\ No newline at end of file

.idea/inspectionProfiles/profiles_settings.xml

+<component name="InspectionProjectProfileManager">
+  <settings>
+    <option name="USE_PROJECT_PROFILE" value="false" />
+    <version value="1.0" />
+  </settings>
+</component>
\ No newline at end of file

.idea/misc.xml

+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="ProjectRootManager" version="2" project-jdk-name="Python 3.9 (KairosDB-task)" project-jdk-type="Python SDK" />
+</project>
\ No newline at end of file

.idea/modules.xml

+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="ProjectModuleManager">
+    <modules>
+      <module fileurl="file://$PROJECT_DIR$/.idea/KairosDB-task.iml" filepath="$PROJECT_DIR$/.idea/KairosDB-task.iml" />
+    </modules>
+  </component>
+</project>
\ No newline at end of file

.idea/vcs.xml

+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="VcsDirectoryMappings">
+    <mapping directory="$PROJECT_DIR$" vcs="Git" />
+  </component>
+</project>
\ No newline at end of file

main.py

+from scripts.services.app_run_service import main_service_run
+
+main_service_run()

scripts/config/applications_config.py

+import os
+from dotenv import load_dotenv
+
+load_dotenv()
+
+from_date = os.getenv("START_TIME")
+end_date = os.getenv("END_TIME")
+
+kairos_api = os.getenv("kairos_api")
+
+mongodb_uri = os.getenv("MONGO_URI")
+port = os.getenv("PORT")
+server_path = os.getenv("SERVER_PATH")
+
+name = os.getenv("name")
+
+project_id = os.getenv("PROJECT_ID")
+hierarchy_id = os.getenv("HIERARCHY_ID")
+username = os.getenv("USER")
+password = os.getenv("PASSWORD")
+

scripts/core/handlers/convert_time_epoch.py

+from datetime import datetime
+
+
+def time_get_epoch(time_string):
+    # Convert the timestamp string into a datetime object
+    timestamp_dt = datetime.strptime(time_string, "%Y-%m-%d %H:%M:%S.%f")
+
+    # Calculate the difference between the timestamp and Unix reference time in seconds
+    epoch = int(timestamp_dt.timestamp()) * 1000
+    return epoch

scripts/core/handlers/create_csv_data.py

+import pandas as pd
+
+from scripts.core.handlers.epoch_to_date_time import epoch_date_time
+
+
+def create_csv(key_, data_):
+    da = pd.DataFrame(columns=['Timestamp'])
+    for key, value in data_.items():
+        dp = pd.DataFrame()
+        if len(value) != 0:
+            dp[['Timestamp', key_[key]]] = [[epoch_date_time(time), data] for time, data in value]
+        da = pd.concat([da, dp])
+    new_da = da.fillna("NA")
+    new_da.to_csv("scripts/temp/ilens.csv", index=False)

scripts/core/handlers/dictionary_values_tags.py

+import json
+
+
+def key_value_dict(query_data):
+    resp = json.loads(query_data.text)
+    temp_json = {}
+
+    if resp["queries"][0]["sample_size"] > 0:
+        for i in range(0, len(resp["queries"][0]["results"])):
+            c3 = resp["queries"][0]["results"][i]["tags"]["c3"][0]
+            master_key = c3
+            if master_key not in temp_json:
+                temp_json[master_key] = resp["queries"][0]["results"][i]["values"]
+            else:
+                temp_json[master_key].extend(resp["queries"][0]["results"][i]["values"])
+    return temp_json
+

scripts/core/handlers/epoch_to_date_time.py

+from datetime import datetime
+
+
+def epoch_date_time(epoch):
+    # Convert the datetime object into a timestamp string
+    timestamp_dt = datetime.fromtimestamp(epoch/1000).strftime("%Y-%m-%d %H:%M:%S")
+
+    return timestamp_dt

scripts/core/handlers/query_to_json.py

+import json
+
+from scripts.utils.query_kairos import query_kairos_get
+
+
+def get_query(tag_query, tag_name, start_time, end_time):
+    query_ = query_kairos_get(tag_query, tag_name, start_time, end_time)
+    query = json.dumps(query_)
+    return query

scripts/core/handlers/send_data.py

+from pymongo import MongoClient
+import pandas as pd
+from scripts.config.applications_config import mongodb_uri, hierarchy_id, username, password
+
+client = MongoClient(mongodb_uri,
+                     username=username,
+                     password=password)
+database = client['ilens_configuration']
+collection = database['tag_hierarchy']
+
+
+class SendData:
+    @staticmethod
+    def extract_data():
+        try:
+            tag_hierarchy = collection
+            query = {}
+            hierarchies = hierarchy_id.split('$')
+            for each_hierarchy in hierarchies:
+                if "site_" in each_hierarchy:
+                    query["site_id"] = each_hierarchy
+                if "dept_" in each_hierarchy:
+                    query["dept_id"] = each_hierarchy
+                elif "line_" in each_hierarchy:
+                    query["line_id"] = each_hierarchy
+                elif "equipment_" in each_hierarchy:
+                    query["equipment_id"] = each_hierarchy
+
+            pipeline = [
+                {
+                    "$match": query
+                },
+                {
+                    '$lookup': {
+                        'from': 'tags',
+                        'localField': 'parameter_id',
+                        'foreignField': 'id',
+                        'as': 'tag_details'
+                    }
+                }, {
+                    '$unwind': '$tag_details'
+                }, {
+                    '$group': {
+                        '_id': '$id',
+                        'first_document': {
+                            '$first': '$$ROOT'
+                        }
+                    }
+                }, {
+                    '$replaceRoot': {
+                        'newRoot': '$first_document'
+                    }
+                }, {
+                    '$project': {
+                        'site_id': '$site_id',
+                        'line_id': '$line_id',
+                        'equipment_id': '$equipment_id',
+                        'site_name': '$site_name',
+                        'dept_name': '$dept_name',
+                        'line_name': '$line_name',
+                        'equipment_name': '$equipment_name',
+                        'id': '$id',
+                        'tag_name': '$tag_details.tag_name',
+                        '_id': 0
+                    }
+                }
+            ]
+            tag_names = tag_hierarchy.aggregate(pipeline)
+            data = list(tag_names)
+            dataset = pd.DataFrame(data)
+            return dataset
+        except Exception as e:
+            print(e)
+
+    @staticmethod
+    def final_dict(dataframe):
+        try:
+
+            final_data = {}
+            for index, row in dataframe.iterrows():
+                hierarchy_name = ""
+                if row["site_name"] is not None:
+                    hierarchy_name = row["site_name"]
+                if row["dept_name"] != "":
+                    hierarchy_name = hierarchy_name + ">" + row["dept_name"]
+                if row["line_name"] != "":
+                    hierarchy_name = hierarchy_name + ">" + row["line_name"]
+                if row["equipment_name"] != "":
+                    hierarchy_name = hierarchy_name + ">" + row["equipment_name"]
+                if row["tag_name"] != "":
+                    hierarchy_name = hierarchy_name + ":" + row["tag_name"]
+                new_dict = {row["id"]: hierarchy_name}
+                final_data.update(new_dict)
+            return final_data
+        except Exception as e:
+            print(e)

scripts/services/app_run_service.py

+import requests
+
+from scripts.config import applications_config
+from scripts.core.handlers.convert_time_epoch import time_get_epoch
+from scripts.core.handlers.create_csv_data import create_csv
+from scripts.core.handlers.dictionary_values_tags import key_value_dict
+from scripts.core.handlers.query_to_json import get_query
+from scripts.core.handlers.send_data import SendData
+
+
+def main_service_run():
+    # Define the base URL for the KairosDB API
+    base_url = applications_config.kairos_api
+    start_time = time_get_epoch(applications_config.from_date)
+    end_time = time_get_epoch(applications_config.end_date)
+
+    mongo_obj = SendData()
+    df = mongo_obj.extract_data()
+    key_ = mongo_obj.final_dict(df)
+
+    dict_ = []
+    for key, value in key_.items():
+        dict_.append(key)
+    query_data = get_query(dict_, applications_config.name, start_time, end_time)
+    # Send the query to the KairosDB API
+    response = requests.post(
+        base_url,
+        data=query_data
+    )
+    data_ = key_value_dict(response)
+    create_csv(key_, data_)

scripts/utils/query_kairos.py

+def query_kairos_get(tag_query, tag_name, start_time, end_time):
+    query = {
+        "metrics": [
+            {
+                "tags": {
+                    "c3":
+                        tag_query
+
+                },
+                "name": f'{tag_name}',
+                "group_by": [
+                    {
+                        "name": "tag",
+                        "tags": [
+                            "c3"
+                        ]
+                    }
+                ],
+                "aggregators": [
+                    {
+                        "name": "sum",
+                        "sampling": {
+                            "value": "1",
+                            "unit": "milliseconds"
+                        },
+                        "align_sampling": "true"
+                    }
+                ]
+            }
+        ],
+        "plugins": [],
+        "cache_time": 0,
+        "time_zone": "Asia/Calcutta",
+        "start_absolute": f'{int(start_time)}',
+        "end_absolute": f'{int(end_time)}'
+    }
+    return query