from abc import abstractmethod import asyncio import cbpi import copy import json import yaml import logging import os.path import pathlib from os import listdir from os.path import isfile, join import shortuuid from cbpi.api.dataclasses import Fermenter, FermenterStep, Props, Step from cbpi.controller.basic_controller2 import BasicController from tabulate import tabulate import sys, os from ..api.step import CBPiStep, StepMove, StepResult, StepState, CBPiFermentationStep class FermentStep: def __init__(self, cbpi, step, on_done) -> None: self.cbpi = cbpi self.logger = logging.getLogger(__name__) self.step = step self.props = step.props self._done_callback = on_done self.task = None self.summary = "" def _done(self, task): if self._done_callback is not None: try: result = task.result() self._done_callback(self, result) except Exception as e: self.logger.error(e) @abstractmethod async def run(self): while self.running: logging.info(self.step) await asyncio.sleep(1) pass async def _run(self): try: await self.step.instance.on_start() await self.step.instance.run() #await self.on_start() #await self.run() self.cancel_reason = StepResult.DONE except asyncio.CancelledError as e: pass finally: await self.on_stop() return self.cancel_reason async def start(self): self.logger.info("Start {}".format(self.step.name)) self.running = True self.task = asyncio.create_task(self._run()) self.task.add_done_callback(self._done) async def next(self): self.running = False self.cancel_reason = StepResult.NEXT self.task.cancel() await self.task async def stop(self): try: self.running = False if self.task is not None and self.task.done() is False: self.logger.info("Stopping Task") self.cancel_reason = StepResult.STOP self.task.cancel() await self.task except Exception as e: self.logger.error(e) async def on_start(self): #self.props.hello = "WOOHOo" pass async def on_stop(self): pass class FermentationController: def __init__(self, cbpi): self.update_key = "fermenterupdate" self.cbpi = cbpi self.logger = logging.getLogger(__name__) self.path = os.path.join(".", 'config', "fermenter_data.json") self._loop = asyncio.get_event_loop() self.data = [] self.types = {} self.steptypes = {} self.cbpi.app.on_cleanup.append(self.shutdown) async def init(self): logging.info("INIT Fermentation Controller") self.check_fermenter_file() await self.load() pass def check_fermenter_file(self): if os.path.exists(os.path.join(".", 'config', "fermenter_data.json")) is False: logging.info("INIT fermenter_data.json file") data = { "data": [ ] } destfile = os.path.join(".", 'config', "fermenter_data.json") json.dump(data,open(destfile,'w'),indent=4, sort_keys=True) pathlib.Path(os.path.join(".", 'config/fermenterrecipes')).mkdir(parents=True, exist_ok=True) async def shutdown(self, app=None, fermenterid=None): self.save() if (fermenterid == None): for fermenter in self.data: self.logger.info("Shutdown {}".format(fermenter.name)) for step in fermenter.steps: try: self.logger.info("Stop {}".format(step.name)) await step.instance.stop() except Exception as e: self.logger.error(e) else: fermenter = self._find_by_id(fermenterid) self.logger.info("Shutdown {}".format(fermenter.name)) for step in fermenter.steps: try: self.logger.info("Stop {}".format(step.name)) await step.instance.stop() except Exception as e: self.logger.error(e) async def load(self): # if os.path.exists(self.path) is False: # with open(self.path, "w") as file: # json.dump(dict(basic={}, steps=[]), file, indent=4, sort_keys=True) with open(self.path) as json_file: data = json.load(json_file) for i in data["data"]: self.data.append(self._create(i)) def _create_step(self, fermenter, item): id = item.get("id") name = item.get("name") props = Props(item.get("props")) status = StepState(item.get("status", "I")) type = item.get("type") try: type_cfg = self.steptypes.get(type) clazz = type_cfg.get("class") instance = clazz(self.cbpi, fermenter, item, props, self._done) except Exception as e: logging.warning("Failed to create step instance %s - %s" % (id, e)) instance = None step = FermenterStep(id=id, name=name, fermenter=fermenter, props=props, type=type, status=status, instance=instance) return step def _done(self, step_instance, result, fermenter): step_instance.step["status"] = "D" self.save() if result == StepResult.NEXT: asyncio.create_task(self.start(fermenter)) def _create(self, data): try: id = data.get("id") name = data.get("name") sensor = data.get("sensor") heater = data.get("heater") cooler = data.get("cooler") logictype = data.get("type") temp = data.get("target_temp") brewname = data.get("brewname") description = data.get("description") props = Props(data.get("props", {})) fermenter = Fermenter(id, name, sensor, heater, cooler, brewname, description, props, temp, logictype) fermenter.steps = list(map(lambda item: self._create_step(fermenter, item), data.get("steps", []))) self.push_update() return fermenter except: return def _find_by_id(self, id): return next((item for item in self.data if item.id == id), None) async def get_all(self): return list(map(lambda x: x.to_dict(), self.data)) def get_types(self): result = {} for key, value in self.types.items(): result[key] = dict(name=value.get("name"), properties=value.get("properties"), actions=value.get("actions")) return result def get_steptypes(self): result = {} for key, value in self.steptypes.items(): result[key] = dict(name=value.get("name"), properties=value.get("properties"), actions=value.get("actions")) return result def get_state(self): if self.data == []: #logging.info(self.data) pass return {"data": list(map(lambda x: x.to_dict(), self.data)), "types":self.get_types(), "steptypes":self.get_steptypes()} def get_step_state(self, fermenterid=None): if self.data == []: pass fermentersteps=[] steplist=list(map(lambda x: x.to_dict(), self.data)) for fermenter in steplist: if fermenterid == fermenter.get("id"): fermentersteps={"id": fermenter.get("id"), "steps": fermenter.get("steps")} return fermentersteps def get_fermenter_steps(self): if self.data == []: #logging.info(self.data) pass fermentersteps=[] steplist=list(map(lambda x: x.to_dict(), self.data)) for fermenter in steplist: fermenterstep={"id": fermenter.get("id"), "steps": fermenter.get("steps")} fermentersteps.append(fermenterstep) return fermentersteps async def find_step_by_id(self, id): actionstep = None for item in self.data: step = self._find_step_by_id(item.steps, id) if step is not None: actionstep=step return actionstep async def get(self, id: str ): return self._find_by_id(id) async def create(self, data: Fermenter ): data.id = shortuuid.uuid() self.data.append(data) self.save() self.push_update() return data async def update(self, item: Fermenter ): def _update(old_item: Fermenter, item: Fermenter): old_item.name = item.name old_item.sensor = item.sensor old_item.heater = item.heater old_item.cooler = item.cooler old_item.type = item.type old_item.brewname = item.brewname old_item.description = item.description old_item.props = item.props old_item.target_temp = item.target_temp return old_item self.data = list(map(lambda old: _update(old, item) if old.id == item.id else old, self.data)) self.save() self.push_update() return item async def set_target_temp(self, id: str, target_temp): try: item = self._find_by_id(id) logging.info(item.target_temp) if item: item.target_temp = target_temp self.save() self.push_update() except Exception as e: logging.error("Failed to set Target Temp {} {}".format(id, e)) async def delete(self, id: str ): item = self._find_by_id(id) self.data = list(filter(lambda item: item.id != id, self.data)) self.save() self.push_update() def save(self): data = dict(data=list(map(lambda item: item.to_dict(), self.data))) with open(self.path, "w") as file: json.dump(data, file, indent=4, sort_keys=True) async def create_step(self, id, item): try: stepid = shortuuid.uuid() props = item.get("props") status = StepState("I") type = item.get("type") name = item.get("name") props = Props(item.get("props")) fermenter = self._find_by_id(id) try: type_cfg = self.steptypes.get(type) clazz = type_cfg.get("class") instance = clazz(self.cbpi, fermenter, item, props, self._done) except Exception as e: logging.warning("Failed to create step instance %s - %s" % (id, e)) instance = None step = FermenterStep(id=stepid, name=name, fermenter=fermenter, props=props, type=type, status=status, instance=instance) fermenter.steps.append(step) self.save() self.push_update("fermenterstepupdate") return step except Exception as e: self.logger.error(e) async def update_step(self, id, item): fermenter = self._find_by_id(id) stepid = item.get("id") props = item.get("props") status = StepState("I") type = item.get("type") #logging.info(type) name = item.get("name") props = Props(item.get("props")) logging.info("update step") try: type_cfg = self.steptypes.get(type) #logging.info(type_cfg) clazz = type_cfg.get("class") #logging.info(clazz) instance = clazz(self.cbpi, fermenter, item, props, self._done) except Exception as e: logging.warning("Failed to create step instance %s - %s " % (item.id, e)) instance = None step = FermenterStep(id=stepid, name=name, fermenter=fermenter, props=props, type=type, status=status, instance=instance) #logging.info(step) #logging.info(fermenter.steps) try: fermenter.steps = list(map(lambda old: step if old.id == step.id else old, fermenter.steps)) except Exception as e: logging.info(e) #logging.info(fermenter.steps) self.save() #logging.info("SAVEUPDATE") self.push_update("fermenterstepupdate") async def delete_step(self, id, stepid): item = self._find_by_id(id) # might require later check if step is active item.steps = list(filter(lambda item: item.id != stepid, item.steps)) self.save() self.push_update("fermenterstepupdate") async def clearsteps(self, id): item = self._find_by_id(id) # might require later check if step is active item.steps = [] self.save() self.push_update("fermenterstepupdate") def _find_by_status(self, data, status): return next((item for item in data if item.status == status), None) def _find_step_by_id(self, data, id): return next((item for item in data if item.id == id), None) async def start(self, id): self.logger.info("Start {}".format(id)) try: item = self._find_by_id(id) step = self._find_by_status(item.steps, StepState.ACTIVE) if step is not None: logging.error("Steps already running") return step = self._find_by_status(item.steps, StepState.STOP) if step is not None: await step.instance.start() logging.info("Restarting step {}".format(step.name)) step.status = StepState.ACTIVE self.save() self.push_update("fermenterstepupdate") return step = self._find_by_status(item.steps, StepState.INITIAL) logging.info(step) if step is None: self.logger.info("No futher step to start") else: await step.instance.start() logging.info("Starting step {}".format(step.name)) step.status = StepState.ACTIVE self.save() self.push_update("fermenterstepupdate") except Exception as e: self.logger.error(e) async def stop(self, id): try: item = self._find_by_id(id) step = self._find_by_status(item.steps, StepState.ACTIVE) logging.info(step) logging.info(step.status) if step != None: logging.info("CALLING STOP STEP") try: await step.instance.stop() step.status = StepState.STOP self.save() except Exception as e: logging.error("Failed to stop fermenterstep - Id: %s" % step.id) self.push_update() self.push_update("fermenterstepupdate") except Exception as e: self.logger.error(e) async def start_logic(self, id): try: item = self._find_by_id(id) logging.info("{} Start Id {} ".format(item.name, id)) if item.instance is not None and item.instance.running is True: logging.warning("{} already running {}".format(item.name, id)) return if item.type is None: logging.warning("{} No Type {}".format(item.name, id)) return clazz = self.types[item.type]["class"] item.instance = clazz(self.cbpi, item.id, item.props) await item.instance.start() item.instance.running = True item.instance.task = self._loop.create_task(item.instance._run()) logging.info("{} started {}".format(item.name, id)) except Exception as e: logging.error("{} Cant start {} - {}".format(item.name, id, e)) async def toggle(self, id): try: item = self._find_by_id(id) if item.instance is None or item.instance.state == False: await self.start_logic(id) else: await item.instance.stop() self.push_update() except Exception as e: logging.error("Failed to switch on FermenterLogic {} {}".format(id, e)) async def next(self, id): self.logger.info("Next {} ".format(id)) try: item = self._find_by_id(id) logging.info(item) step = self._find_by_status(item.steps, StepState.ACTIVE) logging.info(step) if step is not None: if step.instance is not None: step.status = StepState.DONE await step.instance.next() step = self._find_by_status(item.steps, StepState.STOP) logging.info(step) if step is not None: if step.instance is not None: logging.info(step) step.status = StepState.DONE logging.info(step) self.save() await self.start(id) else: logging.info("No Step is running") self.push_update("fermenterstepupdate") except Exception as e: self.logger.error(e) async def reset(self, id): self.logger.info("Reset") try: item = self._find_by_id(id) for step in item.steps: self.logger.info("Stopping Step {} {}".format(step.name, step.id)) try: await step.instance.stop() step.status = StepState.INITIAL except Exception as e: self.logger.error(e) self.save() self.push_update("fermenterstepupdate") except Exception as e: self.logger.error(e) async def move_step(self, fermenter_id, step_id, direction): try: fermenter = self._find_by_id(fermenter_id) index = next((i for i, item in enumerate(fermenter.steps) if item.id == step_id), None) if index == None: return if index == 0 and direction == -1: return if index == len(fermenter.steps)-1 and direction == 1: return fermenter.steps[index], fermenter.steps[index+direction] = fermenter.steps[index+direction], fermenter.steps[index] self.save() self.push_update("fermenterstepupdate") except Exception as e: self.logger.error(e) def push_update(self, key="fermenterupdate"): if key == self.update_key: self.cbpi.ws.send(dict(topic=key, data=list(map(lambda item: item.to_dict(), self.data)))) #self.cbpi.push_update("cbpi/{}".format(self.update_key), list(map(lambda item: item.to_dict(), self.data))) for item in self.data: self.cbpi.push_update("cbpi/{}/{}".format(self.update_key,item.id), item.to_dict()) pass else: #logging.info("FERMENTERSTEPUPDATE {}".format(key)) fermentersteps=self.get_fermenter_steps() self.cbpi.ws.send(dict(topic=key, data=fermentersteps)) async def call_action(self, id, action, parameter) -> None: logging.info("FermenterStep Controller - call Action {} {}".format(id, action)) try: item = await self.find_step_by_id(id) logging.info(item) await item.instance.__getattribute__(action)(**parameter) except Exception as e: logging.error("FermenterStep Controller - Failed to call action on {} {} {}".format(id, action, e)) # todo: Sensors may need to be removed when saving the recipe -> need to be replaced when assinging later to Fermenter with 'fermenter.sensor' async def savetobook(self, fermenterid): name = shortuuid.uuid() path = os.path.join(".", 'config', "fermenterrecipes", "{}.yaml".format(name)) fermenter=self._find_by_id(fermenterid) try: brewname = fermenter.brewname description = fermenter.description # todo add escription at later point of time, once description has been added to fermenter dataclass except: brewname = "" description = "" self.basic_data={"name": brewname, "description": description} try: fermentersteps = fermenter.steps except: fermentersteps = [] data = dict(basic=self.basic_data, steps=list(map(lambda item: item.to_dict(), fermentersteps))) with open(path, "w") as file: yaml.dump(data, file) async def load_recipe(self, data, fermenterid): try: await self.shutdown(None, fermenterid) except: pass fermenter = self._find_by_id(fermenterid) def add_runtime_data(item): item["status"] = "I" item["id"] = shortuuid.uuid() item["props"]["Sensor"] = fermenter.sensor list(map(lambda item: add_runtime_data(item), data.get("steps"))) fermenter.description = data['basic']['desc'] fermenter.brewname = data['basic']['name'] fermenter.steps=[] await self.update(fermenter) for item in data.get("steps"): await self.create_step(fermenterid, item)