131 lines
4.1 KiB
Python
131 lines
4.1 KiB
Python
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, temp: float = 20, isAutoOn: bool = False, valve_state: str = "closed",
|
|
heater_state: str = "off"):
|
|
self._id = _id
|
|
self.moisture = moisture
|
|
self.temp = temp
|
|
self.isAutoOn = isAutoOn
|
|
self.valve_state = valve_state
|
|
self.heater_state = heater_state
|
|
|
|
self.dataPublisher = KafkaProducer(
|
|
bootstrap_servers=['localhost:9092'],
|
|
client_id=f'manager{self._id}_producer',
|
|
value_serializer=lambda v: dumps(v).encode('utf-8')
|
|
)
|
|
|
|
self.controllerConsumer = KafkaConsumer(
|
|
'commands',
|
|
bootstrap_servers=['localhost:9092'],
|
|
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=['localhost:9092'],
|
|
client_id=f'manager{self._id}_producer',
|
|
value_serializer=lambda v: dumps(v).encode('utf-8')
|
|
)
|
|
|
|
def sendData(self):
|
|
print("sending data...")
|
|
message = {
|
|
'id': self._id,
|
|
'moisture': self.moisture,
|
|
'temp': self.temp,
|
|
'valveStatus': str(self.valve_state),
|
|
'heaterStatus': str(self.heater_state),
|
|
'isAutoOn': self.isAutoOn
|
|
}
|
|
|
|
print(message)
|
|
self.dataPublisher.send('data', message)
|
|
|
|
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):
|
|
|
|
status = {
|
|
'request_id': request_id,
|
|
'greenhouse_id': greenhouse_id,
|
|
'valve_state': self.valve_state,
|
|
'heater_state': self.heater_state
|
|
}
|
|
self.sendDataCommand(status)
|
|
print("Updating info...\n")
|
|
|
|
def sendDataCommand(self, message):
|
|
print("sending data...")
|
|
|
|
self.dataPublisher.send('response', message)
|
|
|
|
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']
|
|
self.toggle_device(self.command, self.request_id, self.greenhouse_id)
|
|
|
|
@app.route(f'/webhook', methods=['POST'])
|
|
def webhook():
|
|
for manager in managers:
|
|
if request.args.get('id') == manager._id and request.method == 'POST':
|
|
print("Data received from Webhook is", request.json)
|
|
|
|
body = request.json
|
|
for key, value in body.items():
|
|
setattr(manager, key, value)
|
|
|
|
return f"Webhook received for manager {manager._id}"
|
|
return "Webhook ignored"
|
|
|
|
manager1 = Manager(_id=1)
|
|
managers = [manager1]
|
|
|
|
t1 = threading.Thread(target=start_manager)
|
|
|
|
if __name__ == "__main__":
|
|
app.run(host="0.0.0.0", port=20002) |