Skip to content
Snippets Groups Projects
Commit a737c646 authored by Florian Schröder's avatar Florian Schröder
Browse files

Fix merge

parent 8880bf8e
No related branches found
No related tags found
1 merge request!71Resolve "Refactoring Environment class + file"
Pipeline #47672 failed
Stage: test
Hide whitespace changes
Inline Side-by-side
cooperative_cuisine/environment.py
+ 13
− 17
View file @ a737c646
......@@ -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,
......
cooperative_cuisine/recipes.py
+ 231
− 34
View file @ a737c646
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
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment