craftbeerpi4-pione/cbpi/controller/step_controller.py

353 lines
13 KiB
Python
Raw Permalink Normal View History

2021-01-22 23:25:20 +01:00
import asyncio
2021-03-07 22:11:25 +01:00
import cbpi
2021-02-16 20:37:51 +01:00
import copy
2021-01-22 23:25:20 +01:00
import json
2021-05-27 20:35:10 +02:00
import yaml
2021-01-22 23:25:20 +01:00
import logging
import os.path
2021-02-27 20:09:19 +01:00
from os import listdir
import os
2021-02-27 20:09:19 +01:00
from os.path import isfile, join
2021-02-16 20:37:51 +01:00
import shortuuid
from cbpi.api.dataclasses import NotificationAction, NotificationType, Props, Step
2021-02-16 20:37:51 +01:00
from tabulate import tabulate
2018-12-08 14:21:00 +01:00
2021-02-16 20:37:51 +01:00
from ..api.step import StepMove, StepResult, StepState
2018-12-09 22:20:33 +01:00
2019-07-27 21:08:19 +02:00
2021-01-22 23:25:20 +01:00
class StepController:
2018-11-29 21:59:08 +01:00
2018-12-03 22:16:03 +01:00
def __init__(self, cbpi):
self.cbpi = cbpi
2019-01-04 09:29:09 +01:00
self.logger = logging.getLogger(__name__)
self.path = self.cbpi.config_folder.get_file_path("step_data.json")
#self._loop = asyncio.get_event_loop()
2021-01-22 23:25:20 +01:00
self.basic_data = {}
self.step = None
self.types = {}
self.cbpi.app.on_cleanup.append(self.shutdown)
2019-07-27 21:08:19 +02:00
async def init(self):
2021-01-22 23:25:20 +01:00
logging.info("INIT STEP Controller")
self.load(startActive=True)
2021-02-16 20:37:51 +01:00
def create(self, data):
id = data.get("id")
name = data.get("name")
type = data.get("type")
status = StepState(data.get("status", "I"))
2021-03-07 23:52:20 +01:00
props = Props(data.get("props", {}))
2021-02-16 20:37:51 +01:00
try:
type_cfg = self.types.get(type)
clazz = type_cfg.get("class")
2021-03-07 23:52:20 +01:00
instance = clazz(self.cbpi, id, name, props, self.done)
2021-02-16 20:37:51 +01:00
except Exception as e:
logging.warning("Failed to create step instance %s - %s" % (id, e))
instance = None
step=Step(id, name, type=type, status=status, instance=instance, props=props )
return step
2021-02-16 20:37:51 +01:00
2021-01-22 23:25:20 +01:00
def load(self, startActive=False):
# create file if not exists
if os.path.exists(self.path) is False:
logging.warning("Missing step_data.json file. INIT empty file")
2021-01-22 23:25:20 +01:00
with open(self.path, "w") as file:
2021-02-27 20:09:19 +01:00
json.dump(dict(basic={}, steps=[]), file, indent=4, sort_keys=True)
2021-01-22 23:25:20 +01:00
#load from json file
try:
with open(self.path) as json_file:
data = json.load(json_file)
self.basic_data = data["basic"]
self.profile = data["steps"]
# Start step after start up
self.profile = list(map(lambda item: self.create(item), self.profile))
if startActive is True:
active_step = self.find_by_status("A")
if active_step is not None:
asyncio.get_event_loop().create_task(self.start_step(active_step))
#self._loop.create_task(self.start_step(active_step))
except:
logging.warning("Invalid step_data.json file - Creating empty file")
os.remove(self.path)
with open(self.path, "w") as file:
json.dump(dict(basic={"name": ""}, steps=[]), file, indent=4, sort_keys=True)
2021-01-22 23:25:20 +01:00
with open(self.path) as json_file:
data = json.load(json_file)
self.basic_data = data["basic"]
self.profile = data["steps"]
2021-02-27 20:09:19 +01:00
# Start step after start up
self.profile = list(map(lambda item: self.create(item), self.profile))
if startActive is True:
active_step = self.find_by_status("A")
if active_step is not None:
asyncio.get_event_loop().create_task(self.start_step(active_step))
#self._loop.create_task(self.start_step(active_step))
2021-01-22 23:25:20 +01:00
2021-02-16 20:37:51 +01:00
async def add(self, item: Step):
logging.debug("Add step")
2021-02-16 20:37:51 +01:00
item.id = shortuuid.uuid()
item.status = StepState.INITIAL
try:
type_cfg = self.types.get(item.type)
clazz = type_cfg.get("class")
item.instance = clazz(self.cbpi, item.id, item.name, item.props, self.done)
except Exception as e:
logging.warning("Failed to create step instance %s - %s " % (id, e))
item.instance = None
2021-01-22 23:25:20 +01:00
self.profile.append(item)
await self.save()
return item
2021-02-16 20:37:51 +01:00
async def update(self, item: Step):
2021-01-22 23:25:20 +01:00
logging.info("update step")
2021-02-16 20:37:51 +01:00
try:
type_cfg = self.types.get(item.type)
clazz = type_cfg.get("class")
item.instance = clazz(self.cbpi, item.id, item.name, item.props, self.done)
except Exception as e:
logging.warning("Failed to create step instance %s - %s " % (item.id, e))
item.instance = None
self.profile = list(map(lambda old: item if old.id == item.id else old, self.profile))
2021-01-22 23:25:20 +01:00
await self.save()
2021-02-16 20:37:51 +01:00
return item
2021-01-22 23:25:20 +01:00
async def save(self):
logging.debug("save profile")
2021-02-27 20:09:19 +01:00
data = dict(basic=self.basic_data, steps=list(map(lambda item: item.to_dict(), self.profile)))
2021-01-22 23:25:20 +01:00
with open(self.path, "w") as file:
json.dump(data, file, indent=4, sort_keys=True)
2021-02-10 07:38:55 +01:00
self.push_udpate()
2021-01-22 23:25:20 +01:00
async def start(self):
2021-02-16 20:37:51 +01:00
if self.find_by_status(StepState.ACTIVE) is not None:
2021-01-22 23:25:20 +01:00
logging.error("Steps already running")
return
2021-02-16 20:37:51 +01:00
step = self.find_by_status(StepState.STOP)
2021-01-22 23:25:20 +01:00
if step is not None:
logging.info("Resume step")
self.cbpi.push_update(topic="cbpi/notification", data=dict(type="info", title="Resume", message="Calling resume step"))
2021-01-22 23:25:20 +01:00
await self.start_step(step)
await self.save()
return
2021-02-16 20:37:51 +01:00
step = self.find_by_status(StepState.INITIAL)
2021-01-22 23:25:20 +01:00
if step is not None:
2021-02-16 20:37:51 +01:00
logging.info("Start Step")
self.cbpi.push_update(topic="cbpi/notification", data=dict(type="info", title="Start", message="Calling start step"))
self.push_udpate(complete=True)
2021-01-22 23:25:20 +01:00
await self.start_step(step)
await self.save()
return
2021-03-07 22:11:25 +01:00
self.cbpi.notify("Brewing Complete", "Now the yeast will take over",action=[NotificationAction("OK")])
self.cbpi.push_update(topic="cbpi/notification", data=dict(type="info", title="Brewing completed", message="Now the yeast will take over"))
2021-01-22 23:25:20 +01:00
logging.info("BREWING COMPLETE")
2021-02-27 20:09:19 +01:00
async def previous(self):
logging.info("Trigger Previous")
2021-02-27 20:09:19 +01:00
2021-01-22 23:25:20 +01:00
async def next(self):
logging.info("Trigger Next")
#print("\n\n\n\n")
#print(self.profile)
#print("\n\n\n\n")
2021-02-16 20:37:51 +01:00
step = self.find_by_status(StepState.ACTIVE)
2021-01-22 23:25:20 +01:00
if step is not None:
2021-02-16 20:37:51 +01:00
if step.instance is not None:
await step.instance.next()
step = self.find_by_status(StepState.STOP)
2021-02-10 07:38:55 +01:00
if step is not None:
2021-02-16 20:37:51 +01:00
if step.instance is not None:
step.status = StepState.DONE
2021-02-10 07:38:55 +01:00
await self.save()
await self.start()
2019-01-04 09:29:09 +01:00
else:
2021-01-22 23:25:20 +01:00
logging.info("No Step is running")
async def resume(self):
step = self.find_by_status("P")
if step is not None:
instance = step.get("instance")
if instance is not None:
await self.start_step(step)
else:
logging.info("Nothing to resume")
async def stop(self):
2021-02-16 20:37:51 +01:00
step = self.find_by_status(StepState.ACTIVE)
if step != None:
2021-01-22 23:25:20 +01:00
logging.info("CALLING STOP STEP")
2021-02-16 20:37:51 +01:00
try:
await step.instance.stop()
self.cbpi.push_update(topic="cbpi/notification", data=dict(type="info", title="Pause", message="Calling paue step"))
2021-02-16 20:37:51 +01:00
step.status = StepState.STOP
2021-02-16 20:37:51 +01:00
await self.save()
except Exception as e:
logging.error("Failed to stop step - Id: %s" % step.id)
2021-01-22 23:25:20 +01:00
async def reset_all(self):
2021-02-16 20:37:51 +01:00
if self.find_by_status(StepState.ACTIVE) is not None:
2021-01-22 23:25:20 +01:00
logging.error("Please stop before reset")
return
2021-02-16 20:37:51 +01:00
2021-01-22 23:25:20 +01:00
for item in self.profile:
2021-02-16 20:37:51 +01:00
logging.info("Reset %s" % item)
item.status = StepState.INITIAL
try:
await item.instance.reset()
self.cbpi.push_update(topic="cbpi/notification", data=dict(type="info", title="Stop", message="Calling stop step"))
2021-02-16 20:37:51 +01:00
except:
logging.warning("No Step Instance - Id: %s", item.id)
await self.save()
2021-02-10 07:38:55 +01:00
self.push_udpate()
2021-01-22 23:25:20 +01:00
2021-02-16 20:37:51 +01:00
def get_types(self):
2021-01-22 23:25:20 +01:00
result = {}
for key, value in self.types.items():
#if "ferment" not in str(value.get("class")).lower():
result[key] = dict(name=value.get("name"), properties=value.get("properties"), actions=value.get("actions"))
2021-01-22 23:25:20 +01:00
return result
2019-01-04 09:29:09 +01:00
2021-01-22 23:25:20 +01:00
def get_state(self):
2021-02-27 20:09:19 +01:00
return {"basic": self.basic_data, "steps": list(map(lambda item: item.to_dict(), self.profile)), "types":self.get_types()}
2019-07-27 21:08:19 +02:00
2021-02-16 20:37:51 +01:00
async def move(self, id, direction: StepMove):
2021-01-22 23:25:20 +01:00
index = self.get_index_by_id(id)
if direction not in [-1, 1]:
self.logger.error("Cant move. Direction 1 and -1 allowed")
2019-07-27 21:08:19 +02:00
return
2021-01-22 23:25:20 +01:00
self.profile[index], self.profile[index+direction] = self.profile[index+direction], self.profile[index]
await self.save()
2021-02-10 07:38:55 +01:00
self.push_udpate()
2021-01-22 23:25:20 +01:00
async def delete(self, id):
step = self.find_by_id(id)
2021-02-16 20:37:51 +01:00
if step is None:
logging.error("Cant find step - Nothing deleted - Id: %s", id)
return
if step.status == StepState.ACTIVE:
2021-01-22 23:25:20 +01:00
logging.error("Cant delete active Step %s", id)
return
2021-02-16 20:37:51 +01:00
self.profile = list(filter(lambda item: item.id != id, self.profile))
2021-01-22 23:25:20 +01:00
await self.save()
2021-02-16 20:37:51 +01:00
2021-02-27 20:09:19 +01:00
async def shutdown(self, app=None):
2022-04-02 16:20:48 +02:00
logging.info("Mash Profile Shutdown")
2021-01-22 23:25:20 +01:00
for p in self.profile:
2021-02-16 20:37:51 +01:00
instance = p.instance
2021-01-22 23:25:20 +01:00
# Stopping all running task
if hasattr(instance, "task") and instance.task != None and instance.task.done() is False:
2021-01-22 23:25:20 +01:00
logging.info("Stop Step")
2021-02-10 07:38:55 +01:00
await instance.stop()
2021-01-22 23:25:20 +01:00
await instance.task
await self.save()
2021-02-27 20:09:19 +01:00
self.push_udpate()
2021-01-22 23:25:20 +01:00
2021-02-16 20:37:51 +01:00
def done(self, step, result):
if result == StepResult.NEXT:
step_current = self.find_by_id(step.id)
step_current.status = StepState.DONE
async def wrapper():
await self.save()
await self.start()
asyncio.create_task(wrapper())
2019-01-04 09:29:09 +01:00
2021-01-22 23:25:20 +01:00
def find_by_status(self, status):
2021-02-16 20:37:51 +01:00
return next((item for item in self.profile if item.status == status), None)
2021-01-22 23:25:20 +01:00
def find_by_id(self, id):
2021-02-16 20:37:51 +01:00
return next((item for item in self.profile if item.id == id), None)
2021-01-22 23:25:20 +01:00
def get_index_by_id(self, id):
2021-02-16 20:37:51 +01:00
return next((i for i, item in enumerate(self.profile) if item.id == id), None)
2021-01-22 23:25:20 +01:00
2021-02-27 20:09:19 +01:00
def push_udpate(self, complete=False):
if complete is True:
self.cbpi.ws.send(dict(topic="mash_profile_update", data=self.get_state()))
for item in self.profile:
self.cbpi.push_update(topic="cbpi/stepupdate/{}".format(item.id), data=(item.to_dict()))
2021-02-27 20:09:19 +01:00
else:
self.cbpi.ws.send(dict(topic="step_update", data=list(map(lambda item: item.to_dict(), self.profile))))
step = self.find_by_status(StepState.ACTIVE)
if step != None:
self.cbpi.push_update(topic="cbpi/stepupdate/{}".format(step.id), data=(step.to_dict()))
2021-03-14 11:52:46 +01:00
2021-01-22 23:25:20 +01:00
async def start_step(self,step):
2021-02-10 07:38:55 +01:00
try:
2021-02-16 20:37:51 +01:00
logging.info("Try to start step %s" % step)
await step.instance.start()
step.status = StepState.ACTIVE
2021-02-10 07:38:55 +01:00
except Exception as e:
self.cbpi.notify("Error", "Can't start step. Please check step in Mash Profile", NotificationType.ERROR)
2022-04-02 16:27:09 +02:00
logging.error("Failed to start step %s" % step)
2021-01-22 23:25:20 +01:00
async def save_basic(self, data):
logging.info("SAVE Basic Data")
self.basic_data = {**self.basic_data, **data,}
await self.save()
2021-02-10 07:38:55 +01:00
self.push_udpate()
async def call_action(self, id, action, parameter) -> None:
logging.info("Step Controller - call all Action {} {}".format(id, action))
try:
item = self.find_by_id(id)
await item.instance.__getattribute__(action)(**parameter)
except Exception as e:
2022-04-02 16:20:48 +02:00
logging.error("Step Controller -Failed to call action on {} {} {}".format(id, action, e))
2021-02-27 20:09:19 +01:00
async def load_recipe(self, data):
try:
await self.shutdown()
except:
pass
def add_runtime_data(item):
item["status"] = "I"
item["id"] = shortuuid.uuid()
list(map(lambda item: add_runtime_data(item), data.get("steps")))
with open(self.path, "w") as file:
json.dump(data, file, indent=4, sort_keys=True)
self.load()
self.push_udpate(complete=True)
async def clear(self):
try:
await self.shutdown()
except:
pass
data = dict(basic=dict(), steps=[])
with open(self.path, "w") as file:
json.dump(data, file, indent=4, sort_keys=True)
self.load()
self.push_udpate(complete=True)
2021-05-27 20:35:10 +02:00
async def savetobook(self):
name = shortuuid.uuid()
path = os.path.join(self.cbpi.config_folder.get_file_path("recipes"), "{}.yaml".format(name))
2021-05-27 20:35:10 +02:00
data = dict(basic=self.basic_data, steps=list(map(lambda item: item.to_dict(), self.profile)))
with open(path, "w") as file:
yaml.dump(data, file)
self.push_udpate()