HAN-7: Implementar conexión con RabbitMQ y envío de paquetes

.env.dist

@@ -0,0 +1,5 @@
+RABBITMQ_HOST=
+QUEUE_NAME=
+LOGGING_LEVEL=
+WEATHER_API_KEY=
+WEATHER_API_URL=

.github/workflows/linters.yml

@@ -0,0 +1,31 @@
+
+name: Linters
+
+on: 'push'
+
+jobs:
+  run-linters:
+    name: Run linters
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Check out Git repository
+        uses: actions/checkout@v2
+
+      - name: Set up Python
+        uses: actions/setup-python@v1
+        with:
+          python-version: 3.8
+
+      - name: Install Python dependencies
+        run: pip install black flake8
+
+      - name: Run linters
+        uses: wearerequired/lint-action@v1
+        with:
+          black: true
+
+      - name: flake8 Lint
+        uses: py-actions/flake8@v1.2.0
+        with:
+          max-line-length: "88"

Dockerfile

@@ -0,0 +1,10 @@
+FROM python:3.9.7-slim
+
+WORKDIR /app
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY src src
+
+CMD ["python", "src/main.py"]

Makefile

@@ -0,0 +1,20 @@
+default: docker-compose-up
+
+all:
+
+docker-image:
+	docker build -f Dockerfile -t main-app .
+.PHONY: docker-image
+
+docker-compose-up: docker-image
+	docker-compose -f docker-compose.yaml up -d --build
+.PHONY: docker-compose-up
+
+docker-compose-down:
+	docker-compose -f docker-compose.yaml stop -t 20
+	docker-compose -f docker-compose.yaml down --remove-orphans
+.PHONY: docker-compose-down
+
+docker-compose-logs:
+	docker-compose -f docker-compose.yaml logs -f
+.PHONY: docker-compose-logs

README.md

