diff --git a/bec_lib/bec_lib/lmfit_serializer.py b/bec_lib/bec_lib/lmfit_serializer.py index dee651cf2..96bd26e57 100644 --- a/bec_lib/bec_lib/lmfit_serializer.py +++ b/bec_lib/bec_lib/lmfit_serializer.py @@ -52,18 +52,32 @@ def serialize_lmfit_params(params: Parameters) -> dict: return {v.name: serialize_param_object(v) for v in params} -def deserialize_param_object(obj: dict) -> Parameter: +def deserialize_param_object(obj: dict[str, dict | Parameter]) -> Parameters: """ Deserialize dictionary representation of lmfit.Parameter object. Args: - obj (dict): Dictionary representation of the parameters + obj (dict[str, dict | Parameter]): Dictionary representation of the parameters Returns: - Parameter: Parameter object + Parameters: Parameters object """ param = Parameters() for k, v in obj.items(): - v.pop("name") - param.add(k, **v) + if isinstance(v, Parameter): + param.add( + k, + value=v.value, + vary=v.vary, + min=v.min, + max=v.max, + expr=v.expr, + brute_step=v.brute_step, + ) + continue + if isinstance(v, dict): + v.pop("name", None) + v_copy = v.copy() + v_copy.pop("name", None) + param.add(k, **v_copy) return param diff --git a/bec_lib/tests/test_lmfit_serializer.py b/bec_lib/tests/test_lmfit_serializer.py index e73613d48..c5c487c6a 100644 --- a/bec_lib/tests/test_lmfit_serializer.py +++ b/bec_lib/tests/test_lmfit_serializer.py @@ -33,3 +33,19 @@ def test_serialize_lmfit_params(): obj = deserialize_param_object(result) assert obj == params + + # `name` is optional for deserialization (key is the param name) + result_without_names = { + k: {kk: vv for kk, vv in v.items() if kk != "name"} for k, v in result.items() + } + obj = deserialize_param_object(result_without_names) + assert obj == params + + +def test_deserialize_param_object_accepts_parameter_objects(): + params = lmfit.Parameters() + params.add("a", value=1.0, vary=True, min=-2.0, max=3.0) + params.add("b", value=2.0, vary=False) + + obj = deserialize_param_object({"a": params["a"], "b": params["b"]}) + assert obj == params diff --git a/bec_server/bec_server/data_processing/dap_service_manager.py b/bec_server/bec_server/data_processing/dap_service_manager.py index 257eee81d..93d0acf03 100644 --- a/bec_server/bec_server/data_processing/dap_service_manager.py +++ b/bec_server/bec_server/data_processing/dap_service_manager.py @@ -53,7 +53,6 @@ def process_dap_request(self, dap_request_msg: messages.DAPRequestMessage) -> No Args: dap_request_msg (DAPRequestMessage): DAPRequestMessage instance """ - logger.info(f"Processing dap request {dap_request_msg}") try: dap_cls = self._get_dap_cls(dap_request_msg) if not dap_cls: diff --git a/bec_server/bec_server/data_processing/lmfit1d_service.py b/bec_server/bec_server/data_processing/lmfit1d_service.py index 16967e44c..348aecc4d 100644 --- a/bec_server/bec_server/data_processing/lmfit1d_service.py +++ b/bec_server/bec_server/data_processing/lmfit1d_service.py @@ -3,6 +3,7 @@ import inspect import threading import time +from collections.abc import Iterable, Sequence import lmfit import numpy as np @@ -25,13 +26,15 @@ class LmfitService1D(DAPServiceBase): AUTO_FIT_SUPPORTED = True - def __init__(self, model: str, *args, continuous: bool = False, **kwargs): + def __init__( + self, model: str | list[str] | tuple[str, ...], *args, continuous: bool = False, **kwargs + ): """ Initialize the lmfit service. This is a multiplexer service that provides access to multiple lmfit models. Args: - model (str): Model name + model (str | list[str]): Model name or list of model names for a composite. continuous (bool, optional): Continuous processing. Defaults to False. """ super().__init__(*args, **kwargs) @@ -41,14 +44,267 @@ def __init__(self, model: str, *args, continuous: bool = False, **kwargs): self.device_y = None self.signal_y = None self.parameters = None + self.override_params = None + self.model_sequence: list[str] | None = None + self.model_name_sequence: list[str] | None = None + self.model_name_to_component: dict[str, str] | None = None + self._parameter_override_names = [] self.current_scan_item = None self.finished_id = None - self.model = getattr(lmfit.models, model)() + self.model_components: dict[str, lmfit.Model] | None = None + self.model_prefixes: dict[str, str] | None = None + self.model = self._build_model(model) self.finish_event = None self.data = None self.continuous = continuous self.oversample = 1 + def _build_model(self, model: str | list[str] | tuple[str, ...]) -> lmfit.Model: + if isinstance(model, (list, tuple)): + return self._build_composite_model(self._coerce_model_list(model)) + if isinstance(model, str): + return self._build_single_model(model) + raise ValueError(f"Unknown model {model}") + + def _build_single_model(self, model_name: str) -> lmfit.Model: + model_cls = self._get_model_cls(model_name) + return model_cls() + + def _build_composite_model(self, model_list: Sequence[str]) -> lmfit.Model: + if not model_list: + raise ValueError("Composite model list cannot be empty.") + self.model_components = {} + self.model_prefixes = {} + self.model_sequence = [] + self.model_name_sequence = list(model_list) + composite_model: lmfit.model.Model | None = None + for index, model_name in enumerate(model_list): + component_name = self._component_name(model_name, index) + component = self._create_component(model_name, component_name) + composite_model = component if composite_model is None else composite_model + component + self._build_component_lookup() + logger.debug( + f"Initialized composite lmfit model with components={list(self.model_components.keys())} " + f"prefixes={self.model_prefixes}" + ) + return composite_model + + @staticmethod + def _get_model_cls(model_name: str): + model_cls = getattr(lmfit.models, model_name, None) + if not model_cls: + raise ValueError(f"Unknown model {model_name}") + return model_cls + + @staticmethod + def _component_name(model_name: str, index: int) -> str: + return f"{model_name}_{index}" + + def _create_component(self, model_name: str, component_name: str): + model_cls = self._get_model_cls(model_name) + prefix = f"{component_name}_" + self.model_sequence.append(component_name) + self.model_prefixes[component_name] = prefix + component = model_cls(prefix=prefix) + self.model_components[component_name] = component + return component + + @staticmethod + def _coerce_model_list(model_list: Iterable[str]) -> list[str]: + return list(model_list) + + def _build_component_lookup(self) -> None: + unique_names = len(set(self.model_name_sequence)) == len(self.model_name_sequence) + if unique_names: + self.model_name_to_component = { + name: self.model_sequence[idx] for idx, name in enumerate(self.model_name_sequence) + } + else: + self.model_name_to_component = None + + def _expand_composite_parameters(self, parameters: dict | list | tuple) -> dict: + if self.model_components is None or self.model_prefixes is None: + return parameters + if isinstance(parameters, (list, tuple)): + return self._expand_composite_list(parameters) + if isinstance(parameters, dict): + return self._expand_composite_dict(parameters) + raise DAPError("Composite parameters must be a dict or list.") + + def _expand_composite_list(self, parameters: list | tuple) -> dict: + if self.model_sequence is None or len(parameters) != len(self.model_sequence): + raise DAPError( + "Composite parameters list must match the length of the composite model list." + ) + expanded: dict = {} + for index, param_map in enumerate(parameters): + if param_map is None: + continue + if not isinstance(param_map, dict): + raise DAPError( + f"Composite parameters list item {index} must be a dict of parameter overrides." + ) + component_name = self.model_sequence[index] + expanded.update(self._expand_param_map(component_name, param_map)) + return expanded + + def _expand_composite_dict(self, parameters: dict) -> dict: + expanded: dict = {} + component_keys = set(self.model_components.keys()) + if set(parameters.keys()).issubset(component_keys): + for component_name, param_map in parameters.items(): + if param_map is None: + continue + if not isinstance(param_map, dict): + raise DAPError( + f"Composite parameters for '{component_name}' must be a dict of parameter overrides." + ) + expanded.update(self._expand_param_map(component_name, param_map)) + return expanded + + component_map = self._resolve_model_name_map(parameters) + for model_name, param_map in parameters.items(): + if param_map is None: + continue + if not isinstance(param_map, dict): + raise DAPError( + f"Composite parameters for '{model_name}' must be a dict of parameter overrides." + ) + component_name = component_map[model_name] + expanded.update(self._expand_param_map(component_name, param_map)) + return expanded + + def _resolve_model_name_map(self, parameters: dict) -> dict[str, str]: + if self.model_name_to_component is None: + raise DAPError( + "Composite parameters are ambiguous with duplicate model names. " + "Use a list aligned to the model list or keys like 'ModelName_0'." + ) + invalid_models = set(parameters.keys()) - set(self.model_name_to_component.keys()) + if invalid_models: + raise DAPError( + f"Invalid parameter groups for composite model: {sorted(invalid_models)}" + ) + return self.model_name_to_component + + def _expand_param_map(self, component_name: str, param_map: dict) -> dict: + prefix = self.model_prefixes[component_name] + expanded = {} + for param_name, spec in param_map.items(): + expanded[f"{prefix}{param_name}"] = spec + return expanded + + def _guess_parameters(self, x: np.ndarray, y: np.ndarray) -> lmfit.Parameters: + guessed_params = self.model.make_params() + if self.model_components is not None: + for name, component in self.model_components.items(): + self._update_guess_from_component(guessed_params, component, name, x, y) + else: + self._update_guess_from_component(guessed_params, self.model, None, x, y) + self._log_guess(guessed_params) + return guessed_params + + @staticmethod + def _update_guess_from_component( + params: lmfit.Parameters, + component: lmfit.Model, + component_name: str | None, + x: np.ndarray, + y: np.ndarray, + ) -> None: + guess_fn = getattr(component, "guess", None) + if not callable(guess_fn): + return + try: + component_guess = guess_fn(y, x=x) + params.update(component_guess) + except Exception as guess_exc: + name = component_name or component.__class__.__name__ + logger.debug(f"lmfit guess failed for component={name}: {guess_exc}") + + def _log_guess(self, guessed_params: lmfit.Parameters) -> None: + logger.debug( + f"Using lmfit guess params for model={self.model.__class__.__name__}: " + f"{list(guessed_params.keys())}" + ) + logger.debug( + f"lmfit initial params for model={self.model.__class__.__name__}: " + f"{serialize_lmfit_params(guessed_params)}" + ) + + @staticmethod + def _apply_override_params( + params: lmfit.Parameters, overrides: lmfit.Parameters + ) -> lmfit.Parameters: + for name, override in overrides.items(): + params[name].set( + value=override.value, + vary=override.vary, + min=override.min, + max=override.max, + expr=override.expr, + brute_step=getattr(override, "brute_step", None), + ) + return params + + def _coerce_parameters(self, parameters: dict | list | tuple | lmfit.Parameters | None) -> dict: + raw_parameters: dict = {} + if not parameters: + return raw_parameters + if isinstance(parameters, lmfit.Parameters): + if self.model_components is not None: + raise DAPError( + "Composite models require parameters to be passed as a dict keyed by model name." + ) + raw_parameters.update({name: param for name, param in parameters.items()}) + return raw_parameters + if isinstance(parameters, (dict, list, tuple)): + if self.model_components is not None: + raw_parameters.update(self._expand_composite_parameters(parameters)) + elif isinstance(parameters, dict): + raw_parameters.update(parameters) + else: + raise DAPError("Non-dict parameters are only supported for composite models.") + return raw_parameters + raise DAPError( + f"Invalid parameters type {type(parameters)}. Expected dict or lmfit.Parameters." + ) + + def _filter_override_params(self, override_params: lmfit.Parameters) -> lmfit.Parameters: + if not override_params: + return override_params + param_names = set(getattr(self.model, "param_names", [])) + model_params = self.model.make_params() + model_param_names = set(model_params.keys()) + if not model_param_names: + return override_params + invalid_names = set(override_params.keys()) - model_param_names + derived_names = model_param_names - param_names + for name in list(override_params.keys()): + if name in invalid_names: + logger.warning( + f"Ignoring unknown lmfit parameter '{name}' for model '{self.model.__class__.__name__}'." + ) + override_params.pop(name, None) + elif name in derived_names: + logger.debug( + f"Ignoring derived lmfit parameter '{name}' for model '{self.model.__class__.__name__}'." + ) + override_params.pop(name, None) + return override_params + + def _build_parameters_from_overrides(self, overrides: lmfit.Parameters) -> lmfit.Parameters: + full_params = self.model.make_params() + return self._apply_override_params(full_params, overrides) + + def _prepare_fit_params(self, x: np.ndarray, y: np.ndarray) -> lmfit.Parameters: + if self.parameters is None: + return self._guess_parameters(x, y) + if self.override_params is not None and len(self.override_params) > 0: + guessed_params = self._guess_parameters(x, y) + return self._apply_override_params(guessed_params, self.override_params) + return self.parameters + @staticmethod def available_models(): models = [] @@ -123,7 +379,7 @@ def on_scan_status_update(self, status: dict, metadata: dict): Process a scan segment. Args: - data (dict): Scan segment data + status: (dict): Scan segment data metadata (dict): Scan segment metadata """ if self.finish_event is None: @@ -171,6 +427,7 @@ def configure( data_y: np.ndarray = None, x_min: float = None, x_max: float = None, + parameters: dict | list | None = None, amplitude: lmfit.Parameter = None, center: lmfit.Parameter = None, sigma: lmfit.Parameter = None, @@ -188,7 +445,9 @@ def configure( data_y (np.ndarray): Data for y instead of a scan item x_min (float): Minimum x value x_max (float): Maximum x value - parameters (dict): Fit parameters + parameters (dict | list): Fit parameters. For composite models, pass either + a list aligned to the model list (each item is a param dict), or + `{"ModelName": {"param": {...}}}` per model (unique model names only). oversample (int): Oversample factor """ # we only receive scan IDs from the client. However, users may @@ -197,15 +456,31 @@ def configure( self.oversample = oversample - self.parameters = {} + raw_parameters = self._coerce_parameters(parameters) if amplitude: - self.parameters["amplitude"] = amplitude + raw_parameters["amplitude"] = amplitude if center: - self.parameters["center"] = center + raw_parameters["center"] = center if sigma: - self.parameters["sigma"] = sigma + raw_parameters["sigma"] = sigma - self.parameters = deserialize_param_object(self.parameters) + override_params = deserialize_param_object(raw_parameters) + override_params = self._filter_override_params(override_params) + + self._parameter_override_names = list(override_params.keys()) + self.override_params = override_params + if len(override_params) > 0: + self.parameters = self._build_parameters_from_overrides(override_params) + logger.debug( + f"Configured lmfit model={self.model.__class__.__name__} with override_params={serialize_lmfit_params(override_params)}" + ) + else: + self.parameters = None + if parameters or amplitude or center or sigma: + logger.debug( + f"No usable lmfit parameter overrides after validation for model={self.model.__class__.__name__} " + f"(input_keys={list(raw_parameters.keys())})" + ) if data_x is not None and data_y is not None: self.data = { @@ -344,9 +619,12 @@ def get_data_from_current_scan( "scan_data": True, } - def process(self) -> tuple[dict, dict]: + def process(self) -> tuple[dict, dict] | None: """ Process data and return the result. + + Returns: + tuple[dict, dict]: Processed data and metadata if successful, None otherwise. """ # get the data if not self.data: @@ -356,10 +634,29 @@ def process(self) -> tuple[dict, dict]: y = self.data["y"] # fit the data + model_name = self.model.__class__.__name__ if self.parameters: - result = self.model.fit(y, x=x, params=self.parameters) + logger.debug( + f"Running lmfit fit: model={model_name} points={len(x)} fixed/override_params={self._parameter_override_names}" + ) else: - result = self.model.fit(y, x=x) + logger.debug(f"Running lmfit fit: model={model_name} points={len(x)} params=") + + try: + fit_params = self._prepare_fit_params(x, y) + result = self.model.fit(y, x=x, params=fit_params) + except Exception as exc: # pylint: disable=broad-except + if self.parameters is not None: + try: + params_str = serialize_lmfit_params(self.parameters) + except Exception as ser_exc: + params_str = f"" + else: + params_str = "" + logger.warning( + f"lmfit fit failed: model={model_name} points={len(x)} parameters={params_str} error={exc}" + ) + return # if the fit was only on a subset of the data, add the original x values to the output if self.data["x_lim"] or self.oversample != 1: @@ -388,6 +685,14 @@ def process(self) -> tuple[dict, dict]: metadata["input"] = {"parameters": serialize_lmfit_params(self.parameters)} metadata["fit_parameters"] = result.best_values metadata["fit_summary"] = result.summary() - logger.info(f"fit summary: {metadata['fit_summary']}") + logger.debug( + "fit summary: " + f"model={model_name} chi-square={result.chisqr:.6g} " + f"redchi={result.redchi:.6g} aic={result.aic:.6g} bic={result.bic:.6g}" + ) + if self.model_components is not None: + logger.debug( + f"Composite lmfit best params for model={model_name}: {metadata['fit_parameters']}" + ) return (stream_output, metadata) diff --git a/bec_server/tests/tests_data_processing/test_lmfit1d_service.py b/bec_server/tests/tests_data_processing/test_lmfit1d_service.py index 9e16ba0cb..8ea15aff9 100644 --- a/bec_server/tests/tests_data_processing/test_lmfit1d_service.py +++ b/bec_server/tests/tests_data_processing/test_lmfit1d_service.py @@ -19,7 +19,7 @@ def test_LmfitService1D(model, exists): if exists: service = LmfitService1D(model=model, client=client) return - with pytest.raises(AttributeError): + with pytest.raises(ValueError): service = LmfitService1D(model=model, client=client) @@ -104,12 +104,21 @@ def test_LmfitService1D_process(lmfit_service): "scan_data": True, } lmfit_service.model = mock.MagicMock() + lmfit_service.model.fit.return_value = mock.MagicMock( + best_fit=[4, 5, 6], + best_values={}, + summary=mock.MagicMock(return_value="summary"), + chisqr=1.0, + redchi=1.0, + aic=1.0, + bic=1.0, + ) result = lmfit_service.process() assert isinstance(result, tuple) assert isinstance(result[0], dict) assert isinstance(result[1], dict) - lmfit_service.model.fit.assert_called_once_with([4, 5, 6], x=[1, 2, 3]) + lmfit_service.model.fit.assert_called_once() def test_LmfitService1D_on_scan_status_update(lmfit_service): @@ -161,6 +170,54 @@ def test_LmfitService1D_configure_selected_devices(lmfit_service): get_data.assert_called_once() +def test_LmfitService1D_configure_accepts_generic_parameters_and_filters_invalid(lmfit_service): + x = np.linspace(-1.0, 1.0, 15) + y = np.exp(-(x**2)) + lmfit_service.configure( + data_x=x, + data_y=y, + parameters={"amplitude": {"value": 1.0, "vary": False}, "frequency": {"value": 2.0}}, + ) + assert lmfit_service.parameters["amplitude"].value == 1.0 + assert lmfit_service.parameters["amplitude"].vary is False + assert "frequency" not in lmfit_service.parameters + + +def test_LmfitService1D_configure_accepts_lmfit_parameters_object(lmfit_service): + x = np.linspace(-1.0, 1.0, 15) + y = np.exp(-(x**2)) + params = lmfit.models.GaussianModel().make_params() + params["amplitude"].set(value=1.0, vary=False) + lmfit_service.configure(data_x=x, data_y=y, parameters=params) + assert lmfit_service.parameters["amplitude"].value == 1.0 + assert lmfit_service.parameters["amplitude"].vary is False + + +def test_LmfitService1D_configure_invalid_parameters_type_raises(lmfit_service): + x = np.linspace(-1.0, 1.0, 15) + y = np.exp(-(x**2)) + with pytest.raises(DAPError): + lmfit_service.configure(data_x=x, data_y=y, parameters=["amplitude", 1.0]) # type: ignore[arg-type] + + +def test_LmfitService1D_configure_parameters_work_for_sine_model(): + if not hasattr(lmfit.models, "SineModel"): + pytest.skip("lmfit.models.SineModel not available in this environment") + service = LmfitService1D(model="SineModel", continuous=False, client=mock.MagicMock()) + x = np.linspace(0.0, 2.0 * np.pi, 25) + y = np.sin(x) + service.configure( + data_x=x, + data_y=y, + parameters={"frequency": {"value": 1.0, "vary": False}, "center": {"value": 0.0}}, + ) + assert service.parameters["frequency"].value == 1.0 + assert service.parameters["frequency"].vary is False + assert "center" not in service.parameters + assert "amplitude" in service.parameters + assert "shift" in service.parameters + + def test_LmfitService1D_get_model(lmfit_service): model = lmfit_service.get_model("GaussianModel") assert model.__name__ == "GaussianModel" @@ -168,3 +225,107 @@ def test_LmfitService1D_get_model(lmfit_service): with pytest.raises(ValueError): lmfit_service.get_model("ModelDoesntExist") + + +def test_LmfitService1D_composite_parameters_list_for_duplicate_models(): + client = mock.MagicMock() + service = LmfitService1D( + model=["GaussianModel", "GaussianModel", "GaussianModel"], client=client, continuous=False + ) + x = np.linspace(-3.0, 3.0, 50) + y = np.exp(-(x**2)) + service.configure( + data_x=x, + data_y=y, + parameters=[ + {"center": {"value": -1.0, "vary": True}}, + {"center": {"value": 0.0, "vary": True}}, + {"center": {"value": 1.0, "vary": True}}, + ], + ) + assert "GaussianModel_0_center" in service.parameters + assert "GaussianModel_1_center" in service.parameters + assert "GaussianModel_2_center" in service.parameters + + +def test_LmfitService1D_composite_parameters_dict_for_unique_models(): + client = mock.MagicMock() + service = LmfitService1D(model=["SineModel", "LinearModel"], client=client, continuous=False) + x = np.linspace(0.0, 2.0 * np.pi, 25) + y = np.sin(x) + service.configure( + data_x=x, + data_y=y, + parameters={ + "SineModel": {"frequency": {"value": 1.0, "vary": False}}, + "LinearModel": {"intercept": {"value": 0.0, "vary": True}}, + }, + ) + assert "SineModel_0_frequency" in service.parameters + assert "LinearModel_1_intercept" in service.parameters + + +def test_LmfitService1D_composite_parameters_dict_duplicate_models_rejected(): + client = mock.MagicMock() + service = LmfitService1D( + model=["GaussianModel", "GaussianModel"], client=client, continuous=False + ) + x = np.linspace(-1.0, 1.0, 15) + y = np.exp(-(x**2)) + with pytest.raises(DAPError): + service.configure( + data_x=x, data_y=y, parameters={"GaussianModel": {"center": {"value": 0.0}}} + ) + + +def test_LmfitService1D_non_composite_list_parameters_rejected(lmfit_service): + x = np.linspace(-1.0, 1.0, 15) + y = np.exp(-(x**2)) + with pytest.raises(DAPError): + lmfit_service.configure(data_x=x, data_y=y, parameters=[{"center": {"value": 0.0}}]) + + +def test_LmfitService1D_expand_composite_list_length_mismatch(): + client = mock.MagicMock() + service = LmfitService1D( + model=["GaussianModel", "GaussianModel"], client=client, continuous=False + ) + with pytest.raises(DAPError): + service._expand_composite_list([{"center": 0.0}]) # noqa: SLF001 + + +def test_LmfitService1D_expand_composite_dict_component_keys(): + client = mock.MagicMock() + service = LmfitService1D( + model=["GaussianModel", "GaussianModel"], client=client, continuous=False + ) + expanded = service._expand_composite_dict( # noqa: SLF001 + { + "GaussianModel_0": {"center": {"value": -1.0}}, + "GaussianModel_1": {"center": {"value": 1.0}}, + } + ) + assert "GaussianModel_0_center" in expanded + assert "GaussianModel_1_center" in expanded + + +def test_LmfitService1D_resolve_model_name_map_rejects_duplicates(): + client = mock.MagicMock() + service = LmfitService1D( + model=["GaussianModel", "GaussianModel"], client=client, continuous=False + ) + with pytest.raises(DAPError): + service._resolve_model_name_map({"GaussianModel": {"center": 0.0}}) # noqa: SLF001 + + +def test_LmfitService1D_prepare_fit_params_uses_guess(monkeypatch, lmfit_service): + x = np.linspace(-1.0, 1.0, 15) + y = np.exp(-(x**2)) + guessed = lmfit.models.GaussianModel().make_params() + guess_spy = mock.MagicMock(return_value=guessed) + monkeypatch.setattr(lmfit_service, "_guess_parameters", guess_spy) + lmfit_service.parameters = None + lmfit_service.override_params = None + params = lmfit_service._prepare_fit_params(x, y) # noqa: SLF001 + assert params is guessed + guess_spy.assert_called_once_with(x, y)