denis
170 lines
5.1 KiB
Python
170 lines
5.1 KiB
Python
'''
|
|
Created on 23 janv. 2026
|
|
|
|
@author: denis
|
|
'''
|
|
# process.py
|
|
#from django.utils.translation import gettext_lazy as _
|
|
import os
|
|
os.environ['BOARD_DEVICE']="RPI"
|
|
import json
|
|
from threading import Thread, Event
|
|
from asgiref.sync import async_to_sync #, sync_to_async
|
|
from channels.layers import get_channel_layer
|
|
|
|
from django.conf import settings
|
|
from django.db import transaction
|
|
|
|
from celery import Task
|
|
from celery.exceptions import Ignore
|
|
from celery.utils.log import get_task_logger
|
|
from redis import Redis
|
|
|
|
from modules import gpiolib as gpio, devices as dev
|
|
from . import models
|
|
|
|
logger = get_task_logger(__name__)
|
|
redisDB = Redis(host=settings.REDIS_HOST, port=settings.REDIS_PORT, db=0, decode_responses=True)
|
|
|
|
class RelayManager:
|
|
|
|
def __init__(self):
|
|
self.shift_out_register = gpio.ShiftOutRegister74HC595(
|
|
latch_pin=settings.CTL_LATCH_PIN,
|
|
clock_pin=settings.CTL_CLOCK_PIN,
|
|
data_pin=settings.CTL_DATA_PIN,
|
|
number=settings.CTL_NUMBER,
|
|
)
|
|
|
|
def create_relay(self, device, publish=None):
|
|
return dev.Relay(self.shift_out_register, device.to_register(), publish=publish)
|
|
|
|
|
|
relay_manager = RelayManager()
|
|
|
|
|
|
class RelayProcess(Task):
|
|
|
|
def __init__(self, uuid):
|
|
super().__init__()
|
|
self.channel_layer = get_channel_layer()
|
|
self.uuid = uuid
|
|
self.group = f'relay_{uuid}'
|
|
self.stop_event = Event()
|
|
self.heartbeat = 10
|
|
self.relay = None
|
|
self.device = None
|
|
|
|
def __call__(self, uuid, *args, **kwargs):
|
|
self.uuid = uuid
|
|
return self.start(*args, **kwargs)
|
|
|
|
|
|
def redis_pub(self, state, **msg):
|
|
async_to_sync(self.channel_layer.group_send)(
|
|
self.group, {
|
|
"type": 'relay.message',
|
|
"text": { 'topic': state, **msg }
|
|
}
|
|
)
|
|
self.mqtt_pub(state, **msg)
|
|
|
|
|
|
def mqtt_pub(self, state, **payload):
|
|
topic = f"relay/{self.uuid}/{state}"
|
|
payload.pop("uuid", None)
|
|
redisDB.publish('mqtt_pub', json.dumps({"type": "mqtt", "topic": topic, "payload": payload}))
|
|
#logger.info(f"mqtt_pub {topic}: {payload}")
|
|
|
|
|
|
def start(self, *args, **kwargs):
|
|
try:
|
|
with transaction.atomic():
|
|
self.device = models.Device.objects.select_for_update().get(uuid=self.uuid)
|
|
self.stop_event.clear()
|
|
self.start_services()
|
|
|
|
except Exception as e:
|
|
logger.error(f"Relay error {self.uuid}: {e}")
|
|
raise Ignore()
|
|
|
|
|
|
def stop(self):
|
|
try:
|
|
self.redis_pub('state', stop="done")
|
|
self.stop_event.set()
|
|
logger.info(f"Relay {self.uuid} stopped.")
|
|
Event().wait(1.0)
|
|
finally:
|
|
self.stop_event.set()
|
|
gpio.IO_CLEANUP()
|
|
|
|
|
|
def start_services(self):
|
|
Thread(target=self._listen_to_redis, daemon=True).start()
|
|
Thread(target=self._relay_process, daemon=True).start()
|
|
|
|
|
|
def _listen_to_redis(self):
|
|
try:
|
|
logger.info(f"RelayProcess {self.group}: listen via redisDB")
|
|
pubsub = redisDB.pubsub()
|
|
pubsub.subscribe(f"{self.group}", "relay_broadcast", "mqtt_message")
|
|
try:
|
|
for message in pubsub.listen():
|
|
#logger.info(f"{message}")
|
|
if self.stop_event.is_set():
|
|
break
|
|
|
|
cmd = json.loads(str(message.get('data')))
|
|
if not isinstance(cmd, dict):
|
|
continue
|
|
|
|
if cmd["type"]=="relay" or cmd["type"]=="mqtt":
|
|
topic = cmd["topic"]
|
|
|
|
if topic == "start":
|
|
duration = cmd.get("value", 0)
|
|
self.relay.relay_start(duration=duration)
|
|
|
|
elif topic == "stop":
|
|
self.relay.relay_stop()
|
|
|
|
elif topic in ["status", "init"]:
|
|
self.relay.relay_status()
|
|
|
|
elif topic == "halt":
|
|
self.relay.relay_halt()
|
|
|
|
elif topic == "restart":
|
|
self.relay.relay_restart()
|
|
except Exception as e:
|
|
logger.error(f'listen_to_redis: {e}')
|
|
finally:
|
|
pubsub.unsubscribe()
|
|
pubsub.close()
|
|
|
|
|
|
def _relay_process(self):
|
|
try:
|
|
logger.info(f"RelayProcess Start {self.uuid}")
|
|
self.relay = relay_manager.create_relay(self.device, publish=self.redis_pub)
|
|
self.relay.relay_stop()
|
|
|
|
while not self.stop_event.is_set():
|
|
try:
|
|
Event().wait(self.device.heartbeat)
|
|
self.relay.relay_status()
|
|
|
|
except Exception as e:
|
|
logger.error(f'====> {e}')
|
|
Event().wait(1.0)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Relay {self.uuid}: {e}")
|
|
finally:
|
|
|
|
logger.info(f"End of relay process: {self.uuid}")
|
|
self.redis_pub('state', relay="stop")
|
|
|