Resolve "Score of order depends on remaining time"

Closes #128 (closed)

cooperative_cuisine/scores.py

@@ -39,18 +39,123 @@ hook_callbacks:
     callback_class: !!python/name:cooperative_cuisine.scores.ScoreViaHooks ''
     callback_class_kwargs:
       static_score: -10
+
+The score can be modified based on the time remaining for the order to be completed.
+hook_callbacks:
+  orders:
+    hooks: [ completed_order ]
+    callback_class: !!python/name:cooperative_cuisine.scores.ScoreViaHooks ''
+    callback_class_kwargs:
+    time_dependence_func: !!python/name:cooperative_cuisine.scores.linear_score ''
+      time_dependence_kwargs:
+        min_score_ratio: 0.3
+        round_decimals: 2
+      static_score: 100
+      score_on_specific_kwarg: meal_name
+      score_map: []
+
+    hooks: [ completed_order ]
+    callback_class: !!python/name:cooperative_cuisine.scores.ScoreViaHooks ''
+    callback_class_kwargs:
+    time_dependence_func: !!python/name:cooperative_cuisine.scores.stepped_score ''
+      time_dependence_kwargs:
+        steps: [0.3, 0.6, 1.0]
+        ratios: [0.3, 0.6, 1.0]
+        round_decimals: 2
+      static_score: 100
+      score_on_specific_kwarg: meal_name
+      score_map: []
+
+    hooks: [ completed_order ]
+    callback_class: !!python/name:cooperative_cuisine.scores.ScoreViaHooks ''
+    callback_class_kwargs:
+    time_dependence_func: !!python/name:cooperative_cuisine.scores.constant_score ''
+      time_dependence_kwargs: {}
+      static_score: 100
+      score_on_specific_kwarg: meal_name
+      score_map: []
 ```
 
 
 # Code Documentation
 """
+from __future__ import annotations
+
+from typing import Any, TYPE_CHECKING
 
-from typing import Any
+import numpy as np
 
-from cooperative_cuisine.environment import Environment
+if TYPE_CHECKING:
+    from cooperative_cuisine.environment import Environment
 from cooperative_cuisine.hooks import HookCallbackClass
 
 
+def constant_score(max_score: float, time_percentage: float = 1.0):
+    """Returns the constant score. No time dependence.
+
+    Args:
+        max_score: The maximum score to be returned.
+        time_percentage: The time percentage. Default is 1.0.
+    """
+    return max_score
+
+
+def linear_score(
+    max_score: float,
+    min_score_ratio: float = 0.0,
+    time_percentage: float = 1.0,
+    round_decimals: int = 0,
+):
+    """
+    Returns the linear score based on the time percentage.
+
+    Args:
+        max_score: Maximum possible score.
+        min_score_ratio: Minimum score ratio. Default is 0.0.
+        time_percentage: Time percentage of the order. Default is 1.0.
+        round_decimals: Number of decimals to round the score to. Default is 0.
+
+    Returns: Modified score based on the time percentage.
+    """
+    modified_score = float(
+        np.round(
+            max(max_score * time_percentage, max_score * min_score_ratio),
+            decimals=round_decimals,
+        )
+    )
+    return modified_score
+
+
+def stepped_score(
+    max_score: float,
+    steps: list[float],
+    score_ratios: list[float],
+    round_decimals: int = 0,
+    time_percentage: float = 1.0,
+):
+    """
+    Modifies the score based on a step function based on the time percentage.
+
+    Args:
+        max_score: Maximum possible score.
+        steps: Steps for the score ratios.
+        score_ratios: Ratios for the score based on the steps.
+        round_decimals: Number of decimals to round the score to. Default is 0.
+        time_percentage: Time percentage of the order. Default is 1.0.
+
+    Returns: Modified score based on the time percentage.
+
+    """
+    if len(steps) != len(score_ratios):
+        raise ValueError("steps and vals must have the same length")
+
+    for threshold, ratio in zip(reversed(steps), reversed(score_ratios)):
+        if max(time_percentage, 0) >= threshold:
+            return float(np.round(max_score * ratio, decimals=round_decimals))
+
+    assert False, "This should not happen."
+
+
 class ScoreViaHooks(HookCallbackClass):
     """
     Defines a class ScoreViaHooks that extends the HookCallbackClass.
@@ -71,6 +176,8 @@ class ScoreViaHooks(HookCallbackClass):
         static_score: float = 0,
         score_map: dict[str, float] = None,
         score_on_specific_kwarg: str = None,
+        time_dependence_func: callable = constant_score,
+        time_dependence_kwargs: dict[str, Any] = None,
         kwarg_filter: dict[str, Any] = None,
         **kwargs,
     ):
@@ -94,24 +201,47 @@ class ScoreViaHooks(HookCallbackClass):
         """Filtering condition for keyword arguments."""
         self.score_on_specific_kwarg: str = score_on_specific_kwarg
         """The specific keyword argument to score on."""
+        self.time_dependence_func: callable = time_dependence_func
+        """The function to calculate the score based on time."""
+        self.time_dependence_kwargs: dict[str, Any] = (
+            time_dependence_kwargs if time_dependence_kwargs else {}
+        )
+        """The keyword arguments to be passed to the time_dependence_func."""
 
-    def __call__(self, hook_ref: str, env: Environment, **kwargs):
+    def get_score(self, hook_ref: str, **kwargs):
         if self.score_on_specific_kwarg:
             if kwargs[self.score_on_specific_kwarg] in self.score_map:
-                self.env.increment_score(
-                    self.score_map[kwargs[self.score_on_specific_kwarg]],
-                    info=f"{hook_ref} - {kwargs[self.score_on_specific_kwarg]}",
-                )
+                score = self.score_map[kwargs[self.score_on_specific_kwarg]]
+                info = f"{hook_ref} - {kwargs[self.score_on_specific_kwarg]}"
             else:
-                self.env.increment_score(self.static_score, info=hook_ref)
+                score = self.static_score
+                info = hook_ref
         elif self.score_map and hook_ref in self.score_map:
             if self.kwarg_filter:
                 if self.kwarg_filter.items() <= kwargs.items():
-                    self.env.increment_score(
-                        self.score_map[hook_ref],
-                        info=f"{hook_ref} - {self.kwarg_filter}",
-                    )
+                    score = self.score_map[hook_ref]
+                    info = f"{hook_ref} - {self.kwarg_filter}"
+                else:
+                    score = 0
+                    info = "NO INFO?"
             else:
-                self.env.increment_score(self.score_map[hook_ref], info=hook_ref)
+                score = self.score_map[hook_ref]
+                info = hook_ref
         else:
-            self.env.increment_score(self.static_score, info=hook_ref)
+            score = self.static_score
+            info = hook_ref
+
+        if score:
+            if hook_ref == "completed_order":
+                ratio = kwargs["remaining_time_ratio"]
+            else:
+                ratio = 1.0
+            modified_score = self.time_dependence_func(
+                score, time_percentage=ratio, **self.time_dependence_kwargs
+            )
+            return modified_score, info
+        return 0, "NO INFO?"
+
+    def __call__(self, hook_ref: str, env: Environment, **kwargs):
+        score, info = self.get_score(hook_ref, **kwargs)
+        self.env.increment_score(score, info=info)