From 2374b9a7eddadc2a3fa5c9cc8dd6dc0534b91485 Mon Sep 17 00:00:00 2001 From: Lance-Drane Date: Sun, 7 Jun 2026 12:07:51 -0400 Subject: [PATCH 01/12] #40 - add more return values to INTERSECT get_surrogate_values API Signed-off-by: Lance-Drane --- src/dial_dataclass/__init__.py | 1 + .../dial_dataclass_responses.py | 25 +- src/dial_service/dial_service.py | 691 +++++++++--------- 3 files changed, 381 insertions(+), 336 deletions(-) diff --git a/src/dial_dataclass/__init__.py b/src/dial_dataclass/__init__.py index ecda389..f1f2617 100644 --- a/src/dial_dataclass/__init__.py +++ b/src/dial_dataclass/__init__.py @@ -12,5 +12,6 @@ from .dial_dataclass_responses import ( DialDataResponse1D, DialDataResponse2D, + DialSurrogateValuesResponse, ) from .pydantic_helpers import ValidatedObjectId diff --git a/src/dial_dataclass/dial_dataclass_responses.py b/src/dial_dataclass/dial_dataclass_responses.py index 8d14972..b5d8118 100644 --- a/src/dial_dataclass/dial_dataclass_responses.py +++ b/src/dial_dataclass/dial_dataclass_responses.py @@ -1,7 +1,11 @@ -from pydantic import BaseModel +from typing import Annotated + +from pydantic import BaseModel, Field from .pydantic_helpers import ValidatedObjectId +PositiveIntType = Annotated[int, Field(ge=0)] + class DialDataResponse1D(BaseModel): """Possible response from DIAL""" @@ -19,3 +23,22 @@ class DialDataResponse2D(BaseModel): """Raw data""" workflow_id: ValidatedObjectId """The same workflow ID that was used to get the data, to facilitate possible load balancing.""" + + +class DialSurrogateValuesResponse(BaseModel): + """Response structure from calling get_surrogate_values()""" + + values: list[float] + """The computed values (for example, from Gaussian backends, the means) from calling get_surrogate_values()""" + transformed_stddevs: list[float] + """The computed uncertainties from calling get_surrogate_values(), with an inverse transform. If inverse-transforming is not possible (due to log-preprocessing), this will be all -1""" + stddevs: list[float] # TODO will probably remove in future + """The computed raw uncertainties from calling get_surrogate_values(), without an inverse transform""" + dim_x: int + """Number of dimensions of the associated data, derived from workflow""" + bounds: list[list[float]] + """Bounding box of the data, derived from workflow""" + points_to_predict: list[list[float]] + """Original list of points provided from the get_surrogate_values() input""" + workflow_id: ValidatedObjectId + """The same workflow ID that was used to get the data, to facilitate possible load balancing.""" diff --git a/src/dial_service/dial_service.py b/src/dial_service/dial_service.py index af5ce68..e925aa1 100644 --- a/src/dial_service/dial_service.py +++ b/src/dial_service/dial_service.py @@ -1,335 +1,356 @@ -import logging -import pickle -import traceback -from typing import Any - -from intersect_sdk import ( - IntersectBaseCapabilityImplementation, - IntersectCapabilityError, - intersect_message, - intersect_status, -) - -from dial_dataclass import ( - DialDataResponse1D, - DialDataResponse2D, - DialInputMultiple, - DialInputPredictions, - DialInputSingle, - DialWorkflowDatasetUpdate, - DialWorkflowDatasetUpdates, -) -from dial_dataclass.pydantic_helpers import ValidatedObjectId - -from . import core -from .mongo_handler import MongoDBCredentials, MongoDBHandler -from .serverside_data import ( - ServersideInputBase, - ServersideInputMultiple, - ServersideInputPrediction, - ServersideInputSingle, -) -from .service_specific_dataclasses import DialWorkflowCreationParamsService - -logger = logging.getLogger(__name__) - - -class DialCapabilityImplementation(IntersectBaseCapabilityImplementation): - """Internal guts for GP usage.""" - - intersect_sdk_capability_name = 'dial' - - def __init__(self, credentials: dict[str, Any]): - super().__init__() - self.mongo_handler = MongoDBHandler(MongoDBCredentials(**credentials)) - - ### STATEFUL + WORKFLOW FUNCTIONS ### - - @intersect_message() - def initialize_workflow(self, client_data: DialWorkflowCreationParamsService) -> str: - """Initializes a stateful workflow for DIALED. - - Takes in initial data points, and returns the ID of the associated workflow. - """ - try: - server_data = ServersideInputBase(client_data) - if client_data.dataset_x: - # the user provided some initial data, so train a model - model = pickle.dumps(core.train_model(server_data), protocol=5) - else: - # no initial data was provided, so just initialize a workflow ID and some common settings for the user - model = pickle.dumps(core.initialize_model(server_data), protocol=5) - workflow_id = self.mongo_handler.create_workflow(client_data.model_dump(), model) - except Exception: - logger.exception('initialize_workflow exception') - workflow_id = None - if not workflow_id: - msg = "Couldn't initialize workflow" - raise IntersectCapabilityError(msg) - return workflow_id - - @intersect_message() - def get_workflow_data(self, uuid: ValidatedObjectId) -> DialWorkflowCreationParamsService: - """Returns the current state of the workflow associated with the id""" - try: - db_result = self.mongo_handler.get_workflow(uuid) - except Exception: - logger.exception('get_workflow_data exception for %s', uuid) - db_result = None - if not db_result: - msg = f"Couldn't get workflow data with id {uuid}" - raise IntersectCapabilityError(msg) - return DialWorkflowCreationParamsService(**db_result) - - @intersect_message() - def update_workflow_with_data( - self, update_params: DialWorkflowDatasetUpdate - ) -> ValidatedObjectId: - """Updates the DB with the provided params. Success of operation is based off whether or not the INTERSECT response is an error.""" - - # TODO - all exceptions should realistically provide error information to the client. INTERSECT-SDK v0.9 will introduce a specific exception we can throw which will allow us to do this. - try: - db_get_result = self.mongo_handler.get_workflow(update_params.workflow_id) - except Exception: - logger.exception('update_workflow exception for %s', update_params.workflow_id) - db_get_result = None - if not db_get_result: - msg = f'Could not get workflow with id {update_params.workflow_id}' - raise IntersectCapabilityError(msg) - - try: - pretrain_result = DialWorkflowCreationParamsService(**db_get_result) - except Exception: - logger.exception( - 'update_workflow validation exception for %s', update_params.workflow_id - ) - pretrain_result = None - if not pretrain_result or ( - len(pretrain_result.dataset_x) > 0 - and len(pretrain_result.dataset_x[0]) != len(update_params.next_x) - ): - msg = f'Length mismatch in update function for workflow ID {update_params.workflow_id}' - raise IntersectCapabilityError(msg) - - try: - pretrain_result.dataset_x.append(update_params.next_x) - pretrain_result.dataset_y.append(update_params.next_y) - server_data = ServersideInputBase(pretrain_result) - - if update_params.backend_args is not None: - server_data.backend_args = update_params.backend_args - - if update_params.kernel_args is not None: - server_data.kernel_args = update_params.kernel_args - - if update_params.extra_args is not None: - server_data.extra_args = update_params.extra_args - - model = pickle.dumps(core.train_model(server_data), protocol=5) - - db_update_result = self.mongo_handler.update_workflow_dataset(update_params, model) - except Exception: - logger.exception('update_workflow exception for %s', update_params.workflow_id) - db_update_result = None - if not db_update_result: - msg = f"Couldn't update workflow with new data for workflow {update_params.workflow_id}" - raise IntersectCapabilityError(msg) - - return update_params.workflow_id - - @intersect_message() - def update_workflow_with_batch_data( - self, update_params: DialWorkflowDatasetUpdates - ) -> ValidatedObjectId: - try: - db_get_result = self.mongo_handler.get_workflow( - update_params.workflow_id, include_model=True - ) - except Exception: - logger.exception('update_workflow_with_batch_data init %s', update_params.workflow_id) - db_get_result = None - if not db_get_result: - exc = f'Could not get workflow with id {update_params.workflow_id}' - raise IntersectCapabilityError(exc) - - try: - pretrain = DialWorkflowCreationParamsService(**db_get_result) - except Exception: - logger.exception( - 'update_workflow_with_batch_data validation %s', update_params.workflow_id - ) - pretrain = None - if not pretrain: - exc = f'Workflow validation failed for {update_params.workflow_id}' - raise IntersectCapabilityError(exc) - - # shape check - expected_dim = ( - len(pretrain.dataset_x[0]) if pretrain.dataset_x else len(update_params.next_x_list[0]) - ) - for row in update_params.next_x_list: - if len(row) != expected_dim: - exc = 'Length mismatch in update function' - raise IntersectCapabilityError(exc) - - try: - pretrain.dataset_x.extend(update_params.next_x_list) - pretrain.dataset_y.extend(update_params.next_y_list) - server_data = ServersideInputBase(pretrain) - - if update_params.backend_args is not None: - server_data.backend_args = update_params.backend_args - if update_params.kernel_args is not None: - server_data.kernel_args = update_params.kernel_args - if update_params.extra_args is not None: - server_data.extra_args = update_params.extra_args - - model = pickle.dumps(core.train_model(server_data), protocol=5) - db_update_result = self.mongo_handler.update_workflow_dataset_batch( - update_params, model - ) - except Exception: - logger.exception( - 'update_workflow_with_batch_data training %s', update_params.workflow_id - ) - db_update_result = None - if not db_update_result: - exc = f"Couldn't update workflow with new batch data for {update_params.workflow_id}" - raise IntersectCapabilityError(exc) - - return update_params.workflow_id - - ### STATELESS FUNCTIONS ### - - @intersect_message() - # trains a model and then recommends a point to measure based on user's requested strategy: - def get_next_point(self, client_data: DialInputSingle) -> DialDataResponse1D: - """Trains a model, and then gets the next point for optimization based on the provided strategy. - - Args: - client_data (DialInputSingle): Input data containing bounds, strategy, and other parameters. - - Returns: - list[float]: The selected point for the next iteration. - """ - try: - workflow_state = self.mongo_handler.get_workflow(client_data.workflow_id) - except Exception: - logger.exception( - 'get_next_point exception (state initialization) for %s', client_data.workflow_id - ) - workflow_state = None - if not workflow_state: - msg = f'No workflow with id {client_data.workflow_id} exists' - raise IntersectCapabilityError(msg) - - try: - model = pickle.loads(workflow_state['model']) # noqa: S301 (XXX - this is technically trusted data as long as the DB hasn't been modified) - validated_state = DialWorkflowCreationParamsService(**workflow_state) - if client_data.extra_args: - if validated_state.extra_args: - validated_state.extra_args.update(client_data.extra_args) - else: - validated_state.extra_args = client_data.extra_args - data = ServersideInputSingle(validated_state, client_data) - - return_data = core.get_next_point(data, model) - return DialDataResponse1D( - data=return_data, - workflow_id=client_data.workflow_id, - ) - except Exception as err: - logger.exception( - 'get_next_point exception (primary logic) for %s', client_data.workflow_id - ) - raise IntersectCapabilityError(traceback.format_exc()) from err - - @intersect_message - def get_next_points(self, client_data: DialInputMultiple) -> DialDataResponse2D: - """ - Get multiple next points for optimization based on the provided strategy. - - Args: - client_data: Input data containing bounds, strategy, and other parameters. - - Returns: - list[list[float]]: A list of selected points for the next iteration. - """ - try: - workflow_state = self.mongo_handler.get_workflow(client_data.workflow_id) - except Exception: - logger.exception( - 'get_next_pointS exception (state initialization) for %s', client_data.workflow_id - ) - workflow_state = None - if not workflow_state: - msg = f'No workflow with id {client_data.workflow_id} exists' - raise IntersectCapabilityError(msg) - - try: - model = pickle.loads(workflow_state['model']) # noqa: S301 (XXX - this is technically trusted data as long as the DB hasn't been modified) - validated_state = DialWorkflowCreationParamsService(**workflow_state) - if client_data.extra_args: - if validated_state.extra_args: - validated_state.extra_args.update(client_data.extra_args) - else: - validated_state.extra_args = client_data.extra_args - data = ServersideInputMultiple(validated_state, client_data) - - return_data = core.get_next_points(data, model) - return DialDataResponse2D( - data=return_data, - workflow_id=client_data.workflow_id, - ) - except Exception as err: - logger.exception( - 'get_next_pointS exception (primary logic) for %s', client_data.workflow_id - ) - raise IntersectCapabilityError(traceback.format_exc()) from err - - @intersect_message - def get_surrogate_values(self, client_data: DialInputPredictions) -> DialDataResponse2D: - """Trains a model then returns 3 lists based on user-supplied points: - -Index 0: Predicted values. These are inverse transformed (undoing the preprocessing to put them on the same scale as dataset_y) - -Index 1: Inverse-transformed uncertainties. If inverse-transforming is not possible (due to log-preprocessing), this will be all -1 - -Index 2: Uncertainties without inverse transformation - """ - try: - workflow_state = self.mongo_handler.get_workflow( - client_data.workflow_id, include_model=True - ) - except Exception: - logger.exception( - 'get_surrogate_values exception (state initialization) for %s', - client_data.workflow_id, - ) - workflow_state = None - if not workflow_state: - msg = f'No workflow with id {client_data.workflow_id} exists' - raise IntersectCapabilityError(msg) - - try: - model = pickle.loads(workflow_state['model']) # noqa: S301 (XXX - this is technically trusted data as long as the DB hasn't been modified) - validated_state = DialWorkflowCreationParamsService(**workflow_state) - if client_data.extra_args: - if validated_state.extra_args: - validated_state.extra_args.update(client_data.extra_args) - else: - validated_state.extra_args = client_data.extra_args - data = ServersideInputPrediction(validated_state, client_data) - - return_data = core.get_surrogate_values(data, model) - return DialDataResponse2D( - data=return_data, - workflow_id=client_data.workflow_id, - ) - except Exception as err: - logger.exception( - 'get_surrogate_values exception (primary logic) for %s', client_data.workflow_id - ) - raise IntersectCapabilityError(traceback.format_exc()) from err - - @intersect_status() - def status(self) -> str: - """Basic status function which returns a hard-coded string.""" - return 'Up' +import logging +import pickle +import traceback +from typing import Any + +from intersect_sdk import ( + IntersectBaseCapabilityImplementation, + IntersectCapabilityError, + intersect_message, + intersect_status, +) + +from dial_dataclass import ( + DialDataResponse1D, + DialDataResponse2D, + DialInputMultiple, + DialInputPredictions, + DialInputSingle, + DialSurrogateValuesResponse, + DialWorkflowDatasetUpdate, + DialWorkflowDatasetUpdates, +) +from dial_dataclass.pydantic_helpers import ValidatedObjectId + +from . import core +from .mongo_handler import MongoDBCredentials, MongoDBHandler +from .serverside_data import ( + ServersideInputBase, + ServersideInputMultiple, + ServersideInputPrediction, + ServersideInputSingle, +) +from .service_specific_dataclasses import DialWorkflowCreationParamsService + +logger = logging.getLogger(__name__) + + +class DialCapabilityImplementation(IntersectBaseCapabilityImplementation): + """Internal guts for GP usage.""" + + intersect_sdk_capability_name = 'dial' + + def __init__(self, credentials: dict[str, Any]): + super().__init__() + self.mongo_handler = MongoDBHandler(MongoDBCredentials(**credentials)) + + ### STATEFUL + WORKFLOW FUNCTIONS ### + + @intersect_message() + def initialize_workflow(self, client_data: DialWorkflowCreationParamsService) -> str: + """Initializes a stateful workflow for DIALED. + + Takes in initial data points, and returns the ID of the associated workflow. + """ + try: + server_data = ServersideInputBase(client_data) + if client_data.dataset_x: + # the user provided some initial data, so train a model + model = pickle.dumps(core.train_model(server_data), protocol=5) + else: + # no initial data was provided, so just initialize a workflow ID and some common settings for the user + model = pickle.dumps(core.initialize_model(server_data), protocol=5) + workflow_id = self.mongo_handler.create_workflow(client_data.model_dump(), model) + except Exception: + logger.exception('initialize_workflow exception') + workflow_id = None + if not workflow_id: + msg = "Couldn't initialize workflow" + raise IntersectCapabilityError(msg) + return workflow_id + + @intersect_message() + def get_workflow_data(self, uuid: ValidatedObjectId) -> DialWorkflowCreationParamsService: + """Returns the current state of the workflow associated with the id""" + try: + db_result = self.mongo_handler.get_workflow(uuid) + except Exception: + logger.exception('get_workflow_data exception for %s', uuid) + db_result = None + if not db_result: + msg = f"Couldn't get workflow data with id {uuid}" + raise IntersectCapabilityError(msg) + return DialWorkflowCreationParamsService(**db_result) + + @intersect_message() + def update_workflow_with_data( + self, update_params: DialWorkflowDatasetUpdate + ) -> ValidatedObjectId: + """Updates the DB with the provided params. Success of operation is based off whether or not the INTERSECT response is an error.""" + + # TODO - all exceptions should realistically provide error information to the client. INTERSECT-SDK v0.9 will introduce a specific exception we can throw which will allow us to do this. + try: + db_get_result = self.mongo_handler.get_workflow(update_params.workflow_id) + except Exception: + logger.exception('update_workflow exception for %s', update_params.workflow_id) + db_get_result = None + if not db_get_result: + msg = f'Could not get workflow with id {update_params.workflow_id}' + raise IntersectCapabilityError(msg) + + try: + pretrain_result = DialWorkflowCreationParamsService(**db_get_result) + except Exception: + logger.exception( + 'update_workflow validation exception for %s', + update_params.workflow_id, + ) + pretrain_result = None + if not pretrain_result or ( + len(pretrain_result.dataset_x) > 0 + and len(pretrain_result.dataset_x[0]) != len(update_params.next_x) + ): + msg = f'Length mismatch in update function for workflow ID {update_params.workflow_id}' + raise IntersectCapabilityError(msg) + + try: + pretrain_result.dataset_x.append(update_params.next_x) + pretrain_result.dataset_y.append(update_params.next_y) + server_data = ServersideInputBase(pretrain_result) + + if update_params.backend_args is not None: + server_data.backend_args = update_params.backend_args + + if update_params.kernel_args is not None: + server_data.kernel_args = update_params.kernel_args + + if update_params.extra_args is not None: + server_data.extra_args = update_params.extra_args + + model = pickle.dumps(core.train_model(server_data), protocol=5) + + db_update_result = self.mongo_handler.update_workflow_dataset(update_params, model) + except Exception: + logger.exception('update_workflow exception for %s', update_params.workflow_id) + db_update_result = None + if not db_update_result: + msg = f"Couldn't update workflow with new data for workflow {update_params.workflow_id}" + raise IntersectCapabilityError(msg) + + return update_params.workflow_id + + @intersect_message() + def update_workflow_with_batch_data( + self, update_params: DialWorkflowDatasetUpdates + ) -> ValidatedObjectId: + try: + db_get_result = self.mongo_handler.get_workflow( + update_params.workflow_id, include_model=True + ) + except Exception: + logger.exception( + 'update_workflow_with_batch_data init %s', + update_params.workflow_id, + ) + db_get_result = None + if not db_get_result: + exc = f'Could not get workflow with id {update_params.workflow_id}' + raise IntersectCapabilityError(exc) + + try: + pretrain = DialWorkflowCreationParamsService(**db_get_result) + except Exception: + logger.exception( + 'update_workflow_with_batch_data validation %s', + update_params.workflow_id, + ) + pretrain = None + if not pretrain: + exc = f'Workflow validation failed for {update_params.workflow_id}' + raise IntersectCapabilityError(exc) + + # shape check + expected_dim = ( + len(pretrain.dataset_x[0]) if pretrain.dataset_x else len(update_params.next_x_list[0]) + ) + for row in update_params.next_x_list: + if len(row) != expected_dim: + exc = 'Length mismatch in update function' + raise IntersectCapabilityError(exc) + + try: + pretrain.dataset_x.extend(update_params.next_x_list) + pretrain.dataset_y.extend(update_params.next_y_list) + server_data = ServersideInputBase(pretrain) + + if update_params.backend_args is not None: + server_data.backend_args = update_params.backend_args + if update_params.kernel_args is not None: + server_data.kernel_args = update_params.kernel_args + if update_params.extra_args is not None: + server_data.extra_args = update_params.extra_args + + model = pickle.dumps(core.train_model(server_data), protocol=5) + db_update_result = self.mongo_handler.update_workflow_dataset_batch( + update_params, model + ) + except Exception: + logger.exception( + 'update_workflow_with_batch_data training %s', + update_params.workflow_id, + ) + db_update_result = None + if not db_update_result: + exc = f"Couldn't update workflow with new batch data for {update_params.workflow_id}" + raise IntersectCapabilityError(exc) + + return update_params.workflow_id + + ### STATELESS FUNCTIONS ### + + @intersect_message() + # trains a model and then recommends a point to measure based on user's requested strategy: + def get_next_point(self, client_data: DialInputSingle) -> DialDataResponse1D: + """Trains a model, and then gets the next point for optimization based on the provided strategy. + + Args: + client_data (DialInputSingle): Input data containing bounds, strategy, and other parameters. + + Returns: + list[float]: The selected point for the next iteration. + """ + try: + workflow_state = self.mongo_handler.get_workflow(client_data.workflow_id) + except Exception: + logger.exception( + 'get_next_point exception (state initialization) for %s', + client_data.workflow_id, + ) + workflow_state = None + if not workflow_state: + msg = f'No workflow with id {client_data.workflow_id} exists' + raise IntersectCapabilityError(msg) + + try: + model = pickle.loads(workflow_state['model']) # noqa: S301 (XXX - this is technically trusted data as long as the DB hasn't been modified) + validated_state = DialWorkflowCreationParamsService(**workflow_state) + if client_data.extra_args: + if validated_state.extra_args: + validated_state.extra_args.update(client_data.extra_args) + else: + validated_state.extra_args = client_data.extra_args + data = ServersideInputSingle(validated_state, client_data) + + return_data = core.get_next_point(data, model) + return DialDataResponse1D( + data=return_data, + workflow_id=client_data.workflow_id, + ) + except Exception as err: + logger.exception( + 'get_next_point exception (primary logic) for %s', + client_data.workflow_id, + ) + raise IntersectCapabilityError(traceback.format_exc()) from err + + @intersect_message + def get_next_points(self, client_data: DialInputMultiple) -> DialDataResponse2D: + """ + Get multiple next points for optimization based on the provided strategy. + + Args: + client_data: Input data containing bounds, strategy, and other parameters. + + Returns: + list[list[float]]: A list of selected points for the next iteration. + """ + try: + workflow_state = self.mongo_handler.get_workflow(client_data.workflow_id) + except Exception: + logger.exception( + 'get_next_pointS exception (state initialization) for %s', + client_data.workflow_id, + ) + workflow_state = None + if not workflow_state: + msg = f'No workflow with id {client_data.workflow_id} exists' + raise IntersectCapabilityError(msg) + + try: + model = pickle.loads(workflow_state['model']) # noqa: S301 (XXX - this is technically trusted data as long as the DB hasn't been modified) + validated_state = DialWorkflowCreationParamsService(**workflow_state) + if client_data.extra_args: + if validated_state.extra_args: + validated_state.extra_args.update(client_data.extra_args) + else: + validated_state.extra_args = client_data.extra_args + data = ServersideInputMultiple(validated_state, client_data) + + return_data = core.get_next_points(data, model) + return DialDataResponse2D( + data=return_data, + workflow_id=client_data.workflow_id, + ) + except Exception as err: + logger.exception( + 'get_next_pointS exception (primary logic) for %s', + client_data.workflow_id, + ) + raise IntersectCapabilityError(traceback.format_exc()) from err + + @intersect_message + def get_surrogate_values( + self, client_data: DialInputPredictions + ) -> DialSurrogateValuesResponse: + """Trains a model then returns 3 lists based on user-supplied points: + - Predicted values. These are inverse transformed (undoing the preprocessing to put them on the same scale as dataset_y) + - Inverse-transformed uncertainties. If inverse-transforming is not possible (due to log-preprocessing), this will be all -1 + - Uncertainties without inverse transformation + + Additional metadata is also returned in the response. + """ + try: + workflow_state = self.mongo_handler.get_workflow( + client_data.workflow_id, include_model=True + ) + except Exception: + logger.exception( + 'get_surrogate_values exception (state initialization) for %s', + client_data.workflow_id, + ) + workflow_state = None + if not workflow_state: + msg = f'No workflow with id {client_data.workflow_id} exists' + raise IntersectCapabilityError(msg) + + try: + model = pickle.loads(workflow_state['model']) # noqa: S301 (XXX - this is technically trusted data as long as the DB hasn't been modified) + validated_state = DialWorkflowCreationParamsService(**workflow_state) + if client_data.extra_args: + if validated_state.extra_args: + validated_state.extra_args.update(client_data.extra_args) + else: + validated_state.extra_args = client_data.extra_args + data = ServersideInputPrediction(validated_state, client_data) + + return_data = core.get_surrogate_values(data, model) + return DialSurrogateValuesResponse( + values=return_data[0], + transformed_stddevs=return_data[1], + stddevs=return_data[2], + dim_x=validated_state.dim_x, + points_to_predict=client_data.points_to_predict, + bounds=validated_state.bounds, + workflow_id=client_data.workflow_id, + ) + except Exception as err: + logger.exception( + 'get_surrogate_values exception (primary logic) for %s', + client_data.workflow_id, + ) + raise IntersectCapabilityError(traceback.format_exc()) from err + + @intersect_status() + def status(self) -> str: + """Basic status function which returns a hard-coded string.""" + return 'Up' From bb153dec8995cb8e5120a018ba5aac58a78f1201 Mon Sep 17 00:00:00 2001 From: Lance-Drane Date: Sun, 7 Jun 2026 12:10:11 -0400 Subject: [PATCH 02/12] temporarily publish expanded-output docker container Signed-off-by: Lance-Drane --- .github/workflows/docker-release.yml | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/.github/workflows/docker-release.yml b/.github/workflows/docker-release.yml index b3c3b09..cdf494d 100644 --- a/.github/workflows/docker-release.yml +++ b/.github/workflows/docker-release.yml @@ -5,8 +5,10 @@ on: branches: - develop - main + # TODO remove this once this branch is merged into develop + - expanded-output tags: - - 'v*.*.*' + - "v*.*.*" workflow_dispatch: permissions: From 072dab54dc9eee88282714b964f56ac88ba375e9 Mon Sep 17 00:00:00 2001 From: Viktor Reshniak Date: Sun, 7 Jun 2026 17:14:52 -0400 Subject: [PATCH 03/12] add grid strategy --- src/dial_service/core.py | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/src/dial_service/core.py b/src/dial_service/core.py index 43f8b3a..ff9935f 100644 --- a/src/dial_service/core.py +++ b/src/dial_service/core.py @@ -46,6 +46,18 @@ def get_next_point(data: ServersideInputSingle, model: Any) -> list[float]: return selected_point return random_in_bounds(data.bounds, data.numpy_rng) + if data.strategy == 'grid': + if hasattr(data, 'curr_index'): + data.curr_index += 1 + else: + data.curr_index = 0 + _measurement_grid = create_measurement_grid(data) + index = data.curr_index % len(_measurement_grid) + selected_point = _measurement_grid[index] + logger.debug('selected point with grid strategy') + logger.debug(selected_point) + return selected_point + backend = data.backend.lower() module = get_backend_module(backend) selected_point = module.sample(module, model, data) From 8bfdcec33f29ac7d2a642e670816773f0da58e36 Mon Sep 17 00:00:00 2001 From: Viktor Reshniak Date: Sun, 7 Jun 2026 17:17:38 -0400 Subject: [PATCH 04/12] renamed grid to hypercube strategy --- src/dial_service/core.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/dial_service/core.py b/src/dial_service/core.py index ff9935f..a410c89 100644 --- a/src/dial_service/core.py +++ b/src/dial_service/core.py @@ -46,7 +46,7 @@ def get_next_point(data: ServersideInputSingle, model: Any) -> list[float]: return selected_point return random_in_bounds(data.bounds, data.numpy_rng) - if data.strategy == 'grid': + if data.strategy == 'hypercube': if hasattr(data, 'curr_index'): data.curr_index += 1 else: @@ -54,7 +54,7 @@ def get_next_point(data: ServersideInputSingle, model: Any) -> list[float]: _measurement_grid = create_measurement_grid(data) index = data.curr_index % len(_measurement_grid) selected_point = _measurement_grid[index] - logger.debug('selected point with grid strategy') + logger.debug('selected point with hypercube strategy') logger.debug(selected_point) return selected_point From 21de31b6aa84f90a9e5bf98860b5dad7aa0965d4 Mon Sep 17 00:00:00 2001 From: Viktor Reshniak Date: Sun, 7 Jun 2026 17:28:57 -0400 Subject: [PATCH 05/12] fix for data.discrete_measurements = false --- src/dial_service/core.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/src/dial_service/core.py b/src/dial_service/core.py index a410c89..c3e2d29 100644 --- a/src/dial_service/core.py +++ b/src/dial_service/core.py @@ -47,13 +47,14 @@ def get_next_point(data: ServersideInputSingle, model: Any) -> list[float]: return random_in_bounds(data.bounds, data.numpy_rng) if data.strategy == 'hypercube': - if hasattr(data, 'curr_index'): - data.curr_index += 1 - else: + if not data.discrete_measurements: + data.discrete_measurement_grid_size = 10 + if not hasattr(data, 'curr_index'): data.curr_index = 0 _measurement_grid = create_measurement_grid(data) index = data.curr_index % len(_measurement_grid) selected_point = _measurement_grid[index] + data.curr_index += 1 logger.debug('selected point with hypercube strategy') logger.debug(selected_point) return selected_point From 4be4421d38712919bd258fa3601531fef3a380f3 Mon Sep 17 00:00:00 2001 From: Lance-Drane Date: Sun, 7 Jun 2026 18:08:34 -0400 Subject: [PATCH 06/12] change default hypercube discrete_measurement_size to [4], add tests Signed-off-by: Lance-Drane --- src/dial_dataclass/dial_dataclass.py | 545 ++++++++++++++------------- src/dial_service/core.py | 3 +- tests/unit/test_internals.py | 135 ++++++- 3 files changed, 406 insertions(+), 277 deletions(-) diff --git a/src/dial_dataclass/dial_dataclass.py b/src/dial_dataclass/dial_dataclass.py index b517e2a..7685645 100644 --- a/src/dial_dataclass/dial_dataclass.py +++ b/src/dial_dataclass/dial_dataclass.py @@ -1,272 +1,273 @@ -from typing import Annotated, Literal - -from pydantic import BaseModel, Field, field_validator - -from .pydantic_helpers import ValidatedObjectId - -PositiveIntType = Annotated[int, Field(ge=0)] - -_POSSIBLE_BACKENDS = ('sklearn', 'gpax', 'sable') - -BackendType = Literal[_POSSIBLE_BACKENDS] - - -class _DialWorkflowCreationParams(BaseModel): - """This comprises the information needed to create a DIAL workflow. - - This is a base class which should not be directly imported, clients should use "DialWorkflowCreationParamsClient" (in this file) and services should use "DialWorkflowCreationParamsService" (exported from the service) - """ - - dataset_x: Annotated[ - list[ - Annotated[ - list[float], - Field(description='Field lengths of all subarrays should be equal'), - ] - ], - Field(description='The input vectors of the training data'), - ] - dataset_y: Annotated[ - list[float], - Field( - description='The output values of the training data. Length should equal dataset_x', - ), - ] - y_is_good: Annotated[ - bool, - Field( - default=True, # <-- Set default here - description='If true, treat higher y values as better (e.g. y represents yield or profit). If false, opposite (e.g. y represents error or waste)', - ), - ] - kernel: Literal['rbf', 'matern', 'linear'] - bounds: list[ - Annotated[ - Annotated[list[float], Field(min_length=2, max_length=2)], - Field(min_length=2, max_length=2), - ] - ] - seed: Annotated[ - int, - Field( - default=-1, - ge=-1, - le=4294967295, - description='Specific RNG seed - use -1 to use system default', - ), - ] - dim_x: Annotated[int, Field(default=1)] - - preprocess_log: bool = Field(default=False) - preprocess_standardize: bool = Field(default=False) - - kernel_args: dict[str, float | int | bool | str | list[float] | tuple] | None = Field( - default=None - ) - """Additional arguments to provide alongside the kernel type.""" - backend_args: dict[str, float | int | bool | str | list[float] | tuple] | None = Field( - default=None - ) - """Additional arguments to provide alongside the backend type.""" - extra_args: dict[str, float | int | bool | str | list[float] | tuple] | None = Field( - default=None - ) - """Miscellaneous additional arguments.""" - - @field_validator('dataset_x') - @classmethod - def ensure_consistent_dataset_x_lengths(cls, x): - if len(x) < 2: - return x - target_length = len(x[0]) - for row in x[1:]: - if len(row) != target_length: - msg = 'Unequal vector lengths in dataset_x' - raise ValueError(msg) - return x - - # order rows as [low, high] - do NOT error out here, we can efficiently handle normalization - @field_validator('bounds') - @classmethod - def order_bounds(cls, bounds: list[list[float]]): - for row in bounds: - row.sort() - return bounds - - -# this class is specific to clients; they have no way of knowing which backends the Service supports, so we allow all of them -class DialWorkflowCreationParamsClient(_DialWorkflowCreationParams): - """Dataclass which clients can use to help verify requests to the DIAL microservice.""" - - backend: BackendType - - -class DialWorkflowDatasetUpdate(BaseModel): - workflow_id: ValidatedObjectId - next_x: list[float] = Field( - description='The next collection of X values you want to append to your overall data', - min_length=1, - ) - """the next collection of X values you want to append""" - next_y: float = Field(description='The next Y value you want to append to your overall data') - """the next Y value you want to append""" - kernel_args: dict[str, float | int | bool | str | list[float] | tuple] | None = Field( - default=None - ) - """Additional arguments to provide alongside the kernel type. These arguments will OVERRIDE prior saved arguments.""" - backend_args: dict[str, float | int | bool | str | list[float] | tuple] | None = Field( - default=None - ) - """Additional arguments to provide alongside the backend type. These arguments will OVERRIDE prior saved arguments.""" - extra_args: dict[str, float | int | bool | str | list[float] | tuple] | None = Field( - default=None - ) - """Miscellaneous additional arguments. These arguments will OVERRIDE prior saved arguments.""" - - -class DialWorkflowDatasetUpdates(BaseModel): - workflow_id: ValidatedObjectId - next_x_list: list[list[float]] = Field(min_length=1) - next_y_list: list[float] = Field(min_length=1) - kernel_args: dict[str, float | int | bool | str | list[float] | tuple] | None = None - backend_args: dict[str, float | int | bool | str | list[float] | tuple] | None = None - extra_args: dict[str, float | int | bool | str | list[float] | tuple] | None = None - - -class DialInputSingleConfidenceBound(BaseModel): - workflow_id: ValidatedObjectId - strategy: Literal['confidence_bound'] - strategy_args: dict[str, float | int | bool] | None = Field(default=None) - y_is_good: Annotated[ - bool, - Field( - default=True, # <-- Set default here - description='If true, treat higher y values as better (e.g. y represents yield or profit). If false, opposite (e.g. y represents error or waste)', - ), - ] - bounds: list[ - Annotated[ - Annotated[list[float], Field(min_length=2, max_length=2)], - Field(min_length=2, max_length=2), - ] - ] - extra_args: dict[str, float | int | bool | str | list[float] | tuple] | None = Field( - default=None - ) - """These extra arguments will be MERGED with the saved extra_args, with these arguments taking place over the saved values when applicable.""" - optimization_points: PositiveIntType = Field(default=1000) - confidence_bound: float = Field(gt=0.5, lt=1) - discrete_measurements: bool = Field(default=False) - discrete_measurement_grid_size: list[PositiveIntType] = Field(default=[20, 20]) - - -class DialInputSingleOtherStrategy(BaseModel): - workflow_id: ValidatedObjectId - strategy: Literal[ - 'random', - 'uncertainty', - 'expected_improvement', - 'upper_confidence_bound', - 'upper_confidence_bound_nomad', - 'polymer_acl_sampler', - ] - strategy_args: dict[str, float | int | bool] | None = Field(default=None) - y_is_good: Annotated[ - bool, - Field( - default=True, # <-- Set default here - description='If true, treat higher y values as better (e.g. y represents yield or profit). If false, opposite (e.g. y represents error or waste)', - ), - ] - bounds: list[ - Annotated[ - Annotated[list[float], Field(min_length=2, max_length=2)], - Field(min_length=2, max_length=2), - ] - ] - seed: Annotated[ - int, - Field( - default=-1, - ge=-1, - le=4294967295, - description='Specific RNG seed - use -1 to use system default', - ), - ] - extra_args: dict[str, float | int | bool | str | list[float] | tuple] | None = Field( - default=None - ) - """These extra arguments will be MERGED with the saved extra_args, with these arguments taking place over the saved values when applicable.""" - optimization_points: PositiveIntType = Field(default=1000) - discrete_measurements: bool = Field(default=False) - discrete_measurement_grid_size: list[PositiveIntType] = Field(default=[20, 20]) - - -DialInputSingle = Annotated[ - DialInputSingleConfidenceBound | DialInputSingleOtherStrategy, - Field( - discriminator='strategy', - description='This is the input dataclass for Dial for selecting a single new point to measure.', - ), -] - - -class DialInputMultipleOtherStrategy(BaseModel): - workflow_id: ValidatedObjectId - points: PositiveIntType - strategy: Literal[ - 'random', - 'uncertainty', - 'expected_improvement', - 'upper_confidence_bound', - 'upper_confidence_bound_nomad', - 'polymer_acl_sampler', - 'hypercube', - ] - strategy_args: dict[str, float | int | bool] | None = Field(default=None) - y_is_good: Annotated[ - bool, - Field( - default=True, # <-- Set default here - description='If true, treat higher y values as better (e.g. y represents yield or profit). If false, opposite (e.g. y represents error or waste)', - ), - ] - bounds: list[ - Annotated[ - Annotated[list[float], Field(min_length=2, max_length=2)], - Field(min_length=2, max_length=2), - ] - ] - seed: Annotated[ - int, - Field( - default=-1, - ge=-1, - le=4294967295, - description='Specific RNG seed - use -1 to use system default', - ), - ] - extra_args: dict[str, float | int | bool | str | list[float] | tuple] | None = Field( - default=None - ) - """These extra arguments will be MERGED with the saved extra_args, with these arguments taking place over the saved values when applicable.""" - optimization_points: PositiveIntType = Field(default=1000) - discrete_measurements: bool = Field(default=False) - discrete_measurement_grid_size: list[PositiveIntType] = Field(default=[20, 20]) - - -DialInputMultiple = Annotated[ - DialInputMultipleOtherStrategy, - Field(discriminator='strategy'), -] - - -class DialInputPredictions(BaseModel): - """This is the input dataclass for Dial for requesting a surrogate evaluation at a given number of points.""" - - workflow_id: ValidatedObjectId - points_to_predict: list[list[float]] - extra_args: dict[str, float | int | bool | str | list[float] | tuple] | None = Field( - default=None - ) - """These extra arguments will be MERGED with the saved extra_args, with these arguments taking place over the saved values when applicable.""" +from typing import Annotated, Literal + +from pydantic import BaseModel, Field, field_validator + +from .pydantic_helpers import ValidatedObjectId + +PositiveIntType = Annotated[int, Field(ge=0)] + +_POSSIBLE_BACKENDS = ('sklearn', 'gpax', 'sable') + +BackendType = Literal[_POSSIBLE_BACKENDS] + + +class _DialWorkflowCreationParams(BaseModel): + """This comprises the information needed to create a DIAL workflow. + + This is a base class which should not be directly imported, clients should use "DialWorkflowCreationParamsClient" (in this file) and services should use "DialWorkflowCreationParamsService" (exported from the service) + """ + + dataset_x: Annotated[ + list[ + Annotated[ + list[float], + Field(description='Field lengths of all subarrays should be equal'), + ] + ], + Field(description='The input vectors of the training data'), + ] + dataset_y: Annotated[ + list[float], + Field( + description='The output values of the training data. Length should equal dataset_x', + ), + ] + y_is_good: Annotated[ + bool, + Field( + default=True, # <-- Set default here + description='If true, treat higher y values as better (e.g. y represents yield or profit). If false, opposite (e.g. y represents error or waste)', + ), + ] + kernel: Literal['rbf', 'matern', 'linear'] + bounds: list[ + Annotated[ + Annotated[list[float], Field(min_length=2, max_length=2)], + Field(min_length=2, max_length=2), + ] + ] + seed: Annotated[ + int, + Field( + default=-1, + ge=-1, + le=4294967295, + description='Specific RNG seed - use -1 to use system default', + ), + ] + dim_x: Annotated[int, Field(default=1)] + + preprocess_log: bool = Field(default=False) + preprocess_standardize: bool = Field(default=False) + + kernel_args: dict[str, float | int | bool | str | list[float] | tuple] | None = Field( + default=None + ) + """Additional arguments to provide alongside the kernel type.""" + backend_args: dict[str, float | int | bool | str | list[float] | tuple] | None = Field( + default=None + ) + """Additional arguments to provide alongside the backend type.""" + extra_args: dict[str, float | int | bool | str | list[float] | tuple] | None = Field( + default=None + ) + """Miscellaneous additional arguments.""" + + @field_validator('dataset_x') + @classmethod + def ensure_consistent_dataset_x_lengths(cls, x): + if len(x) < 2: + return x + target_length = len(x[0]) + for row in x[1:]: + if len(row) != target_length: + msg = 'Unequal vector lengths in dataset_x' + raise ValueError(msg) + return x + + # order rows as [low, high] - do NOT error out here, we can efficiently handle normalization + @field_validator('bounds') + @classmethod + def order_bounds(cls, bounds: list[list[float]]): + for row in bounds: + row.sort() + return bounds + + +# this class is specific to clients; they have no way of knowing which backends the Service supports, so we allow all of them +class DialWorkflowCreationParamsClient(_DialWorkflowCreationParams): + """Dataclass which clients can use to help verify requests to the DIAL microservice.""" + + backend: BackendType + + +class DialWorkflowDatasetUpdate(BaseModel): + workflow_id: ValidatedObjectId + next_x: list[float] = Field( + description='The next collection of X values you want to append to your overall data', + min_length=1, + ) + """the next collection of X values you want to append""" + next_y: float = Field(description='The next Y value you want to append to your overall data') + """the next Y value you want to append""" + kernel_args: dict[str, float | int | bool | str | list[float] | tuple] | None = Field( + default=None + ) + """Additional arguments to provide alongside the kernel type. These arguments will OVERRIDE prior saved arguments.""" + backend_args: dict[str, float | int | bool | str | list[float] | tuple] | None = Field( + default=None + ) + """Additional arguments to provide alongside the backend type. These arguments will OVERRIDE prior saved arguments.""" + extra_args: dict[str, float | int | bool | str | list[float] | tuple] | None = Field( + default=None + ) + """Miscellaneous additional arguments. These arguments will OVERRIDE prior saved arguments.""" + + +class DialWorkflowDatasetUpdates(BaseModel): + workflow_id: ValidatedObjectId + next_x_list: list[list[float]] = Field(min_length=1) + next_y_list: list[float] = Field(min_length=1) + kernel_args: dict[str, float | int | bool | str | list[float] | tuple] | None = None + backend_args: dict[str, float | int | bool | str | list[float] | tuple] | None = None + extra_args: dict[str, float | int | bool | str | list[float] | tuple] | None = None + + +class DialInputSingleConfidenceBound(BaseModel): + workflow_id: ValidatedObjectId + strategy: Literal['confidence_bound'] + strategy_args: dict[str, float | int | bool] | None = Field(default=None) + y_is_good: Annotated[ + bool, + Field( + default=True, # <-- Set default here + description='If true, treat higher y values as better (e.g. y represents yield or profit). If false, opposite (e.g. y represents error or waste)', + ), + ] + bounds: list[ + Annotated[ + Annotated[list[float], Field(min_length=2, max_length=2)], + Field(min_length=2, max_length=2), + ] + ] + extra_args: dict[str, float | int | bool | str | list[float] | tuple] | None = Field( + default=None + ) + """These extra arguments will be MERGED with the saved extra_args, with these arguments taking place over the saved values when applicable.""" + optimization_points: PositiveIntType = Field(default=1000) + confidence_bound: float = Field(gt=0.5, lt=1) + discrete_measurements: bool = Field(default=False) + discrete_measurement_grid_size: list[PositiveIntType] = Field(default=[20, 20]) + + +class DialInputSingleOtherStrategy(BaseModel): + workflow_id: ValidatedObjectId + strategy: Literal[ + 'random', + 'hypercube', + 'uncertainty', + 'expected_improvement', + 'upper_confidence_bound', + 'upper_confidence_bound_nomad', + 'polymer_acl_sampler', + ] + strategy_args: dict[str, float | int | bool] | None = Field(default=None) + y_is_good: Annotated[ + bool, + Field( + default=True, # <-- Set default here + description='If true, treat higher y values as better (e.g. y represents yield or profit). If false, opposite (e.g. y represents error or waste)', + ), + ] + bounds: list[ + Annotated[ + Annotated[list[float], Field(min_length=2, max_length=2)], + Field(min_length=2, max_length=2), + ] + ] + seed: Annotated[ + int, + Field( + default=-1, + ge=-1, + le=4294967295, + description='Specific RNG seed - use -1 to use system default', + ), + ] + extra_args: dict[str, float | int | bool | str | list[float] | tuple] | None = Field( + default=None + ) + """These extra arguments will be MERGED with the saved extra_args, with these arguments taking place over the saved values when applicable.""" + optimization_points: PositiveIntType = Field(default=1000) + discrete_measurements: bool = Field(default=False) + discrete_measurement_grid_size: list[PositiveIntType] = Field(default=[20, 20]) + + +DialInputSingle = Annotated[ + DialInputSingleConfidenceBound | DialInputSingleOtherStrategy, + Field( + discriminator='strategy', + description='This is the input dataclass for Dial for selecting a single new point to measure.', + ), +] + + +class DialInputMultipleOtherStrategy(BaseModel): + workflow_id: ValidatedObjectId + points: PositiveIntType + strategy: Literal[ + 'random', + 'uncertainty', + 'expected_improvement', + 'upper_confidence_bound', + 'upper_confidence_bound_nomad', + 'polymer_acl_sampler', + 'hypercube', + ] + strategy_args: dict[str, float | int | bool] | None = Field(default=None) + y_is_good: Annotated[ + bool, + Field( + default=True, # <-- Set default here + description='If true, treat higher y values as better (e.g. y represents yield or profit). If false, opposite (e.g. y represents error or waste)', + ), + ] + bounds: list[ + Annotated[ + Annotated[list[float], Field(min_length=2, max_length=2)], + Field(min_length=2, max_length=2), + ] + ] + seed: Annotated[ + int, + Field( + default=-1, + ge=-1, + le=4294967295, + description='Specific RNG seed - use -1 to use system default', + ), + ] + extra_args: dict[str, float | int | bool | str | list[float] | tuple] | None = Field( + default=None + ) + """These extra arguments will be MERGED with the saved extra_args, with these arguments taking place over the saved values when applicable.""" + optimization_points: PositiveIntType = Field(default=1000) + discrete_measurements: bool = Field(default=False) + discrete_measurement_grid_size: list[PositiveIntType] = Field(default=[20, 20]) + + +DialInputMultiple = Annotated[ + DialInputMultipleOtherStrategy, + Field(discriminator='strategy'), +] + + +class DialInputPredictions(BaseModel): + """This is the input dataclass for Dial for requesting a surrogate evaluation at a given number of points.""" + + workflow_id: ValidatedObjectId + points_to_predict: list[list[float]] + extra_args: dict[str, float | int | bool | str | list[float] | tuple] | None = Field( + default=None + ) + """These extra arguments will be MERGED with the saved extra_args, with these arguments taking place over the saved values when applicable.""" diff --git a/src/dial_service/core.py b/src/dial_service/core.py index c3e2d29..a86d5da 100644 --- a/src/dial_service/core.py +++ b/src/dial_service/core.py @@ -48,7 +48,8 @@ def get_next_point(data: ServersideInputSingle, model: Any) -> list[float]: if data.strategy == 'hypercube': if not data.discrete_measurements: - data.discrete_measurement_grid_size = 10 + # use sane default + data.discrete_measurement_grid_size = [4] * data.dim_x if not hasattr(data, 'curr_index'): data.curr_index = 0 _measurement_grid = create_measurement_grid(data) diff --git a/tests/unit/test_internals.py b/tests/unit/test_internals.py index 13a5b76..c6b4957 100644 --- a/tests/unit/test_internals.py +++ b/tests/unit/test_internals.py @@ -87,7 +87,11 @@ def single_1D_discrete_grid(backend, strategy, strategy_args, discrete_measureme return ServersideInputSingle(workflow_state, params) -def single_2D(backend, strategy, strategy_args, discrete_measurement_grid_size=None): +def single_2D( + backend, + strategy, + strategy_args, +): workflow_state = DialWorkflowCreationParamsService( dataset_x=[ [0.9317758694133622, -0.23597335497782845], @@ -134,8 +138,69 @@ def single_2D(backend, strategy, strategy_args, discrete_measurement_grid_size=N strategy=strategy, strategy_args=strategy_args, bounds=[[-2, 2], [-2, 2]], - discrete_measurements=bool(discrete_measurement_grid_size), - discrete_measurement_grid_size=discrete_measurement_grid_size or [20, 20], + seed=42, + ) + return ServersideInputSingle(workflow_state, params) + + +def single_2D_discrete_grid( + backend, + strategy, + strategy_args, + discrete_measurement_grid_size, +): + bounds = [ + [0, discrete_measurement_grid_size[0] - 1], + [0, discrete_measurement_grid_size[1] - 1], + ] + workflow_state = DialWorkflowCreationParamsService( + dataset_x=[ + [0.9317758694133622, -0.23597335497782845], + [-0.7569874398003542, -0.76891211613756], + [-0.38457336507729645, -1.1327391183311766], + [-0.9293590899359039, 0.25039725076881014], + [1.984696498789749, -1.7147926093003538], + [1.2001856430453541, 1.572387611848939], + [0.5080666898409634, -1.566722183270571], + [-1.871124738716507, 1.9022651997285078], + [-1.572941300813352, 1.0014173171150125], + [0.033053333077524005, 0.44682040004191537], + ], + dim_x=2, + dataset_y=[ + 2.08609604, + 2.26284928, + 2.21989834, + 1.61634392, + 3.50481457, + 0.25065034, + 2.52277171, + 2.42139515, + 2.34930184, + 1.31811177, + ], + bounds=bounds, + kernel='rbf', + kernel_args={ + 'length_scale': 0.15, + 'length_scale_bounds': 'fixed', + 'noise_level': 0.0, + 'noise_level_bounds': 'fixed', + 'constant_value': 1.0, + 'constant_value_bounds': 'fixed', + }, + backend=backend, + preprocess_standardize=True, + y_is_good=True, + seed=42, + ) + params = DialInputSingleOtherStrategy( + workflow_id=DUMMY_WORKFLOW_ID, + strategy=strategy, + strategy_args=strategy_args, + bounds=bounds, + discrete_measurements=True, + discrete_measurement_grid_size=discrete_measurement_grid_size, seed=42, ) return ServersideInputSingle(workflow_state, params) @@ -391,6 +456,34 @@ def test_random(backend): assert 1 <= output[0] <= 2 +@pytest.mark.parametrize( + ('backend'), + [ + ('sklearn'), + ('gpax'), + ], +) +def test_hypercube_single_point(backend): + data = single_1D(backend, strategy='hypercube', strategy_args=None) + model = core.train_model(data) + output = None + last_point = None + DEFAULT_HYPERCUBE_VAL = ( + 4 # hypercube sets this parameter by default if no discrete grid provided + ) + for i in range(100): + last_point = output + output = core.get_next_point(data, model) + assert len(output) == 1 + output = output[0] + if last_point is not None: + if i % DEFAULT_HYPERCUBE_VAL != 0: + assert output > last_point + else: + assert last_point > output + assert 1 <= output <= 2 + + @pytest.mark.parametrize( ('backend'), [ @@ -416,6 +509,40 @@ def test_random_discrete(backend): assert output[0] in np.arange(60).astype(float) +@pytest.mark.parametrize( + ('backend'), + [ + ('sklearn'), + ('gpax'), + ], +) +def test_hypercube_single_point_discrete(backend): + data = single_1D_discrete_grid( + backend, + strategy='hypercube', + strategy_args=None, + discrete_measurement_grid_size=[60], + ) + model = core.train_model(data) + output = None + last_point = None + for i in range(100): + last_point = output + output = core.get_next_point(data, model) + assert len(output) == 1 + output = output[0] + if last_point is not None: + if i % data.discrete_measurement_grid_size[0] != 0: + assert output > last_point + else: + assert last_point > output + # assert output[0] in _measurement_grid + # assert output[0] == int(output[0]) + assert 0 <= output <= 59 + assert int(output) == output + assert output in np.arange(60).astype(float) + + @pytest.mark.parametrize( ('backend'), [ @@ -438,7 +565,7 @@ def test_random_points(backend): # ('gpax'), ], ) -def test_hypercube(backend): +def test_hypercube_multiple_points(backend): data = multiple_2D(backend, strategy='hypercube') model = core.initialize_model(data) points = core.get_next_points(data, model) From d0c11c00fcbb031bca34c8e64dc8e93456cf57f2 Mon Sep 17 00:00:00 2001 From: Lance-Drane Date: Sun, 7 Jun 2026 18:13:26 -0400 Subject: [PATCH 07/12] return current size of workflow dataset_x with all responses Signed-off-by: Lance-Drane --- src/dial_dataclass/dial_dataclass_responses.py | 6 ++++++ src/dial_service/dial_service.py | 3 +++ 2 files changed, 9 insertions(+) diff --git a/src/dial_dataclass/dial_dataclass_responses.py b/src/dial_dataclass/dial_dataclass_responses.py index b5d8118..4dcaa38 100644 --- a/src/dial_dataclass/dial_dataclass_responses.py +++ b/src/dial_dataclass/dial_dataclass_responses.py @@ -14,6 +14,8 @@ class DialDataResponse1D(BaseModel): """Raw data""" workflow_id: ValidatedObjectId """The same workflow ID that was used to get the data, to facilitate possible load balancing.""" + dataset_x_size: int + """Current length of dataset_x""" class DialDataResponse2D(BaseModel): @@ -23,6 +25,8 @@ class DialDataResponse2D(BaseModel): """Raw data""" workflow_id: ValidatedObjectId """The same workflow ID that was used to get the data, to facilitate possible load balancing.""" + dataset_x_size: int + """Current length of dataset_x""" class DialSurrogateValuesResponse(BaseModel): @@ -42,3 +46,5 @@ class DialSurrogateValuesResponse(BaseModel): """Original list of points provided from the get_surrogate_values() input""" workflow_id: ValidatedObjectId """The same workflow ID that was used to get the data, to facilitate possible load balancing.""" + dataset_x_size: int + """Current length of dataset_x""" diff --git a/src/dial_service/dial_service.py b/src/dial_service/dial_service.py index e925aa1..86f02e5 100644 --- a/src/dial_service/dial_service.py +++ b/src/dial_service/dial_service.py @@ -245,6 +245,7 @@ def get_next_point(self, client_data: DialInputSingle) -> DialDataResponse1D: return DialDataResponse1D( data=return_data, workflow_id=client_data.workflow_id, + dataset_x_size=len(validated_state.dataset_x), ) except Exception as err: logger.exception( @@ -290,6 +291,7 @@ def get_next_points(self, client_data: DialInputMultiple) -> DialDataResponse2D: return DialDataResponse2D( data=return_data, workflow_id=client_data.workflow_id, + dataset_x_size=len(validated_state.dataset_x), ) except Exception as err: logger.exception( @@ -342,6 +344,7 @@ def get_surrogate_values( points_to_predict=client_data.points_to_predict, bounds=validated_state.bounds, workflow_id=client_data.workflow_id, + dataset_x_size=len(validated_state.dataset_x), ) except Exception as err: logger.exception( From e81cd03fc95c7a02c3b26d00c7401b666020410f Mon Sep 17 00:00:00 2001 From: Lance-Drane Date: Sun, 7 Jun 2026 18:37:46 -0400 Subject: [PATCH 08/12] get_surrogate_values : return mean of transformed standard deviations Signed-off-by: Lance-Drane --- src/dial_dataclass/dial_dataclass_responses.py | 2 ++ src/dial_service/dial_service.py | 2 ++ 2 files changed, 4 insertions(+) diff --git a/src/dial_dataclass/dial_dataclass_responses.py b/src/dial_dataclass/dial_dataclass_responses.py index 4dcaa38..6a1ac19 100644 --- a/src/dial_dataclass/dial_dataclass_responses.py +++ b/src/dial_dataclass/dial_dataclass_responses.py @@ -48,3 +48,5 @@ class DialSurrogateValuesResponse(BaseModel): """The same workflow ID that was used to get the data, to facilitate possible load balancing.""" dataset_x_size: int """Current length of dataset_x""" + transformed_stddevs_avg: float + """the average of the transformed stddevs being returned""" diff --git a/src/dial_service/dial_service.py b/src/dial_service/dial_service.py index 86f02e5..71d334d 100644 --- a/src/dial_service/dial_service.py +++ b/src/dial_service/dial_service.py @@ -1,5 +1,6 @@ import logging import pickle +import statistics import traceback from typing import Any @@ -345,6 +346,7 @@ def get_surrogate_values( bounds=validated_state.bounds, workflow_id=client_data.workflow_id, dataset_x_size=len(validated_state.dataset_x), + transformed_stddevs_avg=statistics.mean(return_data[1]), ) except Exception as err: logger.exception( From c678d3daa30dcd8ed1ab4017115b2553b561bc3d Mon Sep 17 00:00:00 2001 From: Lance-Drane Date: Mon, 8 Jun 2026 10:36:15 -0400 Subject: [PATCH 09/12] allow specifying point_index for hypercube in get_next_point Signed-off-by: Lance-Drane --- src/dial_dataclass/dial_dataclass.py | 14 ++ src/dial_service/core.py | 5 +- src/dial_service/serverside_data.py | 339 ++++++++++++++------------- tests/unit/test_internals.py | 40 ++++ 4 files changed, 229 insertions(+), 169 deletions(-) diff --git a/src/dial_dataclass/dial_dataclass.py b/src/dial_dataclass/dial_dataclass.py index 7685645..689abbd 100644 --- a/src/dial_dataclass/dial_dataclass.py +++ b/src/dial_dataclass/dial_dataclass.py @@ -158,6 +158,13 @@ class DialInputSingleConfidenceBound(BaseModel): confidence_bound: float = Field(gt=0.5, lt=1) discrete_measurements: bool = Field(default=False) discrete_measurement_grid_size: list[PositiveIntType] = Field(default=[20, 20]) + point_index: Annotated[ + PositiveIntType, + Field( + default=0, + description='If using a strategy (i.e. "hypercube") which generates multiple points, get the specific point via the index (0-based) (default: 0)', + ), + ] class DialInputSingleOtherStrategy(BaseModel): @@ -201,6 +208,13 @@ class DialInputSingleOtherStrategy(BaseModel): optimization_points: PositiveIntType = Field(default=1000) discrete_measurements: bool = Field(default=False) discrete_measurement_grid_size: list[PositiveIntType] = Field(default=[20, 20]) + point_index: Annotated[ + PositiveIntType, + Field( + default=0, + description='If using a strategy (i.e. "hypercube") which generates multiple points, get the specific point via the index (0-based) (default: 0)', + ), + ] DialInputSingle = Annotated[ diff --git a/src/dial_service/core.py b/src/dial_service/core.py index a86d5da..dfddf4c 100644 --- a/src/dial_service/core.py +++ b/src/dial_service/core.py @@ -50,12 +50,9 @@ def get_next_point(data: ServersideInputSingle, model: Any) -> list[float]: if not data.discrete_measurements: # use sane default data.discrete_measurement_grid_size = [4] * data.dim_x - if not hasattr(data, 'curr_index'): - data.curr_index = 0 _measurement_grid = create_measurement_grid(data) - index = data.curr_index % len(_measurement_grid) + index = data.point_index % len(_measurement_grid) selected_point = _measurement_grid[index] - data.curr_index += 1 logger.debug('selected point with hypercube strategy') logger.debug(selected_point) return selected_point diff --git a/src/dial_service/serverside_data.py b/src/dial_service/serverside_data.py index 74405ec..78f0541 100644 --- a/src/dial_service/serverside_data.py +++ b/src/dial_service/serverside_data.py @@ -1,165 +1,174 @@ -from functools import cached_property - -import numpy as np - -from dial_dataclass import ( - DialInputMultiple, - DialInputPredictions, - DialInputSingle, -) - -from .service_specific_dataclasses import DialWorkflowCreationParamsService - - -# this is an extended version of ActiveLearningInputData. This allows us to add on properties and methods to this class without impacting the client side -class ServersideInputBase: - def __init__(self, data: DialWorkflowCreationParamsService): - self.X_raw = np.array(data.dataset_x) - self.Y_raw = np.array(data.dataset_y) - # it seems like there should be a smarter way to do this, but stuff involving loops doesn't work with static autocompleters: - self.bounds = data.bounds - self.y_is_good = data.y_is_good - self.kernel = data.kernel - self.backend: str = data.backend - self.seed = data.seed - self.numpy_rng = np.random.RandomState(None if data.seed == -1 else data.seed) - self.preprocess_log = data.preprocess_log - self.preprocess_standardize = data.preprocess_standardize - self.backend_args = data.backend_args - self.kernel_args = data.kernel_args - self.extra_args = data.extra_args - self.dim_x = data.dim_x - - @cached_property - def stddev(self) -> float: - return np.std(self.Y_train) - - @cached_property - def Y_best(self) -> float: - return self.Y_train.max() if self.y_is_good else self.Y_train.min() - - @cached_property - def Y_train(self) -> np.ndarray: - y = self.Y_raw - if self.preprocess_log: - y = np.log(y) - if self.preprocess_standardize: - y = (y - np.mean(y)) / np.std(y) - return y - - def _scale_X(self, X: np.ndarray) -> np.ndarray: - """ - Scale X into [0, 1]^D using self.bounds. - X: array of shape (N, D) - """ - X = np.asarray(X, dtype=float) - - if X.size == 0: - D = len(self.bounds) - return np.empty((0, D)) - - bounds = np.asarray(self.bounds, float) # (D, 2) - lows = bounds[:, 0] - highs = bounds[:, 1] - span = np.where(highs - lows == 0, 1.0, highs - lows) - - return (X - lows) / span - - @cached_property - def X_train(self) -> np.ndarray: - """ - Return X scaled to [0, 1] per dimension based on self.bounds. - - dataset_x: list[list[float]], shape (N, D) - bounds: list[[low, high], ...], shape (D, 2) - """ - return self._scale_X(self.X_raw) - - # undoes the preprocessing. - def inverse_transform(self, data: np.ndarray, is_stddev: bool = False): - if len(self.Y_raw) == 0: - return data - - # not possible to un-log the standard deviations (-1 +- 1 in log space != .1 +- 10 in realspace) - if self.preprocess_log and is_stddev: - return np.repeat(-1, len(data)) - if self.preprocess_standardize: - # the data that was used to calculate the standardization: - prestandardized_y = np.log(self.Y_raw) if self.preprocess_log else self.Y_raw - data = data * np.std(prestandardized_y) # not the same as *= (which is in-place) - if not is_stddev: - data = data + np.mean(prestandardized_y) - if self.preprocess_log: - data = np.exp(data) - return data - - -class ServersideInputSingle(ServersideInputBase): - def __init__(self, workflow_state: DialWorkflowCreationParamsService, params: DialInputSingle): - super().__init__(workflow_state) - self.strategy = params.strategy - self.strategy_args = params.strategy_args - self.y_is_good = params.y_is_good - self.bounds = params.bounds - self.numpy_rng = np.random.RandomState(None if params.seed == -1 else params.seed) - - self.optimization_points = params.optimization_points - self.confidence_bound = ( - params.confidence_bound if params.strategy == 'confidence_bound' else 0.0 - ) - self.discrete_measurements = params.discrete_measurements - self.discrete_measurement_grid_size = params.discrete_measurement_grid_size - - def set_x_predict(self, X_raw: np.ndarray) -> None: - """ - Store raw prediction points and their scaled version. - X_raw: shape (N, D) or (D,) for a single point. - """ - raw_vals = np.asarray(X_raw, dtype=float).reshape(-1, self.dim_x) - self.x_predict = self._scale_X(raw_vals) - - -class ServersideInputMultiple(ServersideInputBase): - def __init__( - self, workflow_state: DialWorkflowCreationParamsService, params: DialInputMultiple - ): - super().__init__(workflow_state) - self.strategy = params.strategy - self.points = params.points - self.strategy = params.strategy - self.strategy_args = params.strategy_args - self.y_is_good = params.y_is_good - self.bounds = params.bounds - self.numpy_rng = np.random.RandomState(None if params.seed == -1 else params.seed) - - self.optimization_points = params.optimization_points - self.confidence_bound = ( - params.confidence_bound if params.strategy == 'confidence_bound' else 0.0 - ) - self.discrete_measurements = params.discrete_measurements - self.discrete_measurement_grid_size = params.discrete_measurement_grid_size - - def set_x_predict(self, X_raw: np.ndarray) -> None: - """ - Store raw prediction points and their scaled version. - X_raw: shape (N, D) or (D,) for a single point. - """ - raw_vals = np.asarray(X_raw, dtype=float).reshape(-1, self.dim_x) - self.x_predict = self._scale_X(raw_vals) - - -class ServersideInputPrediction(ServersideInputBase): - def __init__( - self, workflow_state: DialWorkflowCreationParamsService, params: DialInputPredictions - ): - super().__init__(workflow_state) - self.x_predict_raw = np.asarray(params.points_to_predict, dtype=float) - self.set_x_predict(self.x_predict_raw) - - def set_x_predict(self, X_raw: np.ndarray) -> None: - """ - Store raw prediction points and their scaled version. - X_raw: shape (N, D) or (D,) for a single point. - """ - raw_vals = np.asarray(X_raw, dtype=float).reshape(-1, self.dim_x) - self.x_predict = self._scale_X(raw_vals) +from functools import cached_property + +import numpy as np + +from dial_dataclass import ( + DialInputMultiple, + DialInputPredictions, + DialInputSingle, +) + +from .service_specific_dataclasses import DialWorkflowCreationParamsService + + +# this is an extended version of ActiveLearningInputData. This allows us to add on properties and methods to this class without impacting the client side +class ServersideInputBase: + def __init__(self, data: DialWorkflowCreationParamsService): + self.X_raw = np.array(data.dataset_x) + self.Y_raw = np.array(data.dataset_y) + # it seems like there should be a smarter way to do this, but stuff involving loops doesn't work with static autocompleters: + self.bounds = data.bounds + self.y_is_good = data.y_is_good + self.kernel = data.kernel + self.backend: str = data.backend + self.seed = data.seed + self.numpy_rng = np.random.RandomState(None if data.seed == -1 else data.seed) + self.preprocess_log = data.preprocess_log + self.preprocess_standardize = data.preprocess_standardize + self.backend_args = data.backend_args + self.kernel_args = data.kernel_args + self.extra_args = data.extra_args + self.dim_x = data.dim_x + + @cached_property + def stddev(self) -> float: + return np.std(self.Y_train) + + @cached_property + def Y_best(self) -> float: + return self.Y_train.max() if self.y_is_good else self.Y_train.min() + + @cached_property + def Y_train(self) -> np.ndarray: + y = self.Y_raw + if self.preprocess_log: + y = np.log(y) + if self.preprocess_standardize: + y = (y - np.mean(y)) / np.std(y) + return y + + def _scale_X(self, X: np.ndarray) -> np.ndarray: + """ + Scale X into [0, 1]^D using self.bounds. + X: array of shape (N, D) + """ + X = np.asarray(X, dtype=float) + + if X.size == 0: + D = len(self.bounds) + return np.empty((0, D)) + + bounds = np.asarray(self.bounds, float) # (D, 2) + lows = bounds[:, 0] + highs = bounds[:, 1] + span = np.where(highs - lows == 0, 1.0, highs - lows) + + return (X - lows) / span + + @cached_property + def X_train(self) -> np.ndarray: + """ + Return X scaled to [0, 1] per dimension based on self.bounds. + + dataset_x: list[list[float]], shape (N, D) + bounds: list[[low, high], ...], shape (D, 2) + """ + return self._scale_X(self.X_raw) + + # undoes the preprocessing. + def inverse_transform(self, data: np.ndarray, is_stddev: bool = False): + if len(self.Y_raw) == 0: + return data + + # not possible to un-log the standard deviations (-1 +- 1 in log space != .1 +- 10 in realspace) + if self.preprocess_log and is_stddev: + return np.repeat(-1, len(data)) + if self.preprocess_standardize: + # the data that was used to calculate the standardization: + prestandardized_y = np.log(self.Y_raw) if self.preprocess_log else self.Y_raw + data = data * np.std(prestandardized_y) # not the same as *= (which is in-place) + if not is_stddev: + data = data + np.mean(prestandardized_y) + if self.preprocess_log: + data = np.exp(data) + return data + + +class ServersideInputSingle(ServersideInputBase): + def __init__( + self, + workflow_state: DialWorkflowCreationParamsService, + params: DialInputSingle, + ): + super().__init__(workflow_state) + self.strategy = params.strategy + self.strategy_args = params.strategy_args + self.y_is_good = params.y_is_good + self.bounds = params.bounds + self.numpy_rng = np.random.RandomState(None if params.seed == -1 else params.seed) + + self.optimization_points = params.optimization_points + self.confidence_bound = ( + params.confidence_bound if params.strategy == 'confidence_bound' else 0.0 + ) + self.discrete_measurements = params.discrete_measurements + self.discrete_measurement_grid_size = params.discrete_measurement_grid_size + self.point_index = params.point_index + + def set_x_predict(self, X_raw: np.ndarray) -> None: + """ + Store raw prediction points and their scaled version. + X_raw: shape (N, D) or (D,) for a single point. + """ + raw_vals = np.asarray(X_raw, dtype=float).reshape(-1, self.dim_x) + self.x_predict = self._scale_X(raw_vals) + + +class ServersideInputMultiple(ServersideInputBase): + def __init__( + self, + workflow_state: DialWorkflowCreationParamsService, + params: DialInputMultiple, + ): + super().__init__(workflow_state) + self.strategy = params.strategy + self.points = params.points + self.strategy = params.strategy + self.strategy_args = params.strategy_args + self.y_is_good = params.y_is_good + self.bounds = params.bounds + self.numpy_rng = np.random.RandomState(None if params.seed == -1 else params.seed) + + self.optimization_points = params.optimization_points + self.confidence_bound = ( + params.confidence_bound if params.strategy == 'confidence_bound' else 0.0 + ) + self.discrete_measurements = params.discrete_measurements + self.discrete_measurement_grid_size = params.discrete_measurement_grid_size + + def set_x_predict(self, X_raw: np.ndarray) -> None: + """ + Store raw prediction points and their scaled version. + X_raw: shape (N, D) or (D,) for a single point. + """ + raw_vals = np.asarray(X_raw, dtype=float).reshape(-1, self.dim_x) + self.x_predict = self._scale_X(raw_vals) + + +class ServersideInputPrediction(ServersideInputBase): + def __init__( + self, + workflow_state: DialWorkflowCreationParamsService, + params: DialInputPredictions, + ): + super().__init__(workflow_state) + self.x_predict_raw = np.asarray(params.points_to_predict, dtype=float) + self.set_x_predict(self.x_predict_raw) + + def set_x_predict(self, X_raw: np.ndarray) -> None: + """ + Store raw prediction points and their scaled version. + X_raw: shape (N, D) or (D,) for a single point. + """ + raw_vals = np.asarray(X_raw, dtype=float).reshape(-1, self.dim_x) + self.x_predict = self._scale_X(raw_vals) diff --git a/tests/unit/test_internals.py b/tests/unit/test_internals.py index c6b4957..1b2e20f 100644 --- a/tests/unit/test_internals.py +++ b/tests/unit/test_internals.py @@ -1,3 +1,4 @@ +import math from math import e as E_CONSTANT import numpy as np @@ -476,12 +477,14 @@ def test_hypercube_single_point(backend): output = core.get_next_point(data, model) assert len(output) == 1 output = output[0] + assert output != last_point if last_point is not None: if i % DEFAULT_HYPERCUBE_VAL != 0: assert output > last_point else: assert last_point > output assert 1 <= output <= 2 + data.point_index += 1 @pytest.mark.parametrize( @@ -531,6 +534,7 @@ def test_hypercube_single_point_discrete(backend): output = core.get_next_point(data, model) assert len(output) == 1 output = output[0] + assert output != last_point if last_point is not None: if i % data.discrete_measurement_grid_size[0] != 0: assert output > last_point @@ -541,6 +545,42 @@ def test_hypercube_single_point_discrete(backend): assert 0 <= output <= 59 assert int(output) == output assert output in np.arange(60).astype(float) + data.point_index += 1 + + +@pytest.mark.parametrize( + ('backend'), + [ + ('sklearn'), + ('gpax'), + ], +) +def test_hypercube_single_point_discrete_2D(backend): + data = single_2D_discrete_grid( + backend, + strategy='hypercube', + strategy_args=None, + discrete_measurement_grid_size=[6, 6], + ) + model = core.train_model(data) + output = None + last_point = None + num_points = math.prod(data.discrete_measurement_grid_size) + for i in range(100): + last_point = output + output = core.get_next_point(data, model) + assert len(output) == 2 + assert output != last_point + if last_point is not None: + if i % num_points != 0: + assert output[0] > last_point[0] or output[1] > last_point[1] + else: + assert last_point[0] > output[0] or last_point[1] > output[1] + for o in output: + assert 0 <= o <= 59 + assert int(o) == o + assert o in np.arange(60).astype(float) + data.point_index += 1 @pytest.mark.parametrize( From 41f4e04ca2be580eece7bf3e6de6f8a08d7c5867 Mon Sep 17 00:00:00 2001 From: Lance-Drane Date: Mon, 8 Jun 2026 11:47:24 -0400 Subject: [PATCH 10/12] return workflow_id in get_Workflow_data response Signed-off-by: Lance-Drane --- src/dial_dataclass/__init__.py | 1 + src/dial_dataclass/dial_dataclass_responses.py | 7 +++++++ src/dial_service/dial_service.py | 5 +++-- 3 files changed, 11 insertions(+), 2 deletions(-) diff --git a/src/dial_dataclass/__init__.py b/src/dial_dataclass/__init__.py index f1f2617..35377f2 100644 --- a/src/dial_dataclass/__init__.py +++ b/src/dial_dataclass/__init__.py @@ -13,5 +13,6 @@ DialDataResponse1D, DialDataResponse2D, DialSurrogateValuesResponse, + DialWorkflowFullState, ) from .pydantic_helpers import ValidatedObjectId diff --git a/src/dial_dataclass/dial_dataclass_responses.py b/src/dial_dataclass/dial_dataclass_responses.py index 6a1ac19..d319050 100644 --- a/src/dial_dataclass/dial_dataclass_responses.py +++ b/src/dial_dataclass/dial_dataclass_responses.py @@ -2,11 +2,18 @@ from pydantic import BaseModel, Field +from .dial_dataclass import DialWorkflowCreationParamsClient from .pydantic_helpers import ValidatedObjectId PositiveIntType = Annotated[int, Field(ge=0)] +class DialWorkflowFullState(DialWorkflowCreationParamsClient): + """Full state of the workflow.""" + + workflow_id: ValidatedObjectId + + class DialDataResponse1D(BaseModel): """Possible response from DIAL""" diff --git a/src/dial_service/dial_service.py b/src/dial_service/dial_service.py index 71d334d..511197e 100644 --- a/src/dial_service/dial_service.py +++ b/src/dial_service/dial_service.py @@ -20,6 +20,7 @@ DialSurrogateValuesResponse, DialWorkflowDatasetUpdate, DialWorkflowDatasetUpdates, + DialWorkflowFullState, ) from dial_dataclass.pydantic_helpers import ValidatedObjectId @@ -71,7 +72,7 @@ def initialize_workflow(self, client_data: DialWorkflowCreationParamsService) -> return workflow_id @intersect_message() - def get_workflow_data(self, uuid: ValidatedObjectId) -> DialWorkflowCreationParamsService: + def get_workflow_data(self, uuid: ValidatedObjectId) -> DialWorkflowFullState: """Returns the current state of the workflow associated with the id""" try: db_result = self.mongo_handler.get_workflow(uuid) @@ -81,7 +82,7 @@ def get_workflow_data(self, uuid: ValidatedObjectId) -> DialWorkflowCreationPara if not db_result: msg = f"Couldn't get workflow data with id {uuid}" raise IntersectCapabilityError(msg) - return DialWorkflowCreationParamsService(**db_result) + return DialWorkflowFullState(workflow_id=uuid, **db_result) @intersect_message() def update_workflow_with_data( From 7aa9e75d071fc23a7d4c4f163c72591f688775f2 Mon Sep 17 00:00:00 2001 From: Lance-Drane Date: Mon, 8 Jun 2026 17:24:20 -0400 Subject: [PATCH 11/12] add dataset_x_size to dataset_x Signed-off-by: Lance-Drane --- src/dial_dataclass/dial_dataclass_responses.py | 2 ++ src/dial_service/dial_service.py | 6 +++++- 2 files changed, 7 insertions(+), 1 deletion(-) diff --git a/src/dial_dataclass/dial_dataclass_responses.py b/src/dial_dataclass/dial_dataclass_responses.py index d319050..b7cc4f8 100644 --- a/src/dial_dataclass/dial_dataclass_responses.py +++ b/src/dial_dataclass/dial_dataclass_responses.py @@ -12,6 +12,8 @@ class DialWorkflowFullState(DialWorkflowCreationParamsClient): """Full state of the workflow.""" workflow_id: ValidatedObjectId + dataset_x_size: int + """Current length of dataset_x""" class DialDataResponse1D(BaseModel): diff --git a/src/dial_service/dial_service.py b/src/dial_service/dial_service.py index 511197e..3d8af21 100644 --- a/src/dial_service/dial_service.py +++ b/src/dial_service/dial_service.py @@ -82,7 +82,11 @@ def get_workflow_data(self, uuid: ValidatedObjectId) -> DialWorkflowFullState: if not db_result: msg = f"Couldn't get workflow data with id {uuid}" raise IntersectCapabilityError(msg) - return DialWorkflowFullState(workflow_id=uuid, **db_result) + return DialWorkflowFullState( + workflow_id=uuid, + dataset_x_size=len(db_result['dataset_x']), + **db_result, + ) @intersect_message() def update_workflow_with_data( From 2d8af2a4968114edb10681b4f56fdc64dc45b5ce Mon Sep 17 00:00:00 2001 From: Lance-Drane Date: Wed, 17 Jun 2026 16:10:16 -0400 Subject: [PATCH 12/12] get_surrogate_values: get average from numpy data also update the provided scripts Signed-off-by: Lance-Drane --- scripts/1d_sable_client.py | 13 ++++++------- scripts/1d_sinusoidal_growth_client.py | 13 ++++++------- scripts/2d_rosenbrock_client.py | 4 ++-- scripts/manual_client.py | 2 +- src/dial_service/core.py | 9 ++++++--- src/dial_service/dial_service.py | 3 +-- tests/benchmarks/test_strainmap.py | 2 +- tests/unit/test_internals.py | 2 +- 8 files changed, 24 insertions(+), 24 deletions(-) diff --git a/scripts/1d_sable_client.py b/scripts/1d_sable_client.py index 9660dbd..380d66c 100644 --- a/scripts/1d_sable_client.py +++ b/scripts/1d_sable_client.py @@ -215,17 +215,16 @@ def callback_message(self, operation: str, **kwargs) -> IntersectClientCallback: ) def handle_surrogate_values(self, payload): - response_data = payload['data'] + means = payload['values'] + transformed_stddevs = payload['transformed_stddevs'] if self.at_grids: - self.stddev_grid = np.array(response_data[1]).reshape( - (self.meshgrid_size,) * self.num_dims - ) - self.mean_grid = np.array(response_data[0]).reshape( + self.stddev_grid = np.array(transformed_stddevs).reshape( (self.meshgrid_size,) * self.num_dims ) + self.mean_grid = np.array(means).reshape((self.meshgrid_size,) * self.num_dims) else: - self.stddev_test = np.array(response_data[1]) - self.mean_test = np.array(response_data[0]) + self.stddev_test = np.array(transformed_stddevs) + self.mean_test = np.array(means) print( f'Values at testing points {self.x_test.reshape(-1)}: Mean: {self.mean_test}, Stddev: {self.stddev_test}' ) diff --git a/scripts/1d_sinusoidal_growth_client.py b/scripts/1d_sinusoidal_growth_client.py index 88bf688..a321e46 100644 --- a/scripts/1d_sinusoidal_growth_client.py +++ b/scripts/1d_sinusoidal_growth_client.py @@ -187,17 +187,16 @@ def callback_message(self, operation: str, **kwargs) -> IntersectClientCallback: ) def handle_surrogate_values(self, payload): - response_data = payload['data'] + means = payload['values'] + transformed_stddevs = payload['transformed_stddevs'] if self.at_grids: - self.variance_grid = np.array(response_data[1]).reshape( - (self.meshgrid_size,) * self.num_dims - ) - self.mean_grid = np.array(response_data[0]).reshape( + self.variance_grid = np.array(transformed_stddevs).reshape( (self.meshgrid_size,) * self.num_dims ) + self.mean_grid = np.array(means).reshape((self.meshgrid_size,) * self.num_dims) else: - self.variance_test = np.array(response_data[1]) - self.mean_test = np.array(response_data[0]) + self.variance_test = np.array(transformed_stddevs) + self.mean_test = np.array(means) print(f'Test Mean: {self.mean_test}, Variance: {self.variance_test}') # end of active learning loop after max_iter diff --git a/scripts/2d_rosenbrock_client.py b/scripts/2d_rosenbrock_client.py index 8abcb5b..af14344 100644 --- a/scripts/2d_rosenbrock_client.py +++ b/scripts/2d_rosenbrock_client.py @@ -208,8 +208,8 @@ def __call__( if ( operation == 'dial.get_surrogate_values' ): # if we receive a grid of surrogate values, record it for graphing, then ask for the next recommended point - data = payload['data'] - self.mean_grid = np.array(data[0]).reshape((MESHGRID_SIZE,) * NUM_DIMS) + means = payload['values'] + self.mean_grid = np.array(means).reshape((MESHGRID_SIZE,) * NUM_DIMS) return self.assemble_message('get_next_point') if operation == 'dial.get_next_point': diff --git a/scripts/manual_client.py b/scripts/manual_client.py index 8b436d9..b3e5a03 100644 --- a/scripts/manual_client.py +++ b/scripts/manual_client.py @@ -187,7 +187,7 @@ def __call__( if ( operation == 'dial.get_surrogate_values' ): # if we receive a grid of surrogate values, record it for graphing, then ask for the next recommended point - self.mean_grid = np.array(payload[0]).reshape(XX.shape) + self.mean_grid = np.array(payload['values']).reshape(XX.shape) return self.assemble_message('get_next_point') if operation == 'dial.get_next_point': # if we receive an EI recommendation, record it, show the user the current graph, and ask the user for the results of their experiment: diff --git a/src/dial_service/core.py b/src/dial_service/core.py index dfddf4c..61f52ba 100644 --- a/src/dial_service/core.py +++ b/src/dial_service/core.py @@ -96,7 +96,9 @@ def get_next_points(data: ServersideInputMultiple, model: Any) -> list[list[floa # pure functional implementation of message, without MongoDB calls -def get_surrogate_values(data: ServersideInputPrediction, model: Any) -> list[list[float]]: +def get_surrogate_values( + data: ServersideInputPrediction, model: Any +) -> tuple[list[float], list[float], list[float], float]: """ Get surrogate model predictions for given input points. @@ -106,14 +108,15 @@ def get_surrogate_values(data: ServersideInputPrediction, model: Any) -> list[li client_data (DialInputPredictions): Input data containing prediction points and model parameters. Returns: - list[list[float]]: A list containing means, transformed standard deviations, and raw standard deviations. + tuple[list[float], list[float], list[float], float]: A tuple containing means, transformed standard deviations, raw standard deviations, and a float value. """ backend = data.backend.lower() module = get_backend_module(backend) means, stddevs = module.predict(model, data) means = data.inverse_transform(means) transformed_stddevs = data.inverse_transform(stddevs, is_stddev=True) - return [means.tolist(), transformed_stddevs.tolist(), stddevs.tolist()] + average = np.sqrt(np.mean(np.asarray(transformed_stddevs) ** 2)) + return (means.tolist(), transformed_stddevs.tolist(), stddevs.tolist(), float(average)) def train_model(data: ServersideInputBase) -> Any: diff --git a/src/dial_service/dial_service.py b/src/dial_service/dial_service.py index 3d8af21..db8ea61 100644 --- a/src/dial_service/dial_service.py +++ b/src/dial_service/dial_service.py @@ -1,6 +1,5 @@ import logging import pickle -import statistics import traceback from typing import Any @@ -351,7 +350,7 @@ def get_surrogate_values( bounds=validated_state.bounds, workflow_id=client_data.workflow_id, dataset_x_size=len(validated_state.dataset_x), - transformed_stddevs_avg=statistics.mean(return_data[1]), + transformed_stddevs_avg=return_data[3], ) except Exception as err: logger.exception( diff --git a/tests/benchmarks/test_strainmap.py b/tests/benchmarks/test_strainmap.py index 2bca51c..0b3cd4b 100644 --- a/tests/benchmarks/test_strainmap.py +++ b/tests/benchmarks/test_strainmap.py @@ -255,7 +255,7 @@ def run_simulation( points_to_predict=INITIAL_POINTS_TO_PREDICT, ), ) - surrogate_mean, surrogate_std, _ = dial_core.get_surrogate_values(data, model) + surrogate_mean, surrogate_std, _, _ = dial_core.get_surrogate_values(data, model) mean_grid = np.array(surrogate_mean).reshape((-1, 1)) # subtract the true values and save mean absolute error and standard deviation diff --git a/tests/unit/test_internals.py b/tests/unit/test_internals.py index 1b2e20f..ca96d21 100644 --- a/tests/unit/test_internals.py +++ b/tests/unit/test_internals.py @@ -648,7 +648,7 @@ def test_hypercube_multiple_points(backend): def test_surrogate(backend, expected_means, expected_stddevs, expected_raw_stddevs): data = prediction_1D(backend) model = core.train_model(data) - means, stddevs, raw_stddevs = core.get_surrogate_values(data, model) + means, stddevs, raw_stddevs, _ = core.get_surrogate_values(data, model) assert means == pytest.approx(expected_means) assert stddevs[1:4] == pytest.approx(expected_stddevs) assert raw_stddevs[1:4] == pytest.approx(expected_raw_stddevs)