File size: 4,082 Bytes
2e0eaa2
 
 
 
 
 
 
f586f77
 
 
 
 
2e0eaa2
 
 
 
 
f586f77
 
 
2e0eaa2
 
 
 
 
 
 
 
 
f586f77
 
 
 
 
 
2e0eaa2
 
 
 
f586f77
 
 
2e0eaa2
 
 
 
 
 
 
f586f77
2e0eaa2
 
 
 
 
 
f586f77
 
2e0eaa2
 
 
 
 
 
 
 
 
 
 
f586f77
2e0eaa2
 
 
 
 
 
 
f586f77
 
 
2e0eaa2
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
f586f77
2e0eaa2
 
 
f586f77
2e0eaa2
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
from dataclasses import asdict

from sklearn.exceptions import NotFittedError
from sklearn.utils.validation import check_is_fitted

from logic import *


class UserError(Exception):
    """A mistake the student can fix; its message is shown in the UI."""


class Manager:
    def __init__(self):
        self._dataset: Dataset | None = None
        self._model: BaseEstimator | None = None
        self._model_config: ModelSettings | None = None
        # Maps plot coordinates back to inputs; None when the inputs are the plot coordinates
        self._projection: Projection | None = None
        self._input_names: list[str] = []

    def handle_set_dataset(self, dataset: dict) -> None:
        self._reset_model()

        try:
            parsed_dataset = Dataset(
                inputs=np.array(dataset["inputs"]),
                outputs=np.array(dataset["outputs"]),
            )
            input_names = [str(name) for name in dataset.get("input_names", [])]
            projection = dataset.get("projection")
            parsed_projection = None if projection is None else Projection(
                origin=np.array(projection["origin"]),
                components=np.array(projection["components"]),
            )
        except Exception as e:
            raise ValueError(f"Invalid dataset format: {e}")

        self._dataset = parsed_dataset
        self._projection = parsed_projection
        n_inputs = parsed_dataset.inputs.shape[1] if parsed_dataset.inputs.ndim == 2 else 0
        self._input_names = input_names or [f"x{i + 1}" for i in range(n_inputs)]

    def handle_set_model_config(self, settings: dict) -> None:
        self._reset_model()

        try:
            parsed_settings = ModelSettings(
                type=settings["type"],
                params=dict(settings.get("params", {})),
            )
        except Exception as e:
            raise ValueError(f"Invalid model settings format: {e}")

        self._model_config = parsed_settings

    def handle_build_model(self) -> dict | None:
        """Fit the model; returns its metrics and learned parameters (None with no data)."""
        self._reset_model()

        if self._dataset is None:
            raise ValueError("Dataset not set")
        if self._model_config is None:
            raise ValueError("Model settings not set")

        try:
            self._model = init_model(self._model_config)
        except Exception as e:
            # user error
            raise UserError(f"Error initializing model: {e}")

        if self._dataset.inputs.size > 0 and self._dataset.outputs.size > 0:
            try:
                train_model(self._model, self._dataset)
            except Exception as e:
                # user error
                self._reset_model()
                raise UserError(f"Error training model: {e}")
            return describe_model(self._model, self._dataset, self._input_names)
        return None

    def handle_get_decision_boundary(self, settings: dict) -> dict:
        if self._model is None:
            return dict(x=[], y=[], labels=[])

        try:
            check_is_fitted(self._model)
        except NotFittedError:
            return dict(x=[], y=[], labels=[])

        try:
            parsed_settings = DecisionBoundarySettings(
                xmin=settings["xmin"],
                xmax=settings["xmax"],
                ymin=settings["ymin"],
                ymax=settings["ymax"],
                resolution=settings.get("resolution", 500),
            )
        except Exception as e:
            raise ValueError(f"Invalid decision boundary settings format: {e}")

        try:
            result = get_decision_boundary_values(
                self._model, 
                parsed_settings, 
                self._projection.lift if self._projection else None,
            )
            
        except Exception as e:
            raise UserError(f"Error computing decision boundary: {e}")

        return asdict(result)

    def _reset_model(self) -> None:
        self._model = None
        print("Model reset")