addd

main.py

-import uvicorn
-
-from scripts.services.task1 import report_router
-from fastapi import FastAPI
-app = FastAPI()
-app.include_router(report_router)
-if __name__ == "__main__":
-    uvicorn.run ("main:app", host="0.0.0.0", port=2747)
+
+import json
+import time
+
+import redis
+
+# from scripts.core.handler.task3 import num
+
+r = redis.Redis(
+    host='192.168.0.220',
+    port=6379,
+    db=111)
+num = 1
+while True:
+    data = {
+        "data": {"patient_id": num}
+    }
+    num = num + 1
+    r.set("num", json.dumps(data))
+    print(r.get("num"))
+    time.sleep(5)

scripts/core/handler/task1.py

-import pymongo
-import pandas as pd
-from pydantic import BaseModel
-
-myclient = pymongo.MongoClient("mongodb://192.168.0.220:2747/")
-mydb = myclient["factory"]
-mycol = mydb["status"]
-mycol.find()
-
-class Employee_details(BaseModel):
-
-    id: str
-    certificate_number: int
-    business_name: str
-    date: str
-    result: str
-    sector: str
-    address: dict
-# #
-
-dblist = myclient.list_database_names()
-if "factory" in dblist:
-    print("The database exists.")
-
-
-
-class Task1:
-
-    def read_violation(self):
-        agg = list(mycol.aggregate(
-            [
-                {
-                    '$project': {
-                        '_id': '$business_name',
-                        'result': 'Violation Issued'
-                    }
-                }, {
-                '$group': {
-                    '_id': '$_id',
-                    'heghest': {
-                        '$max': '$result'
-                    }
-                }
-            }, {
-                '$sort': {
-                    'highest': -1
-                }
-            }
-            ]))
-        return (agg[0])
-
-
-    def read_noviolation(self):
-        r = mycol.find({"result": "No Violation Issued"})
-        lst = []
-        for i in r:
-            lst.append(i["business_name"])
-        return lst
-
-
-
-
-
-
-
-
-    # class Employee_details(BaseModel):
-    # )
-    #     id: str
-    #     certificate_number: int
-    #     business_name: str
-    #     date: str
-    #     result: str
-    #     sector: str
-    #     address: dict
-
-
-
-    def Fnd(self):
-        q = mycol.find({})
-        if q:
-            a = []
-            for i in q:
-                d = {'Business Name': i['business_name'], 'Date': i["date"], 'Result': i["result"]}
-                a.append(d)
-            a = pd.DataFrame(a)
-            return (a.to_excel("incpection.xlsx", sheet_name="mySheet", index=False))
-        else:
-            return "no data"
-
-
-
-
-
-
-    def Insert(user):
-        mycol.insert_one(user.dict())
-        return "Inserted"
-
-    def check_result(business_name, record_type):
-        x = business_name
-        y = record_type
-        cnt = mycol.count_documents({'$and':[{'business_name':x},{'result':y}]})
-        return str(cnt) +" " +str( y)
\ No newline at end of file

scripts/core/handler/task3.py

+import json
+import time
+
+import redis
+from paho.mqtt import client as mqtt_client
+
+r = redis.Redis(
+    host='192.168.0.220',
+    port=6379,
+    db=111)
+
+broker = '192.168.0.220'
+port = 1883
+topic = ["doc-1", "doc-2", "doc-3"]
+# generate client ID with pub prefix randomly
+client_id = f'["doc-1","doc-2","doc-3"]'
+username = 'Mad'
+password = 'public'
+
+
+def connect_mqtt() -> mqtt_client:
+    def on_connect(client, userdata, flags, rc):
+        if rc == 0:
+            print("Connected to MQTT Broker!")
+        else:
+            print("Failed to connect, return code %d\n", rc)
+
+    client = mqtt_client.Client(client_id)
+    client.username_pw_set(username, password)
+    client.on_connect = on_connect
+    client.connect(broker, port)
+    return client
+
+
+def publish(client: object) -> object:
+    msg_count = 0
+    while True:
+        time.sleep(1)
+        counter = 0
+        num = 1
+        patients = {}
+        while True:
+            msg = {"patient_id": num}
+            result = client.publish(topic[int((counter) % len(topic))], str(msg))
+            print(result)
+            status = result[0]
+            if status == 0:
+                print(f"Send `{str(msg)}` to topic `{topic[int((counter) % len(topic))]}`")
+                if topic[int(counter % len(topic))] not in patients.keys():
+                    patients[topic[int(counter % len(topic))]] = []
+                    patients[topic[int(counter % len(topic))]].append(str(msg['patient_id']))
+                else:
+                    if str(msg['patient_id']) not in patients[topic[int(counter % len(topic))]]:
+                        patients[topic[int(counter % len(topic))]].append(str(msg['patient_id']))
+            else:
+                print(f"Failed to send message to topic {topic}")
+            # counter = counter + 1
+            # num = num + 1
+            if num == 5:
+                # num=1
+                if str(msg['patient_id']) in patients[topic[int(counter % len(topic))]]:
+                    return patients
+            else:
+                num = num + 1
+            counter = counter + 1
+            print(patients)
+            time.sleep(0)
+            # patient_id= patient_id+1
+            # time.sleep(5)
+            # print(counter)
+
+            # msg_count += 1
+
+
+# def run():
+#     client = connect_mqtt()
+#     client.loop_start()
+#     publish(client)
+
+
+def run():
+    client = connect_mqtt()
+    client.loop_start()
+    data = publish(client)
+    # print(data)
+    # redis_conn()
+    return data
+
+
+def redis_conn():
+    data = run()
+    # print(data)
+    print(data)
+    if data:
+        r.set("token_details", json.dumps(data))
+        return "data inserted"
+
+
+# data = run()
+redis_conn()

scripts/services/task1.py

-from fastapi import APIRouter
-from scripts.core.handler.task1 import Task1
-report_router= APIRouter()
-@report_router.get("/getdata1")
-def most_violation():
-    return Task1().read_violation()
-
-@report_router.get("/getdata2")
-def no_violation():
-    return Task1().read_noviolation()
-
-@report_router.get('/getdata')
-def get_data():
-    return Task1().Fnd()
-
-@report_router.get('/getDetails')
-def checkdata(business_name ,record_type):
-    return Task1().check_result(business_name ,record_type)
\ No newline at end of file

scripts/services/task3.py

+import random
+import time
+
+from paho.mqtt import client as mqtt_client
+
+
+broker = '192.168.0.220'
+port = 1883
+topic = "python/mqtt"
+# generate client ID with pub prefix randomly
+client_id = f'python-mqtt-{random.randint(0, 1000)}'
+username = 'Mad'
+password = 'public'
+
+def connect_mqtt():
+    def on_connect(client, userdata, flags, rc):
+        if rc == 0:
+            print("Connected to MQTT Broker!")
+        else:
+            print("Failed to connect, return code %d\n", rc)
+
+    client = mqtt_client.Client(client_id)
+    client.username_pw_set(username, password)
+    client.on_connect = on_connect
+    client.connect(broker, port)
+    return client
+
+
+def publish(client):
+    msg_count = 0
+    while True:
+
+        time.sleep(1)
+        msg = f"messages: {msg_count}"
+        result = client.publish(topic, msg)
+        # result: [0, 1]
+        status = result[0]
+        if status == 0:
+            print(f"Send {msg} to topic {topic}")
+        else:
+            print(f"Failed to send message to topic {topic}")
+        msg_count += 1
+
+
+def run():
+    client = connect_mqtt()
+    client.loop_start()
+    publish(client)
+
+
+if __name__ == '__main__':
+    run()
+