v1

.idea/.gitignore

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

.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 (task8redis_mqtt_receiver)" 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/task8redis_mqtt_receiver.iml" filepath="$PROJECT_DIR$/.idea/task8redis_mqtt_receiver.iml" />
+    </modules>
+  </component>
+</project>
\ No newline at end of file

.idea/task8redis_mqtt_receiver.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="inheritedJdk" />
+    <orderEntry type="sourceFolder" forTests="false" />
+  </component>
+</module>
\ 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

app.py

+import uvicorn
+
+from scripts.config.applications_config import uvicorn_host, uvicorn_port
+
+# starting the application
+if __name__ == "__main__":
+    print("redis MQTT task")
+    uvicorn.run("scripts.services.receiver_app_services_run:app", host=uvicorn_host, port=int(uvicorn_port))

conf/application.conf

+[path]
+base_path = scripts/
+sub_path = temp/
+
+
+[mqtt]
+topic = receptionist
+mqtt_host = 192.168.0.220
+port = 1883
+requests = 60
+
+[uvicorn]
+uvicorn_host = 127.0.0.1
+uvicorn_port = 8001
+
+# database servers
+[connection]
+mongodb = mongodb://localhost:27017
\ No newline at end of file

scripts/config/applications_config.py

+import configparser
+
+
+config = configparser.RawConfigParser()
+config.read("conf/application.conf")
+
+
+# path
+base_path = config.get("path", 'base_path')
+sub_path = config.get("path", "sub_path")
+
+
+# mqtt
+topic_name = config.get("mqtt", "topic")
+mqtt_host = config.get("mqtt", "mqtt_host")
+port = config.get("mqtt", "port")
+request_no = config.get("mqtt", "requests")
+
+# db connection
+client_connect = config.get("connection", "mongodb")
+
+
+# uvicorn
+uvicorn_host = config.get("uvicorn", "uvicorn_host")
+uvicorn_port = config.get("uvicorn", "uvicorn_port")
+
+

scripts/core/engine/mongo_connect.py

+# connecting to mongodb server
+from pymongo import MongoClient
+
+from scripts.config.applications_config import client_connect
+
+# connect to server
+client_mongo = MongoClient(client_connect)

scripts/core/engine/redis_db.py

+import redis
+
+conn = redis.Redis('127.0.0.1')

scripts/core/handlers/insert_mongo_db.py

+# insert the json to mongodb
+from scripts.utils.mongo_db_class import MongoDb
+
+
+def insert_data_mongo(client_mongo, json_data):
+    obj_mongo = MongoDb(client_mongo)
+    obj_mongo.insert_db(json_data)

scripts/core/handlers/redis_insert_json.py

+from scripts.core.engine.redis_db import conn
+
+
+def insert_to_redis(json_data):
+    for key, dict_items in enumerate(json_data):
+        if not conn.exists(dict_items['patient_id']):
+            conn.hmset(dict_items['patient_id'], dict_items)
+        else:
+            continue
+    count = conn.dbsize()
+    print("Number of items in Redis:", count)
+    patient_data_list = []
+    for index in range(2):  # assuming there are 2 hashes stored
+        hash_key = f"patient_{index}"
+        patient_data = conn.hgetall(hash_key)
+        print(patient_data)
+        patient_data_list.append(patient_data)
+    print(patient_data_list)

scripts/services/receiver_app_services_run.py

+import paho.mqtt.client as mqtt
+from fastapi import FastAPI
+import json
+import redis
+import pandas as pd
+
+from scripts.config.applications_config import topic_name, mqtt_host, port, request_no
+from scripts.core.engine.mongo_connect import client_mongo
+from scripts.core.engine.redis_db import conn
+from scripts.core.handlers.insert_mongo_db import insert_data_mongo
+from scripts.core.handlers.redis_insert_json import insert_to_redis
+
+# This is the Subscriber
+
+app = FastAPI()
+
+
+@app.get("/")
+def receiver():
+    def on_connect(client_name, userdata, flags, rc):
+        print(userdata, " ", flags)
+        print("Connected with result code " + str(rc))
+        no_doctors = conn.get("Doctors")
+        for doctors in range(0, int(no_doctors)):
+            client_name.subscribe(f'doctors_{doctors}')
+
+    def on_message(client_name, userdata, msg):
+        print(client_name, " ", userdata)
+        # decoding the msg to string
+        payload_decoded = msg.payload.decode('utf-8')
+        json_data = json.loads(payload_decoded)
+        print(json_data)
+        # decoded msg to dataframe
+        # frames = pd.DataFrame(json_data, index=None)
+        # insert_data_mongo(client_mongo, json_data)
+        # insert_to_redis(json_data)
+
+    client = mqtt.Client()
+    client.connect(mqtt_host, int(port), int(request_no))
+    client.on_connect = on_connect
+    client.on_message = on_message
+    client.loop_forever()

scripts/utils/mongo_db_class.py

+# class for mongodb
+class MongoDb:
+    def __init__(self, client_mongo):
+        try:
+            self.client_con = client_mongo
+        except Exception as e:
+            print("Connection failed: ", e)
+        else:
+            db = self.client_con['patient_details']
+            self.patients = db['patients']
+
+    def insert_db(self, json_data):
+        try:
+            self.patients.insert_many(json_data)
+        except Exception as e:
+            print("Connection failed: ", e)