@@ -20,3 +20,17 @@ This parameters are fetched from the [OpenWeatherMap API](https://openweathermap
 
 ## Commands
 It would be nice to accept commands from TUI to simulate deviations fixes. Something like `<parameter> <increase/decrease> <amount>`
+
+## Usage Instructions
+The repository includes a **Makefile** that encapsulates various commands used frequently in the project as targets. The targets are executed by invoking:
+
+* **make \<target\>**:
+The essential targets to start and stop the system are **docker-compose-up** and **docker-compose-down**, with the remaining targets being useful for debugging and troubleshooting.
+
+Available targets are:
+* **docker-compose-up**: Initializes the development environment (builds docker images for the server and client, initializes the network used by docker, etc.) and starts the containers of the applications that make up the project.
+* **docker-compose-down**: Performs a `docker-compose stop` to stop the containers associated with the compose and then performs a `docker-compose down` to destroy all resources associated with the initialized project. It is recommended to execute this command at the end of each run to prevent the host machine's disk from filling up.
+* **docker-compose-logs**: Allows viewing the current logs of the project. Use with `grep` to filter messages from a specific application within the compose.
+* **docker-image**: Builds the images to be used. This target is used by **docker-compose-up**, so it can be used to test new changes in the images before starting the project.
+
+Important Note: This service assumes a running instance of RabbitMQ and connects to it. Therefore, to run this service, it is necessary to first have the **measurements** service running. Please make sure to also check [measurements repository](https://github.com/Hanagotchi/measurements).

docker-compose.yaml

@@ -0,0 +1,16 @@
+version: '3.9'
+
+services:
+  main-app:
+    build:
+      context: .
+      dockerfile: Dockerfile
+    image: main-app
+    env_file:
+      - .env
+    networks:
+      - common_network
+
+networks:
+  common_network:
+    external: true

src/requirements.txt → requirements.txt

@@ -1,2 +1,3 @@
-requests
+pika
+requests
 python-dotenv

src/common/middleware.py

@@ -0,0 +1,52 @@
+import pika
+import os
+
+
+class Middleware:
+
+    def __init__(self):
+        rabbitmq_host = os.environ.get("RABBITMQ_HOST", "localhost")
+        self._connection = pika.BlockingConnection(
+            pika.ConnectionParameters(host=rabbitmq_host)
+        )
+        self._channel = self._connection.channel()
+        self._exit = False
+        self._remake = False
+
+    def create_queue(self, queue_name):
+        self._channel.queue_declare(queue=queue_name)
+
+    def _setup_message_consumption(self, queue_name, user_function):
+        self._channel.basic_consume(queue=queue_name,
+                                    on_message_callback=lambda channel,
+                                    method, properties, body:
+                                    (user_function(body),
+                                     channel.basic_ack
+                                     (delivery_tag=method.delivery_tag),
+                                     self._verify_connection_end()))
+        self._channel.start_consuming()
+
+    def _verify_connection_end(self):
+        if self._exit:
+            self._channel.close()
+            if self._remake:
+                self._exit = False
+                self._channel = self._connection.channel()
+
+    def finish(self, open_new_channel=False):
+        self._exit = True
+        self._remake = open_new_channel
+
+    # Work queue methods
+    def listen_on(self, queue_name, user_function):
+        self.create_queue(queue_name)
+        self._channel.basic_qos(prefetch_count=30)
+        self._setup_message_consumption(queue_name, user_function)
+
+    def send_message(self, queue_name, message):
+        self._channel.basic_publish(exchange='',
+                                    routing_key=queue_name,
+                                    body=message)
+
+    def __del__(self):
+        self._connection.close()

src/data_packet.py

@@ -4,9 +4,14 @@
 from os import environ
 from typing import Tuple
 from datetime import datetime
+import logging
 import math
+import uuid
 
 load_dotenv()
+logging.getLogger("urllib3").setLevel(logging.WARNING)
+UUID = str(uuid.uuid4()).replace("-", "")
+
 
 def fetch_temperature_and_humidity(location: str) -> Tuple[int, int]:
 
@@ -18,7 +23,7 @@ def fetch_temperature_and_humidity(location: str) -> Tuple[int, int]:
 
     if not res.ok:
         raise Exception(
-            "Could fetch temperature and humidity from weather API"
+            "Could not fetch temperature and humidity from weather API"
             )
 
     result = res.json()
@@ -65,14 +70,12 @@ def fetch_watering():
     return round(watering_simulator(x))
 
 
-def create_packet(temperature: int = None,
-                  humidity: int = None,
-                  light: int = None,
-                  watering: int = None):
+def create_packet(temperature: float = None, humidity: float = None,
+                  light: float = None, watering: float = None):
     '''
     Creates a data packet with simulated data, validating the data before that.
 
-    If all the parameters are empty, returns None. Else, return a dictionary 
+    If all the parameters are empty, returns None. Else, return a dictionary
     with the data.
 
 
@@ -81,44 +84,28 @@ def create_packet(temperature: int = None,
     percentages.
     - Light has to be positive or 0.
     '''
-
-    if temperature is None and humidity is None and light is None and watering is None:
-        return None
-
-    if humidity < 0 or humidity > 100:
-        raise Exception(
-            f"Humidity has to be between 0 and 100. Current value: {humidity}"
-            )
-
-    if watering < 0 or watering > 100:
-        raise Exception(
-            f"Watering has to be between 0 and 100. Current value: {watering}"
-            )
-
-    if light < 0:
-        raise Exception(
-            f"Light has to be positive or 0. Current value: {light}"
-            )
-
-def create_packet(temperature: float = None, humidity: float = None, light: float = None, watering: float = None):
-    if not (temperature and humidity and light and watering):
+    if not (temperature or humidity or light or watering):
         return None
 
     if humidity < 0 or humidity > 100:
-        raise Exception(f"Humidity has to be between 0 and 100. Current value: {humidity}")
+        raise Exception(f"Humidity has to be between 0 and 100."
+                        f"Current value: {humidity}")
 
     if watering < 0 or watering > 100:
-        raise Exception(f"Watering has to be between 0 and 100. Current value: {watering}")
+        raise Exception(f"Watering has to be between 0 and 100. "
+                        f"Current value: {watering}")
 
     if light < 0:
-        raise Exception(f"Light has to be positive or 0. Current value: {light}")
-
+        raise Exception(f"Light has to be positive or 0. "
+                        f"Current value: {light}")
 
     return {
         "temperature": temperature,
         "humidity": humidity,
         "light": light,
-        "watering": watering
+        "watering": watering,
+        "time_stamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
+        "id_device": UUID
     }
 
 
@@ -133,15 +120,16 @@ def generate_data(location="Pilar, AR") -> Tuple[int, int, int, int]:
     '''
 
     temperature, humidity = fetch_temperature_and_humidity(location)
+    # esto no va a producir cambios instantaneamente, pero es para probar
     light = fetch_solar_irradiation()
     watering = fetch_watering()
 
-    return temperature, humidity, light, watering 
+    return temperature, humidity, light, watering
 
 
 def data_has_changed(current, last_sent, deviations):
     '''
-    Compares the current packet and the last sent packet, 
+    Compares the current packet and the last sent packet,
     based in the deviations.
 
     If any parameter differs enough from the last sent packet, then the packet
@@ -162,14 +150,18 @@ def data_has_changed(current, last_sent, deviations):
     - deviations: {temperature: float, humidity: float, light: float,
     watering: float}
     '''
-        
+
     if not last_sent:
         return True
 
-    if parameter_has_changed(current["temperature"], last_sent["temperature"], deviations["temperature"])\
-            or parameter_has_changed(current["humidity"], last_sent["humidity"], deviations["humidity"])\
-            or parameter_has_changed(current["light"], last_sent["light"], deviations["light"])\
-            or parameter_has_changed(current["watering"], last_sent["watering"], deviations["watering"]):
+    if parameter_has_changed(current["temperature"], last_sent["temperature"],
+                             deviations["temperature"])\
+            or parameter_has_changed(current["humidity"], last_sent["humidity"],
+                                     deviations["humidity"])\
+            or parameter_has_changed(current["light"], last_sent["light"],
+                                     deviations["light"])\
+            or parameter_has_changed(current["watering"], last_sent["watering"],
+                                     deviations["watering"]):
         return True
 
     return False

src/main.py

@@ -12,26 +12,37 @@
 import time
 import random
 import logging
+import json
+import os
+from common.middleware import Middleware
+
 from data_packet import generate_data, create_packet, data_has_changed
 
 
 def simulate_packets(config):
+    middleware = Middleware()
+    queue_name = os.environ.get("QUEUE_NAME")
+    middleware.create_queue(queue_name)
     last_sent_packet = None
+    current_packet = None
     while True:
         try:
             temperature, humidity, light, watering = generate_data()
-            current_packet = create_packet(temperature, humidity, light, watering)
+            current_packet = create_packet(temperature, humidity, light,
+                                           watering)
 
-            if not current_packet or not data_has_changed(current_packet, last_sent_packet, config["deviations"]):
+            if not current_packet or not data_has_changed(
+                current_packet,
+                last_sent_packet,
+                config["deviations"]
+            ):
                 continue
-
-            # TODO: Send packet to the RabbitMQ queue
+            middleware.send_message(queue_name, json.dumps(current_packet))
             logging.info(f"Packet sent: {current_packet}")
             last_sent_packet = current_packet
 
-
         except Exception as err:
-            logging.warning(err)
+            logging.warning(f"{err}")
         finally:
             print(current_packet)
             time.sleep(config["packet_period"])
@@ -54,8 +65,27 @@ def read_config_file(path):
     }
 
 
-if __name__ == '__main__':
-    logging.basicConfig(level=logging.INFO)
+def main():
+    logging_level = os.environ.get("LOGGING_LEVEL")
+    initialize_log(logging_level)
     config = read_config_file("")
-
     simulate_packets(config)
+
+
+def initialize_log(logging_level):
+    """
+    Python custom logging initialization
+
+    Current timestamp is added to be able to identify in docker
+    compose logs the date when the log has arrived
+    """
+    logging.basicConfig(
+        format='%(asctime)s %(levelname)-8s %(message)s',
+        level=logging_level,
+        datefmt='%Y-%m-%d %H:%M:%S',
+    )
+    logging.getLogger("pika").setLevel(logging.WARNING)
+
+
+if __name__ == '__main__':
+    main()

tox.ini

@@ -0,0 +1,2 @@
+[flake8]
+max-line-length = 88