Planner Lead Campaigns Support

PyAres/Planning/__init__.py

@@ -1,4 +1,4 @@
-from .planner_models import PlanningParameter, PlanRequest, PlanResponse, ParameterHistoryItem, Plan, PlannedParameter
+from .planner_models import PlanningParameter, PlanRequest, PlanResponse, ParameterHistoryItem, Plan, PlannedParameter, ObjectiveStatus
 from .planning_service import AresPlannerService
 
 __all__ = [
@@ -9,4 +9,5 @@
     "Plan",
     "PlannedParameter",
     "ParameterHistoryItem",
+    "ObjectiveStatus"
 ]

PyAres/Planning/planner_models.py

@@ -21,6 +21,13 @@ def __str__(self):
 
     def __repr__(self) -> str:
         return self.__str__()
+
+class ObjectiveStatus(Enum):
+    """ An enum representing the current status of the objective the planner is trying to achieve (if any) """
+    OBJECTIVE_STATUS_UNSPECIFIED = 0
+    OBJECTIVE_UNACHIEVED = 1
+    OBJECTIVE_ACHIEVED = 2
+    OBJECTIVE_FAILED = 3
 
 class PlanningParameter:
     """
@@ -188,7 +195,8 @@ def __init__(self,
                  parameter_values: Optional[list] = None,
                  parameter_data: Optional[dict[str,Any]] = None,
                  outcome: Outcome = Outcome.SUCCESS, 
-                 error_string: str = ""):
+                 error_string: str = "",
+                 objective_status: ObjectiveStatus = ObjectiveStatus.OBJECTIVE_STATUS_UNSPECIFIED):
         """
         Initializes a PlanResponse. Using either lists of names and values or a python dictonary of name:value pairs
 
@@ -198,6 +206,7 @@ def __init__(self,
             parameter_data: A python dictionary of key:value pairs of planned parameters and planned values
             outcome: An enum of type Outcome that determines whether the planning process succeeded or not, defaults to SUCCESS
             error_string: An optional string for specifying planning failure reasons to be relayed to ARES
+            objective_status: An optional value that specifies the status of the objective your planner is trying to achieve (if any)
         """
         if parameter_data is not None:
             self.parameter_names = list(parameter_data.keys())
@@ -214,13 +223,15 @@ def __init__(self,
 
         self.outcome = outcome
         self.error_string = error_string
+        self.objective_status = objective_status
 
     def __str__(self):
         return (f"PlanResponse object with:\n"
                 f" outcome: {self.outcome}\n"
                 f" parameter_names: {self.parameter_names}\n"
                 f" parameter_values: {self.parameter_values}\n"
-                f" error_string: {self.error_string}\n")
+                f" error_string: {self.error_string}\n"
+                f" objective_status: {self.objective_status}\n")
 
     def __repr__(self) -> str:
         return self.__str__()
@@ -239,13 +250,18 @@ def __repr__(self):
         return f"PlannedParameters(parameter_name={self.parameter_name!r}, parameter_value={self.parameter_value!r})"
 
 class Plan:
-    def __init__(self, planned_parameters: List[PlannedParameter], outcome: Outcome, error_string: str = ""):
+    def __init__(self, 
+                 planned_parameters: List[PlannedParameter], 
+                 outcome: Outcome, 
+                 error_string: str = "", 
+                 objective_status: ObjectiveStatus = ObjectiveStatus.OBJECTIVE_STATUS_UNSPECIFIED):
         self.planned_parameters = planned_parameters
         self.outcome = outcome
         self.error_string = error_string
+        self.objective_status = objective_status
 
     def __str__(self):
-        base_str = f"Plan (Outcome: {self.outcome})"
+        base_str = f"Plan (Outcome: {self.outcome}) \n (Objective Status: {self.objective_status})"
 
         # Format the list of parameters into a readable string
         if self.planned_parameters:

PyAres/Planning/planning_service.py

@@ -23,6 +23,7 @@
 from ..Utils import ares_outcome_utils
 from ..Utils import ares_plan_status_code_utils
 from ..Utils import plan_response_utils
+from ..Utils import ares_objective_status_utils
 
 # Import python models
 from ..Models import ares_data_models, Limits
@@ -146,6 +147,7 @@ def Plan(self, request: plan_pb2.PlanningRequest, context) -> plan_pb2.PlanningR
         if isinstance(python_response, PlanResponse):
             response_proto.planning_outcome = ares_outcome_utils.python_ares_outcome_to_proto_ares_outcome(python_response.outcome)
             response_proto.error_string = python_response.error_string
+            response_proto.objective_status = ares_objective_status_utils.python_ares_outcome_to_proto_ares_outcome(python_response.objective_status)
 
             for i in range(len(python_response.parameter_names)):
                 planned_parameter = plan_pb2.PlannedParameter(parameter_value=ares_value_utils.create_ares_value(python_response.parameter_values[i]))

PyAres/Utils/ares_objective_status_utils.py

@@ -0,0 +1,10 @@
+from ..Planning.planner_models import ObjectiveStatus
+from ares_datamodel.planning import plan_pb2
+
+def python_ares_outcome_to_proto_ares_outcome(py_value: ObjectiveStatus) -> int:
+  """ A method to convert from the python AresDataType class to the protobuf version """
+  return py_value.value
+
+def proto_ares_outcome_to_python_ares_outcome(proto_value: plan_pb2.ObjectiveStatus) -> ObjectiveStatus:
+  """ A method to convert from the protobuf AresDataType class to the python version """
+  return ObjectiveStatus(proto_value)

PyAres/Utils/ares_outcome_utils.py

@@ -1,6 +1,5 @@
 from ..Models import Outcome
 from ares_datamodel import ares_outcome_enum_pb2
-from typing import cast
 
 def python_ares_outcome_to_proto_ares_outcome(py_value: Outcome) -> int:
   """ A method to convert from the python AresDataType class to the protobuf version """

PyAres/__init__.py

@@ -5,6 +5,7 @@
 from .Planning import ParameterHistoryItem
 from .Planning import PlannedParameter
 from .Planning import Plan
+from .Planning import ObjectiveStatus
 from .Analyzing import AresAnalyzerService
 from .Analyzing import AnalysisResponse
 from .Analyzing import AnalysisRequest