Cucumber/GreenhouseManager/manager.py

import os
from kafka import KafkaProducer, KafkaConsumer
import kafka
import socket
from json import dumps, loads
from flask import Flask, request
import time
from enum import Enum
import threading


app = Flask(__name__)

def start_manager():
    return

class Manager:
    def __init__(self, _id: int, moisture: float = 0, target_moisture: float = 30, temp: float = 20, target_temp: float = 30, 
                 isAutoOn: bool = False, valve_state: str = "closed", heater_state: str = "off"):
        KAFKA_URL = os.environ.get('KAFKA_URL')
        print("KAFKA_URL=", KAFKA_URL)
        self._id = _id
        self.moisture = moisture
        self.target_moisture = target_moisture
        self.temp = temp
        self.temp = target_temp
        self.isAutoOn = isAutoOn
        self.valve_state = valve_state
        self.heater_state = heater_state

        self.controllerConsumer = KafkaConsumer(
            'commands',
            bootstrap_servers=[KAFKA_URL],
            auto_offset_reset='earliest',
            enable_auto_commit=True,
            consumer_timeout_ms=2000,
            group_id=f'manager{self._id}',
            value_deserializer=lambda x: loads(x.decode('utf-8'))
        )
        self.controllerConsumerResponse = KafkaProducer(
            bootstrap_servers=[KAFKA_URL],
            client_id=f'manager{self._id}_producer',
            value_serializer=lambda v: dumps(v).encode('utf-8')
        )

    def toggle_device(self, device, request_id, greenhouse_id):

        if device == 'valve':

            if self.valve_state == 'closed':
                self.valve_state = 'open'
                print("Valve opened")
            else:
                self.valve_state = 'closed'
                print("Valve closed")

        elif device == 'heater':

            if self.heater_state == 'off':
                self.heater_state = 'on'
                print("Heater turned on")
            else:
                self.heater_state = 'off'
                print("Heater turned off")

        self.send_status(request_id, greenhouse_id)

    def send_status(self, request_id, greenhouse_id):

        data = {
            'request_id': request_id,
            'greenhouse_id': greenhouse_id,
            'moisture': self.moisture,
            'target_moisture': self.target_moisture,
            'valveStatus': self.valve_state,
            'temp': self.temp,
            'target_temp': self.target_temp,
            'heaterStatus': self.heater_state,
            'isAutoOn': self.isAutoOn
        }
        self.sendDataCommand(status)
        print("Updating info...\n")

    def sendDataCommand(self, message):
        print("sending data...")

        url = "https://localhost:7113/api/webhook"
        headers = {
            "Content-Type": "application/json"
        }

        response = requests.post(url, json=message, headers=headers)

        # Check status code and content
        print("Status Code", response.status_code)
        print("Response Body", response.json())

    def getCommand(self):
        messages = self.controllerConsumer.poll(timeout_ms=1000)

        # Проверяем, есть ли сообщения

        for tp, msgs in messages.items():
            for message in msgs:
                print(f"Manager {self._id} received message: ")
                print(message.value)
                self.request_id = message.value['request_id']
                self.greenhouse_id = message.value['greenhouse_id']
                self.command = message.value['command']
                if message.value.get('target_moisture'):
                    self.target_moisture = message.value['target_moisture']
                if message.value.get('target_temp'):
                    self.target_temp = message.value['target_temp']
                self.toggle_device(self.command, self.request_id, self.greenhouse_id)

t1 = threading.Thread(target=start_manager)
manager1 = Manager(_id=1)
managers = [manager1]

if __name__ == "__main__":
    threading.Thread(target=lambda: app.run(host="0.0.0.0", port=20002, debug=True, use_reloader=False)).start()