Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -165,3 +165,9 @@ cython_debug/

# PyPI configuration file
.pypirc

# VS Code
.vscode/

#macOS
.DS_Store
21 changes: 20 additions & 1 deletion PyAres/Analyzing/analyzer_models.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,17 @@ def __init__(self, inputs: Dict[str, Any], settings: Dict[str, Any], metadata: R
self.settings = settings
self.request_metadata = metadata

def __str__(self) -> str:
"""Returns a string representation of the AnalysisRequest with all information organized."""
metadata_str = str(self.request_metadata).replace('\n', '\n\t')
return (f"AnalysisRequest object with:\n"
f" inputs: {self.inputs}\n"
f" settings: {self.settings}\n"
f" metadata: {metadata_str}")

def __repr__(self) -> str:
return self.__str__()

class Analysis:
""" Represents the result of an analysis process. """

Expand All @@ -24,7 +35,15 @@ def __init__(self, result: float, outcome: Outcome = Outcome.SUCCESS, error_stri
self.result = result
self.outcome = outcome
self.error_string = error_string


def __str__(self) -> str:
return (f"Analysis object with:\n"
f" result: {self.result}\n"
f" outcome: {self.outcome}\n"
f" error_string: {self.error_string}")

def __repr__(self) -> str:
return self.__str__()

class InfoResponse:
""" A response message that provides basic information about your analyzer. """
Expand Down
11 changes: 11 additions & 0 deletions PyAres/Models/ares_data_models.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,17 @@ def __init__(self, proto_metadata: request_metadata_pb2.RequestMetadata):
dt = proto_metadata.experiment_start_time.ToDatetime()
self.experiment_start_time = dt.strftime("%Y-%m-%d %H:%M:%S")

def __str__(self) ->str:
return (f"RequestMetadata object with fields:\n"
f"System Name: {self.system_name}\n"
f"Campaign Name: {self.campaign_name}\n"
f"Campaign ID: {self.campaign_id}\n"
f"Experiment ID: {self.experiment_id}\n"
f"Experiment Start Time: {self.experiment_start_time}\n")

def __repr__(self) -> str:
return self.__str__()

@classmethod
def from_default_values(cls):
""" Alternative constructor for creating fake metadata """
Expand Down
116 changes: 109 additions & 7 deletions PyAres/Planning/planner_models.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
from typing import Dict, Any, List, Sequence
from typing import Dict, Any, List, Sequence, Optional
from ..Models import Outcome, AresDataType, RequestMetadata

class ParameterHistoryItem:
Expand All @@ -13,6 +13,13 @@ def __init__(self, planned_value: Any, achieved_value: Any):
"""
self.planned_value = planned_value
self.achieved_value = achieved_value
def __str__(self):
return (f"ParameterHistoryItem object with:\n"
f" planned_value: {self.planned_value}\n"
f" acheived_value: {self.achieved_value}\n")

def __repr__(self) -> str:
return self.__str__()

class PlanningParameter:
"""
Expand Down Expand Up @@ -47,7 +54,34 @@ def __init__(self, name: str, minimum_value: float,
self.is_result: bool = is_result
self.planner_name: str = planner_name
self.initial_value = initial_value


def __str__(self):
return (f"ParameterHistoryItem object with:\n"
f" name: {self.name}\n"
f" minimum_value: {self.minimum_value}\n"
f" maximum_value: {self.maximum_value}\n"
f" param_history: {len(self.param_history)} records\n"
f" data_type: {self.data_type}\n"
f" is_planned: {self.is_planned}\n"
f" is_result: {self.is_result}\n"
f" planner_name: {self.planner_name}\n"
f" initial_value: {self.initial_value}")

def __repr__(self) -> str:
return self.__str__()

@property
def planned_values(self) -> list:
return [item.planned_value for item in self.param_history]

@property
def achieved_values(self) -> list:
return [item.achieved_value for item in self.param_history]

@property
def bounds(self) -> list:
return [self.minimum_value, self.maximum_value]


class ParamHistoryInfo:
"""
Expand All @@ -66,6 +100,7 @@ def __init__(self, planned_value: Any, achieved_value: Any):
self.planned_value = planned_value
self.achieved_value = achieved_value


class PlanRequest:
"""
Represents a PlanRequest message received from ARES.
Expand All @@ -84,18 +119,85 @@ def __init__(self, parameters: list[PlanningParameter], settings: Dict[str, Any]
self.analysis_results = analysis_results
self.request_metadata = metadata

def __str__(self) -> str:
param_str = "\n ".join(self.parameter_names)
settings_str = "\n ".join([f"{k}: {v}" for k, v in self.settings.items()])
analysis_str = "\n ".join([f"{i}: {val}" for i, val in enumerate(self.analysis_results)])

metadata_str = str(self.request_metadata).replace('\n', '\n ')
return (f"PlanRequest object with:\n"
f"parameters:\n"
f" {param_str}\n"
f"settings:\n"
f" {settings_str}\n"
f"analysis_results:\n"
f" {analysis_str}\n"
f"request_metadata:\n"
f"{metadata_str}")

def __repr__(self) -> str:
return self.__str__()

def __getattr__(self, name):
"""
Allow easier retreival of parameters from the plan request
"""
if name in self.parameter_names:
return self.parameters[self.parameter_names.index(name)]
else:
print("Attribute not found: ", name)
return None

# Easier access to the data the planners will use
@property
def parameter_names(self) -> list[str]:
return [p.name for p in self.parameters]
@property
def planned_parameter_table(self) ->list:
return [p.planned_values for p in self.parameters]
@property
def acheived_parameter_table(self) ->list:
return [p.achieved_values for p in self.parameters]


class PlanResponse:
""" Represents a PlanResponse message to be send to ARES. """
def __init__(self, parameter_names: list[str], parameter_values: list, planning_outcome: Outcome = Outcome.SUCCESS, error_string: str = ""):
def __init__(self,
parameter_names: Optional[list[str]] = None,
parameter_values: Optional[list] = None,
parameter_data: Optional[dict[str,Any]] = None,
planning_outcome: Outcome = Outcome.SUCCESS,
error_string: str = ""):
"""
Initializes a PlanResponse.
Initializes a PlanResponse. Using either lists of names and values or a python dictonary of name:value pairs

Args:
parameter_names: A list of names associated with planned parameters.
parameter_values: A list of values associated with planned parameters.
parameter_data: A python dictionary of key:value pairs of planned parameters and planned values
"""
self.parameter_names = parameter_names
self.parameter_values = parameter_values
if parameter_data is not None:
self.parameter_names = list(parameter_data.keys())
self.parameter_values = list(parameter_data.values())

elif parameter_names is not None and parameter_values is not None:
if len(parameter_names) != len(parameter_values):
raise ValueError("Parameter names list and values lists must have the same length")
self.parameter_names = parameter_names
self.parameter_values = parameter_values

else:
raise ValueError("No values to assign!")

self.outcome = planning_outcome
self.error_string = error_string
self.error_string = error_string

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")

def __repr__(self) -> str:
return self.__str__()