diff --git a/cooperative_cuisine/environment.py b/cooperative_cuisine/environment.py
index 28036d32034f1a82155fc8c93fe09639ab562f3d..5410e5e4355a3d29fa14bc4e7b8fca79ace95004 100644
--- a/cooperative_cuisine/environment.py
+++ b/cooperative_cuisine/environment.py
@@ -4,7 +4,6 @@ import inspect
import json
import logging
import sys
-import warnings
from collections import defaultdict
from datetime import timedelta, datetime
from pathlib import Path
@@ -21,9 +20,6 @@ from cooperative_cuisine.counter_factory import (
)
from cooperative_cuisine.counters import (
PlateConfig,
- PlateDispenser,
- CuttingBoard,
- CookingCounter,
)
from cooperative_cuisine.effects import EffectManager
from cooperative_cuisine.hooks import (
@@ -220,8 +216,19 @@ class Environment:
"""Counters that needs to be called in the step function via the `progress` method."""
self.overwrite_counters(self.counters)
- # TODO Maybe validation can be turned off in config...
- meals_to_be_ordered = self.validate_environment()
+ self.recipe_validation = RecipeValidation(
+ meals=[m for m in self.item_info.values() if m.type == ItemType.Meal]
+ if self.environment_config["meals"]["all"]
+ else [
+ self.item_info[m]
+ for m in self.environment_config["meals"]["list"]
+ if self.item_info[m].type == ItemType.Meal
+ ],
+ item_info=self.item_info,
+ order_manager=self.order_manager,
+ )
+
+ meals_to_be_ordered = self.recipe_validation.validate_environment()
assert meals_to_be_ordered, "Need possible meals for order generation."
available_meals = {meal: self.item_info[meal] for meal in meals_to_be_ordered}
@@ -241,17 +248,6 @@ class Environment:
self.info_msgs_per_player: dict[str, list[InfoMsg]] = defaultdict(list)
- self.recipe_validation = RecipeValidation(
- meals=[m for m in self.item_info.values() if m.type == ItemType.Meal]
- if self.environment_config["meals"]["all"]
- else [
- self.item_info[m]
- for m in self.environment_config["meals"]["list"]
- if self.item_info[m].type == ItemType.Meal
- ],
- item_info=self.item_info,
- )
-
self.hook(
ENV_INITIALIZED,
environment_config=env_config,
diff --git a/cooperative_cuisine/recipes.py b/cooperative_cuisine/recipes.py
index dbd5d1c33bdb5db5b88a5d72b362a712e5a7dfef..cdf34b259801011beff9492b3dfb58a60f72c65f 100644
--- a/cooperative_cuisine/recipes.py
+++ b/cooperative_cuisine/recipes.py
@@ -1,12 +1,21 @@
import os
+import warnings
from concurrent.futures import ThreadPoolExecutor
-from typing import TypedDict, Tuple
+from typing import TypedDict, Tuple, Iterator
import networkx as nx
-from networkx import DiGraph
+from networkx import DiGraph, Graph
from cooperative_cuisine import ROOT_DIR
-from cooperative_cuisine.items import ItemInfo
+from cooperative_cuisine.counters import (
+ Dispenser,
+ CuttingBoard,
+ CookingCounter,
+ PlateDispenser,
+ Counter,
+)
+from cooperative_cuisine.items import ItemInfo, ItemType, Item
+from cooperative_cuisine.orders import OrderManager
class MealGraphDict(TypedDict):
@@ -16,43 +25,67 @@ class MealGraphDict(TypedDict):
class RecipeValidation:
- def __init__(self, meals, item_info):
+ def __init__(self, meals, item_info, order_manager):
self.meals: list[ItemInfo] = meals
self.item_info: dict[str, ItemInfo] = item_info
+ self.order_manager: OrderManager = order_manager
- def get_meal_graph(self, meal: ItemInfo) -> MealGraphDict:
- graph = DiGraph(
- directed=True, rankdir="LR", graph_attr={"nslimit": "0", "nslimit1": "2"}
- )
+ @staticmethod
+ def infer_recipe_graph(item_info) -> DiGraph:
+ colors = {
+ ItemType.Ingredient: "black",
+ ItemType.Equipment: "red",
+ ItemType.Meal: "green",
+ ItemType.Waste: "brown",
+ }
+
+ graph = DiGraph(directed=True)
+ for item_name, item_info in item_info.items():
+ graph.add_node(item_name, color=colors.get(item_info.type, "blue"))
+ if item_info.equipment is None:
+ for item in item_info.needs:
+ graph.add_edge(item, item_name)
+ else:
+ if len(item_info.needs) > 0:
+ for item in item_info.needs:
+ graph.add_edge(item, item_info.equipment.name)
+ graph.add_edge(item_info.equipment.name, item_name)
+ else:
+ graph.add_edge(item_name, item_info.equipment.name)
+ return graph
+
+ def get_meal_graph(self, meal: ItemInfo) -> tuple[Graph, dict[str, list[float]]]:
+ graph = DiGraph(directed=True, rankdir="LR")
- root = f"{meal.name}_0"
+ root = meal.name + "_0"
graph.add_node(root)
- add_queue = [root]
+ add_queue = ["Plate_0", root]
start = True
while add_queue:
current = add_queue.pop()
- current_info, current_index = current.rsplit("_", maxsplit=1)
- current_info = self.item_info[current_info]
+ current_info = self.item_info[current.split("_")[0]]
+ current_index = current.split("_")[-1]
if start:
graph.add_edge("Plate_0", current)
current = "Plate_0"
start = False
- # maybe reduce? double code fragments
if current_info.needs:
if len(current_info.needs) == 1:
- need = f"{current_info.needs[0]}_{current_index}"
+ need = current_info.needs[0] + f"_{current_index}"
add_queue.append(need)
if current_info.equipment:
- equip_id = f"{current_info.equipment.name}_{current_index}"
+ equip_id = current_info.equipment.name + f"_{current_index}"
if current_info.equipment.equipment:
- equip_equip_id = f"{current_info.equipment.equipment.name}_{current_index}"
-
+ equip_equip_id = (
+ current_info.equipment.equipment.name
+ + f"_{current_index}"
+ )
graph.add_edge(equip_equip_id, current)
graph.add_edge(equip_id, equip_equip_id)
graph.add_edge(need, equip_id)
@@ -64,30 +97,194 @@ class RecipeValidation:
elif len(current_info.needs) > 1:
for idx, item_name in enumerate(current_info.needs):
- need = f"{item_name}_{idx}"
- add_queue.append(need)
+ add_queue.append(item_name + f"_{idx}")
if current_info.equipment and current_info.equipment.equipment:
- equip_id = f"{current_info.equipment.name}_{current_index}"
- equip_equip_id = f"{current_info.equipment.equipment.name}_{current_index}"
+ equip_id = current_info.equipment.name + f"_{current_index}"
+ equip_equip_id = (
+ current_info.equipment.equipment.name
+ + f"_{current_index}"
+ )
graph.add_edge(equip_equip_id, current)
graph.add_edge(equip_id, equip_equip_id)
- graph.add_edge(need, equip_id)
+ graph.add_edge(item_name + f"_{idx}", equip_id)
else:
- graph.add_edge(need, current)
- return {
- "meal": meal.name,
- "edges": list(graph.edges),
- "layout": nx.nx_agraph.graphviz_layout(graph, prog="dot"),
- }
+ graph.add_edge(
+ item_name + f"_{idx}",
+ current,
+ )
+
+ agraph = nx.nx_agraph.to_agraph(graph)
+ layout = nx.nx_agraph.graphviz_layout(graph, prog="dot")
+ agraph.draw(
+ ROOT_DIR / "generated" / f"recipe_graph_{meal.name}.png",
+ format="png",
+ prog="dot",
+ )
+
+ return graph, layout
+
+ def reduce_item_node(self, graph, base_ingredients, item, visited):
+ visited.append(item)
+ if item in base_ingredients:
+ return True
+ else:
+ return all(
+ self.reduce_item_node(graph, base_ingredients, pred, visited)
+ for pred in graph.predecessors(item)
+ if pred not in visited
+ )
+
+ def assert_equipment_is_present(self, counters):
+ # TODO until now not called
+ expected = set(
+ name
+ for name, info in self.item_info.items()
+ if info.type == ItemType.Equipment and "Plate" not in info.name
+ )
+ counters = set(c.__class__.__name__ for c in counters).union(
+ set(c.name for c in counters if hasattr(c, "name"))
+ )
+ items = set(
+ c.occupied_by.name
+ for c in counters
+ if c.occupied_by is not None and isinstance(c.occupied_by, Item)
+ )
+ for equipment in expected:
+ if equipment not in counters and equipment not in items:
+ raise ValueError(
+ f"Equipment '{equipment}' from config files not found in the environment layout.\n"
+ f"Config Equipment: {sorted(expected)}\n"
+ f"Layout Counters: {sorted(counters)}\n"
+ f"Layout Items: {sorted(items)}"
+ )
+
+ def assert_plate_cycle_present(self, counters: list[Counter]):
+ # TODO until now not called
+ for plate in ["Plate", "DirtyPlate"]:
+ if plate not in self.item_info:
+ raise ValueError(f"{plate} not found in item info")
+
+ relevant_counters = ["PlateDispenser", "ServingWindow"]
+ for counter in counters:
+ if isinstance(counter, PlateDispenser):
+ if counter.plate_config.return_dirty:
+ relevant_counters = [
+ "PlateDispenser",
+ "ServingWindow",
+ "Sink",
+ "SinkAddon",
+ ]
+
+ counter_names = [c.__class__.__name__ for c in counters]
+ for counter in relevant_counters:
+ if counter not in counter_names:
+ raise ValueError(f"{counter} not found in counters")
+
+ @staticmethod
+ def assert_no_orphans(graph: DiGraph):
+ # TODO until now not called
+ orphans = [
+ n
+ for n in graph.nodes()
+ if graph.in_degree(n) == 0 and graph.out_degree(n) == 0
+ ]
+ if orphans:
+ raise ValueError(
+ f"Expected all items to be part of a recipe, but found orphans: {orphans}"
+ )
+
+ @staticmethod
+ def assert_roots_are_dispensable(graph, base_ingredients):
+ root_nodes = [
+ n for n in graph.nodes() if graph.in_degree(n) == 0 and "Plate" not in n
+ ]
+ if set(root_nodes) != set(base_ingredients):
+ raise ValueError(
+ f"Expected root nodes in the recipe graph and dispensable items to be identical, but found\n "
+ f"Root nodes: {sorted(root_nodes)}\n"
+ f"Dispensable items: {sorted(base_ingredients)}"
+ )
+
+ def assert_meals_are_reducible(self, graph, base_ingredients):
+ meals = [n for n in graph.nodes() if self.item_info[n].type == ItemType.Meal]
+
+ for meal in meals:
+ visited = []
+ if not self.reduce_item_node(graph, base_ingredients, meal, visited):
+ raise ValueError(
+ f"Meal '{meal}' can not be reduced to base ingredients"
+ )
+
+ def get_requirements(self, item_name: str) -> Iterator[str]:
+ """
+ Get all base ingredients and equipment required to create the given meal.
+ """
+ item = self.item_info[item_name]
+ is_equipment = item.type == ItemType.Equipment
+ is_base_ingredient = item.type == ItemType.Ingredient and not item.needs
+
+ if is_equipment or is_base_ingredient:
+ yield item_name
+ for need in item.needs:
+ yield from self.get_requirements(need)
+ if item.equipment is not None:
+ yield from self.get_requirements(item.equipment.name)
+
+ def get_item_info_requirements(self) -> dict[str, set[str]]:
+ recipes = {}
+ for item_name, item_info in self.item_info.items():
+ if item_info.type == ItemType.Meal:
+ requirements = set(r for r in self.get_requirements(item_name))
+ recipes[item_name] = requirements | {"Plate"}
+ return recipes
+
+ def get_layout_requirements(self, counters: list[Counter]):
+ layout_requirements = set()
+ for counter in counters:
+ if isinstance(counter, (Dispenser, PlateDispenser)):
+ layout_requirements.add(counter.dispensing.name)
+ if isinstance(counter, CuttingBoard):
+ layout_requirements.add("CuttingBoard")
+ if isinstance(counter, CookingCounter):
+ layout_requirements.add(counter.name)
+ if counter.occupied_by is not None and hasattr(counter.occupied_by, "name"):
+ layout_requirements.add(counter.occupied_by.name)
+ return layout_requirements
+
+ def validate_environment(self, counters: list[Counter]):
+ graph = self.infer_recipe_graph(self.item_info)
+ os.makedirs(ROOT_DIR / "generated", exist_ok=True)
+ nx.nx_agraph.to_agraph(graph).draw(
+ ROOT_DIR / "generated" / "recipe_graph.png", format="png", prog="dot"
+ )
+
+ expected = self.get_item_info_requirements()
+ present = self.get_layout_requirements(counters)
+ possible_meals = set(meal for meal in expected if expected[meal] <= present)
+ defined_meals = set(map(lambda i: i.name, self.meals))
+
+ # print(f"Ordered meals: {defined_meals}, Possible meals: {possible_meals}")
+ if len(defined_meals - possible_meals) > 0:
+ warnings.warn(
+ f"Ordered meals are not possible: {defined_meals - possible_meals}"
+ )
- def validate_item_info(self):
- """TODO"""
- raise NotImplementedError
+ meals_to_be_ordered = possible_meals.intersection(defined_meals)
+ return meals_to_be_ordered
+ # print("FINAL MEALS:", meals_to_be_ordered)
- def get_recipe_graphs(self) -> list[MealGraphDict]:
+ def get_recipe_graphs(self) -> list:
os.makedirs(ROOT_DIR / "generated", exist_ok=True)
- with ThreadPoolExecutor(max_workers=len(self.meals)) as executor:
- graph_dicts = list(executor.map(self.get_meal_graph, self.meals))
+ # time_start = time.time()
+ with ThreadPoolExecutor(
+ max_workers=len(self.order_manager.available_meals)
+ ) as executor:
+ graph_dicts = list(
+ executor.map(
+ self.get_meal_graph, self.order_manager.available_meals.values()
+ )
+ )
+ # print("DURATION", time.time() - time_start)
return graph_dicts