pub sub completed

.env

+MQTT_HOST=192.168.0.220
+MQTT_PORT=1883
+MQTT_USERNAME=admin
+MQTT_PASSWORD=admin
+REDIS_HOST=127.0.0.1
+REDIS_PORT=6379
+TOKEN=site/
+FILE_PATH=r"C:\Users\logesh.n\PycharmProjects\site_redis_task\temp"
\ No newline at end of file

.idea/.gitignore

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

.idea/inspectionProfiles/Project_Default.xml

+<component name="InspectionProjectProfileManager">
+  <profile version="1.0">
+    <option name="myName" value="Project Default" />
+    <inspection_tool class="PyUnresolvedReferencesInspection" enabled="true" level="WARNING" enabled_by_default="true">
+      <option name="ignoredIdentifiers">
+        <list>
+          <option value="random.random.randint" />
+        </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.10 (site_redis_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/site_redis_task.iml" filepath="$PROJECT_DIR$/.idea/site_redis_task.iml" />
+    </modules>
+  </component>
+</project>
\ No newline at end of file

.idea/site_redis_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="inheritedJdk" />
+    <orderEntry type="sourceFolder" forTests="false" />
+  </component>
+  <component name="PackageRequirementsSettings">
+    <option name="removeUnused" value="true" />
+  </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

conf/config_app.conf

+[file_path]
+path_name=temp/

pub.py

+import json
+import random
+import time
+from paho.mqtt import client
+
+from scripts.config.app_config import mqtt_username, mqtt_password, mqtt_host, mqtt_port, token
+
+
+def connect_mqtt():
+    def on_connect(client, userdata, flags, rc):
+        if rc == 0:
+            print("Connected to MQTT Broker!", client)
+        else:
+            print("Failed to connect, return code %d\n", rc)
+
+    mqtt_client = client.Client()
+    mqtt_client.username_pw_set(mqtt_username, mqtt_password)
+    mqtt_client.on_connect = on_connect
+    mqtt_client.connect(mqtt_host, int(mqtt_port))
+    return mqtt_client
+
+
+def publish(client):
+
+    while True:
+        time.sleep(10)
+        data = {"data": {"PM10": random.randint(0, 100),
+                         "PM2.5": random.randint(0, 100),
+                         "SO2": random.randint(0, 100),
+                         "NO2": random.randint(0, 100)},
+                "site_id": "site"+str(random.randint(0, 100)),
+                "data_quality": random.randint(0, 2)
+                }
+        data_json = json.dumps(data)
+        result = client.publish(token, data_json)
+        # result: [0, 1]
+        status = result[0]
+        if status == 0:
+            print(f"Send `{data}` to topic `{token}`")
+        else:
+            print(f"Failed to send message to topic {token}")
+
+
+def run():
+    mqtt_client = connect_mqtt()
+    mqtt_client.loop_start()
+    publish(mqtt_client)
+
+
+if __name__ == '__main__':
+    run()

requirements.txt

+python-dotenv~=0.21.1
+redis~=4.5.1
+paho-mqtt~=1.6.1
\ No newline at end of file

scripts/config/app_config.py

+import os
+from dotenv import load_dotenv
+from scripts.logging.logger import logger
+
+try:
+    load_dotenv()
+    token = os.getenv("TOKEN")
+    mqtt_host = os.getenv("MQTT_HOST")
+    mqtt_port = os.getenv("MQTT_PORT")
+    mqtt_username = os.getenv("MQTT_USERNAME")
+    mqtt_password = os.getenv("MQTT_PASSWORD")
+    redis_host = os.getenv("redis_host")
+    redis_port = os.getenv("redis_port")
+
+except Exception as e:
+    logger.error(e)
+    print("Exception occurred in app_config file > ", e)

scripts/config/databases/redis/redis_connection.py

+import redis
+
+from scripts.config.app_config import redis_port, redis_host
+
+try:
+    # redis connection
+    connection_1 = redis.Redis(redis_host, redis_port, db=0)
+    connection_2 = redis.Redis(redis_host, redis_port, db=1)
+    connection_3 = redis.Redis(redis_host, redis_port, db=2)
+except Exception as e:
+    print("Exception in redis connection > ", e)

scripts/logging/logger.py

+import logging
+import os
+from logging.handlers import RotatingFileHandler
+
+
+def get_logger():
+    """
+     Creates a rotating log
+     """
+    __logger__ = logging.getLogger('')
+    __logger__.setLevel("ERROR")
+    log_formatter = '%(asctime)s - %(levelname)-6s - [%(threadName)5s:%(funcName)5s():%(lineno)s] - %(message)s'
+    time_format = "%Y-%m-%d %H:%M:%S"
+    formatter = logging.Formatter(log_formatter, time_format)
+    log_files = os.path.join(r"C:\Users\logesh.n\PycharmProjects\site_redis_task\temp\_ERROR.log")
+    temp_handler = RotatingFileHandler(log_files, maxBytes=100000000, backupCount=5)
+    temp_handler.setFormatter(formatter)
+
+    __logger__.addHandler(temp_handler)
+
+    return __logger__
+
+
+logger = get_logger()

sub.py

+import json
+import random
+import time
+from paho.mqtt import client
+
+from scripts.config.app_config import mqtt_username, mqtt_password, mqtt_host, mqtt_port, token
+
+
+def connect_mqtt() -> client:
+    def on_connect(client, userdata, flags, rc):
+        if rc == 0:
+            print("Connected to MQTT Broker!", client)
+        else:
+            print("Failed to connect, return code %d\n", rc)
+
+    mqtt_client = client.Client()
+    mqtt_client.username_pw_set(mqtt_username, mqtt_password)
+    mqtt_client.on_connect = on_connect
+    mqtt_client.connect(mqtt_host, int(mqtt_port))
+    return mqtt_client
+
+
+def subscribe(mqtt_client: client):
+    def on_message(client, userdata, msg):
+        data = msg.payload.decode()
+        data_dict = json.loads(data)
+        print(f"Received `{data_dict}` from `{msg.topic}` topic")
+
+    mqtt_client.subscribe(token)
+    mqtt_client.on_message = on_message
+
+
+def run():
+    mqtt_client = connect_mqtt()
+    subscribe(mqtt_client)
+    mqtt_client.loop_forever()
+
+
+if __name__ == '__main__':
+    run()