"""
You can configure the order creation/generation via the `environment_config.yml`.
It is very configurable by letting you reference own Python classes and functions.
```yaml
orders:
serving_not_ordered_meals: null
order_gen_class: !!python/name:cooperative_cuisine.orders.RandomOrderGeneration ''
order_gen_kwargs:
...
```
`serving_not_ordered_meals` expects a function. It received a meal as an argument and should return a
tuple of a bool and the score. If the bool is true, the score will be added to the score. Otherwise, it will not
accept the meal for serving.
The `order_gen_class` should be a child of the `OrderGeneration` class. The `order_gen_kwargs` depend then on your
class referenced.
This file defines the following classes:
- `Order`
- `OrderGeneration`
- `OrderManager`
Further, it defines same implementations for the basic order generation based on random sampling:
- `RandomOrderGeneration`
For an easier usage of the random orders, also some classes for type hints and dataclasses are defined:
- `RandomOrderKwarg`
- `RandomFuncConfig`
## Code Documentation
"""
from __future__ import annotations
import dataclasses
import logging
import uuid
from abc import abstractmethod
from collections import deque
from datetime import datetime, timedelta
from random import Random
from typing import Callable, Tuple, Any, Deque, TypedDict, Type, Literal
from cooperative_cuisine.hooks import (
Hooks,
SERVE_NOT_ORDERED_MEAL,
SERVE_WITHOUT_PLATE,
COMPLETED_ORDER,
INIT_ORDERS,
NEW_ORDERS,
ORDER_DURATION_SAMPLE,
ORDER_EXPIRED,
)
from cooperative_cuisine.items import Item, Plate, ItemInfo
from cooperative_cuisine.state_representation import OrderState
log = logging.getLogger(__name__)
"""The logger for this module."""
class OrderConfig(TypedDict):
"""The configuration of the order in the `environment_config`under the `order` key."""
meals: dict[Literal["all"] | Literal["list"], bool | list[str]]
"""Config of which meals are ordered in orders.
If 'all: true' all (validated) meals will be ordered, else the ones specified under 'list'."""
order_gen_class: Type[OrderGeneration]
"""The class that should handle the order generation."""
order_gen_kwargs: dict[str, Any]
"""The additional kwargs for the order gen class."""
serving_not_ordered_meals: Callable[[Item], Tuple[bool, float]]
""""""
@dataclasses.dataclass
class Order:
"""Datawrapper for Orders"""
meal: ItemInfo
"""The meal to serve and that should be cooked."""
start_time: datetime
"""The start time relative to the env_time. On which the order is returned from the get_orders func."""
max_duration: timedelta
"""The duration after which the order expires."""
uuid: str = dataclasses.field(default_factory=lambda: uuid.uuid4().hex)
"""The unique identifier for the order."""
finished_info: dict[str, Any] = dataclasses.field(default_factory=dict)
"""Is set after the order is completed."""
class OrderGeneration:
"""Base class for generating orders.
You can set your child class via the `environment_config.yml`.
Example:
```yaml
orders:
order_gen_class: !!python/name:cooperative_cuisine.orders.RandomOrderGeneration ''
kwargs:
...
```
"""
def __init__(
self,
hook: Hooks,
random: Random,
**kwargs,
):
"""Constructor of OrderGeneration.
Args:
hook: An instance of Hooks class.
random: An instance of Random class.
**kwargs: Additional keyword arguments.
"""
self.available_meals: list[ItemInfo] | None = None
"""Available meals restricted through the `environment_config.yml`."""
self.hook: Hooks = hook
"""Reference to the hook manager."""
self.random = random
"""Random instance."""
@abstractmethod
def init_orders(self, now) -> list[Order]:
"""Get the orders the environment starts with."""
...
@abstractmethod
def get_orders(
self,
passed_time: timedelta,
now: datetime,
new_finished_orders: list[Order],
expired_orders: list[Order],
) -> list[Order]:
"""Orders for each progress call. Should often be the empty list."""
...
class OrderManager:
"""The Order and Score Manager that is called from the serving window."""
def __init__(
self,
order_config,
hook: Hooks,
random: Random,
):
"""Constructor of OrderManager.
Args:
order_config: A dictionary containing the configuration for orders.
hook: An instance of the Hooks class.
random: An instance of the Random class.
"""
self.random = random
"""Random instance."""
self.order_gen: OrderGeneration = order_config["order_gen_class"](
hook=hook,
random=random,
kwargs=order_config["order_gen_kwargs"],
)
"""The order generation."""
self.serving_not_ordered_meals: Callable[
[Item], Tuple[bool, float]
] = order_config["serving_not_ordered_meals"]
"""Function that decides if not ordered meals can be served and what score it gives"""
self.available_meals = None
"""The meals for that orders can be sampled from."""
self.open_orders: Deque[Order] = deque()
"""Current open orders. This attribute is used for the environment state."""
# TODO log who / which player served which meal -> for split scores
self.served_meals: list[Tuple[Item, datetime, str]] = []
"""List of served meals. Maybe for the end screen."""
self.last_finished: list[Order] = []
"""Cache last finished orders for `OrderGeneration.get_orders` call. From the served meals."""
self.next_relevant_time: datetime = datetime.max
"""For reduced order checking. Store the next time when to create an order."""
self.last_expired: list[Order] = []
"""Cache last expired orders for `OrderGeneration.get_orders` call."""
self.hook: Hooks = hook
"""Reference to the hook manager."""
def set_available_meals(self, available_meals):
"""Set the available meals from which orders can be generated.
Args:
available_meals (dict): A dictionary containing the available meals and their quantities.
"""
self.available_meals = available_meals
self.order_gen.available_meals = list(available_meals.values())
def update_next_relevant_time(self):
"""For more efficient checking when to do something in the progress call."""
next_relevant_time = datetime.max
for order in self.open_orders:
next_relevant_time = min(
next_relevant_time, order.start_time + order.max_duration
)
self.next_relevant_time = next_relevant_time
def serve_meal(self, item: Item, env_time: datetime, player: str) -> bool:
"""Is called by the ServingWindow to serve a meal. Returns True if the meal can be served and should be
"deleted" from the hands of the player."""
if isinstance(item, Plate):
meal = item.get_potential_meal()
if meal is not None:
if meal.name in self.available_meals:
order = self.find_order_for_meal(meal)
if order is None:
if self.serving_not_ordered_meals:
self.hook(
SERVE_NOT_ORDERED_MEAL,
meal=meal,
meal_name=meal.name,
)
log.info(f"Serving meal without order {meal.name!r}")
self.served_meals.append((meal, env_time, player))
return True
log.info(
f"Do not serve meal {meal.name!r} because it is not ordered"
)
return False
order, index = order
log.info(f"Serving meal {meal.name!r} with order")
self.last_finished.append(order)
del self.open_orders[index]
self.served_meals.append((meal, env_time, player))
self.hook(
COMPLETED_ORDER,
order=order,
meal=meal,
relative_order_time=env_time - order.start_time,
meal_name=meal.name,
)
return True
else:
self.hook(SERVE_WITHOUT_PLATE, item=item)
log.info(f"Do not serve item {item}")
return False
def create_init_orders(self, env_time):
"""Create the initial orders in an environment."""
init_orders = self.order_gen.init_orders(env_time)
self.hook(INIT_ORDERS)
self.open_orders.extend(init_orders)
def progress(self, passed_time: timedelta, now: datetime):
"""Check expired orders and check order generation."""
new_orders = self.order_gen.get_orders(
passed_time=passed_time,
now=now,
new_finished_orders=self.last_finished,
expired_orders=self.last_expired,
)
if new_orders:
self.hook(NEW_ORDERS, new_orders=new_orders)
self.open_orders.extend(new_orders)
self.last_finished = []
self.last_expired = []
if new_orders or self.next_relevant_time <= now:
# reduce checking calls
remove_orders: list[int] = []
for index, order in enumerate(self.open_orders):
if now >= order.start_time + order.max_duration:
# orders expired
self.hook(ORDER_EXPIRED, order=order)
remove_orders.append(index)
continue
expired_orders: list[Order] = []
for remove_order in reversed(remove_orders):
expired_orders.append(self.open_orders[remove_order])
del self.open_orders[remove_order]
self.last_expired = expired_orders
self.update_next_relevant_time()
def find_order_for_meal(self, meal) -> Tuple[Order, int] | None:
"""Get the order that will be fulfilled for a meal. At the moment the oldest order in the list that has the
same meal (name)."""
for index, order in enumerate(self.open_orders):
if order.meal.name == meal.name:
return order, index
def order_state(self) -> list[OrderState]:
"""Similar to the `to_dict` in `Item` and `Counter`. Relevant for the state of the environment"""
return [
{
"id": order.uuid,
"category": "Order",
"meal": order.meal.name,
"start_time": order.start_time.isoformat(),
"max_duration": order.max_duration.total_seconds(),
}
for order in self.open_orders
]
class RandomFuncConfig(TypedDict):
"""Types of the dict for sampling with different random functions from the [`random` library](https://docs.python.org/3/library/random.html).
Example:
Sampling [uniform](https://docs.python.org/3/library/random.html#random.uniform)ly between `10` and `20`.
```yaml
func: uniform
kwargs:
a: 10
b: 20
```
Or in Python:
```python
random_func = {'func': 'uniform', 'kwargs': {'a': 10, 'b': 20}}
```
"""
func: str | Callable
"""the name of a functions in the `random` library."""
kwargs: dict
"""the kwargs of the functions in the `random` library."""
@dataclasses.dataclass
class RandomOrderKwarg:
num_start_meals: int
"""Number of meals sampled at the start."""
sample_on_serving: bool
"""Only sample the delay for the next order after a meal was served."""
sample_on_dur_random_func: RandomFuncConfig
"""How to sample the delay of the next incoming order. Either after a new meal was served or the last order was
generated (based on the `sample_on_serving` attribute)."""
max_orders: int
"""How many orders can maximally be active at the same time."""
order_duration_random_func: RandomFuncConfig
"""How long the order is alive until it expires. If `sample_on_serving` is `true` all orders have no expire time."""
class RandomOrderGeneration(OrderGeneration):
"""A simple order generation based on random sampling with two options.
Either sample the delay when a new order should come in after the last order comes in or after a meal was served
(and an order got removed).
To configure it align your kwargs with the `RandomOrderKwarg` class.
You can set this order generation in your `environment_config.yml` with
```yaml
orders:
order_gen_class: !!python/name:cooperative_cuisine.orders.RandomOrderGeneration ''
# the class to that receives the kwargs. Should be a child class of OrderGeneration in orders.py
order_gen_kwargs:
order_duration_random_func:
# how long should the orders be alive
# 'random' library call with getattr, kwargs are passed to the function
func: uniform
kwargs:
a: 40
b: 60
max_orders: 6
# maximum number of active orders at the same time
num_start_meals: 2
# number of orders generated at the start of the environment
sample_on_dur_random_func:
# 'random' library call with getattr, kwargs are passed to the function
func: uniform
kwargs:
a: 10
b: 20
sample_on_serving: false
# Sample the delay for the next order only after a meal was served.
serving_not_ordered_meals: true
# can meals that are not ordered be served / dropped on the serving window
```
"""
def __init__(
self,
hook: Hooks,
random: Random,
**kwargs,
):
"""Constructor of RandomOrderGeneration.
Args:
hook (Hooks): The hook object.
random (Random): The random object.
**kwargs: Additional keyword arguments.
"""
super().__init__(hook, random, **kwargs)
self.kwargs: RandomOrderKwarg = RandomOrderKwarg(**kwargs["kwargs"])
"""Configuration og the RandomOrder genration. See `RandomOrderKwarg`"""
self.next_order_time: datetime | None = datetime.max
"""For efficient checking to update order removable."""
self.number_cur_orders: int = 0
"""How many orders are currently open."""
self.needed_orders: int = 0
"""For the sample on dur but when it was restricted due to max order number."""
def init_orders(self, now) -> list[Order]:
self.number_cur_orders = self.kwargs.num_start_meals
if not self.kwargs.sample_on_serving:
self.create_random_next_time_delta(now)
if self.available_meals:
return self.create_orders_for_meals(
self.random.choices(
self.available_meals, k=self.kwargs.num_start_meals
),
now,
self.kwargs.sample_on_serving,
)
self.number_cur_orders = 0
return []
def get_orders(
self,
passed_time: timedelta,
now: datetime,
new_finished_orders: list[Order],
expired_orders: list[Order],
) -> list[Order]:
if not self.available_meals:
return []
self.number_cur_orders -= len(new_finished_orders)
self.number_cur_orders -= len(expired_orders)
if self.kwargs.sample_on_serving:
if new_finished_orders:
self.create_random_next_time_delta(now)
return []
if self.needed_orders:
self.needed_orders -= len(new_finished_orders)
self.needed_orders = max(self.needed_orders, 0)
self.number_cur_orders += len(new_finished_orders)
return self.create_orders_for_meals(
self.random.choices(self.available_meals, k=len(new_finished_orders)),
now,
)
if self.next_order_time <= now:
if self.number_cur_orders >= self.kwargs.max_orders:
self.needed_orders += 1
else:
if not self.kwargs.sample_on_serving:
self.create_random_next_time_delta(now)
else:
self.next_order_time = datetime.max
self.number_cur_orders += 1
return self.create_orders_for_meals(
[self.random.choice(self.available_meals)],
now,
)
return []
def create_orders_for_meals(
self, meals: list[ItemInfo], now: datetime, no_time_limit: bool = False
) -> list[Order]:
"""Create order objects for given meals by sampling the duration of the orders.
Args:
meals: A list of ItemInfo objects representing the meals for which orders need to be created.
now: A datetime object representing the current env time (the start time of the orders).
no_time_limit: A boolean indicating whether there should be no limit on the order duration. Defaults to False.
Returns:
A list of Order objects representing the created orders.
"""
orders = []
for meal in meals:
if no_time_limit:
duration = timedelta(days=365)
else:
if isinstance(self.kwargs.order_duration_random_func["func"], str):
seconds = getattr(
self.random, self.kwargs.order_duration_random_func["func"]
)(**self.kwargs.order_duration_random_func["kwargs"])
else:
seconds = self.kwargs.order_duration_random_func["func"](
**self.kwargs.order_duration_random_func["kwargs"]
)
duration = timedelta(seconds=seconds)
self.hook(
ORDER_DURATION_SAMPLE,
duration=duration,
)
log.info(f"Create order for meal {meal} with duration {duration}")
orders.append(
Order(
meal=meal,
start_time=now,
max_duration=duration,
)
)
return orders
def create_random_next_time_delta(self, now: datetime):
if isinstance(self.kwargs.order_duration_random_func["func"], str):
seconds = getattr(
self.random, self.kwargs.order_duration_random_func["func"]
)(**self.kwargs.order_duration_random_func["kwargs"])
else:
seconds = self.kwargs.order_duration_random_func["func"](
**self.kwargs.order_duration_random_func["kwargs"]
)
self.next_order_time = now + timedelta(seconds=seconds)
log.info(f"Next order in {self.next_order_time}")