""" Validation of configs and tutorial/guide creation for recipes. """
import os
import warnings
from typing import TypedDict, Tuple, Iterator, Set
import networkx as nx
from networkx import DiGraph
from cooperative_cuisine import ROOT_DIR
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):
"""Represents a graph for meal creation with edges and layout information."""
meal: str
"""The name of the meal."""
edges: list[Tuple[str, str]]
"""A list of tuples representing the edges between cooking steps."""
layout: dict[str, Tuple[float, float]]
"""A dictionary mapping cooking step names to their layout coordinates."""
class Validation:
"""Class for performing validation tasks.
This class provides methods for performing various validation tasks, such as generating recipe graphs from item
information, creating guide graphs for each recipe/meal, asserting the presence of equipment and plate cycles,
and retrieving base ingredients and equipment required to create a meal.
"""
def __init__(
self,
meals: list[ItemInfo],
item_info: dict[str, ItemInfo],
order_manager: OrderManager,
do_validation: bool,
):
self.meals: list[ItemInfo] = meals
"""A list of ItemInfo objects representing meals."""
self.item_info: dict[str, ItemInfo] = item_info
"""All ItemInfos from the config."""
self.order_manager: OrderManager = order_manager
"""For the available meals for orders."""
self.do_validation: bool = do_validation
"""A boolean indicating whether to perform validation tasks."""
@staticmethod
def infer_recipe_graph(item_info) -> DiGraph:
"""Generate a graph from ingredients and meals and their dependencies.
Args:
item_info: A dictionary containing information about items.
Returns:
DiGraph: A directed graph representing the recipe graph.
"""
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) -> MealGraphDict:
"""Create tutorial/guide graphs for each recipe/meal.
Args:
meal: An instance of ItemInfo representing the meal to create a graph for.
Returns:
A dictionary containing the meal name, the edges of the graph, and a layout of the graph.
"""
graph = DiGraph(directed=True, rankdir="LR")
root = f"{meal.name}_0"
graph.add_node(root)
add_queue = [root] # Add "Plate_0" if dishwashing should be part of the recipe
start = True
while add_queue:
current = add_queue.pop()
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
if current_info.needs:
if len(current_info.needs) == 1:
need = f"{current_info.needs[0]}_{current_index}"
add_queue.append(need)
if current_info.equipment:
equip_id = f"{current_info.equipment.name}_{current_index}"
if current_info.equipment.equipment:
equip_equip_id = f"{current_info.equipment.equipment.name}_{current_index}"
graph.add_edge(equip_equip_id, current)
graph.add_edge(equip_id, equip_equip_id)
graph.add_edge(need, equip_id)
else:
graph.add_edge(equip_id, current)
graph.add_edge(need, equip_id)
else:
graph.add_edge(need, current)
elif len(current_info.needs) > 1:
for idx, item_name in enumerate(current_info.needs):
add_queue.append(f"{item_name}_{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}"
graph.add_edge(equip_equip_id, current)
graph.add_edge(equip_id, equip_equip_id)
graph.add_edge(f"{item_name}_{idx}", equip_id)
else:
graph.add_edge(
f"{item_name}_{idx}",
current,
)
return {
"meal": meal.name,
"edges": list(graph.edges),
"layout": nx.nx_agraph.graphviz_layout(graph, prog="dot"),
}
def reduce_item_node(self, graph, base_ingredients, item, visited):
# until now not called
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):
# 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]):
# 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):
# 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):
# until now not called
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):
# until now not called
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
@staticmethod
def get_layout_requirements(counters: list[Counter]):
"""Infer layout requirements from a list of counters.
Args:
counters: A list of Counter objects representing various counters in a layout.
Returns:
layout_requirements: A set of layout requirements based on the given counters.
This static method takes a list of Counter objects as input and returns a set of layout requirements. The
layout requirements are determined based on the type of counters and their properties.
The method iterates over each counter in the counters list. For each counter, it checks its type using the
isinstance() function and performs the required actions:
- If the counter is an instance of Dispenser or PlateDispenser, it adds the counter's dispensing name to the layout_requirements set.
- If the counter is an instance of CuttingBoard, it adds "CuttingBoard" to the layout_requirements set.
- If the counter is an instance of CookingCounter, it adds the counter's name to the layout_requirements set.
- If the counter's occupied_by property is not None and the occupied_by object has a "name" attribute, it adds the occupied_by object's name to the layout_requirements set.
Finally, the method returns the layout_requirements set.
Note: The method uses type checking and attribute checking to determine the layout requirements based on the counters provided.
Example usage:
```python
counters = [counter1, counter2, counter3]
requirements = MyClass.get_layout_requirements(counters)
print(requirements)
```
"""
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]) -> Set[str]:
"""Validates the environment by generating and saving a recipe graph, checking if ordered meals are possible,
and returning a set of meals that can be ordered.
Args:
counters (list[Counter]): A list of counters.
Returns:
meals_to_be_ordered (set): A set of meals that can be ordered, based on the environment validation.
"""
if self.do_validation:
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}"
)
meals_to_be_ordered = possible_meals.intersection(defined_meals)
return meals_to_be_ordered
else:
return {m.name for m in self.meals}
def get_recipe_graphs(self) -> list[MealGraphDict]:
"""Returns a list of recipe graphs for all available meals.
If there are no available meals, an empty list is returned.
The recipe graphs are generated and stored in the "generated" directory in the root directory (defined as
ROOT_DIR) of the application. If the directory does not exist, it will be created.
The recipe graphs are obtained by calling the `get_meal_graph` method for each available meal in the
`order_manager`. The results are collected and returned as a list.
Returns:
list[MealGraphDict]: A list of recipe graphs for all available meals.
"""
if not self.order_manager.available_meals:
return []
os.makedirs(ROOT_DIR / "generated", exist_ok=True)
return [
self.get_meal_graph(m) for m in self.order_manager.available_meals.values()
]
# # 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