Refactor to use manager in backend

backend/src/manager.py

@@ -0,0 +1,196 @@
+from pathlib import Path
+from typing import Literal
+
+import matplotlib.pyplot as plt
+from sympy import sympify
+
+from logic import (
+    DataGenerationOptions,
+    Dataset,
+    PlotData,
+    compute_plot_values,
+    generate_dataset,
+    load_dataset_from_csv,
+)
+
+
+class Manager:
+    def __init__(self) -> None:
+        self.dataset = Dataset(x=[], y=[])
+        self.plots_data: PlotData | None = None
+
+    def update_dataset(
+        self,
+        dataset_type: Literal["Generate", "CSV"],
+        function: str,
+        data_xmin: float,
+        data_xmax: float,
+        sigma: float,
+        nsample: int,
+        sample_method: Literal["Grid", "Random"],
+        csv_path: str | Path | None,
+        has_header: bool,
+        xcol: int,
+        ycol: int,
+    ) -> None:
+        if dataset_type == "Generate":
+            try:
+                parsed_function = sympify(function)
+            except Exception as exc:
+                raise ValueError(f"Invalid function: {exc}") from exc
+
+            sampling = sample_method.lower()
+            if sampling not in ["grid", "random"]:
+                raise ValueError(f"Unknown sampling method: {sample_method}")
+
+            self.dataset = generate_dataset(
+                parsed_function,
+                (data_xmin, data_xmax),
+                DataGenerationOptions(
+                    method=sampling,
+                    num_samples=nsample,
+                    noise=sigma,
+                ),
+            )
+            return
+
+        normalized_path = self._normalize_csv_path(csv_path)
+        if normalized_path is None:
+            raise ValueError("Please upload a CSV file.")
+
+        self.dataset = load_dataset_from_csv(
+            normalized_path,
+            has_header,
+            xcol,
+            ycol,
+        )
+
+    def compute_plot_data(
+        self,
+        kernel: str,
+        distribution: Literal["Prior", "Posterior"],
+        plot_xmin: float,
+        plot_xmax: float,
+    ) -> None:
+        self.plots_data = compute_plot_values(
+            self.dataset,
+            kernel,
+            distribution,
+            plot_xmin,
+            plot_xmax,
+        )
+
+    def handle_generate_plots(
+        self,
+        dataset_type: Literal["Generate", "CSV"],
+        function: str,
+        data_xmin: float,
+        data_xmax: float,
+        sigma: float,
+        nsample: int,
+        sample_method: Literal["Grid", "Random"],
+        csv_path: str | Path | None,
+        has_header: bool,
+        xcol: int,
+        ycol: int,
+        kernel: str,
+        distribution: Literal["Prior", "Posterior"],
+        plot_xmin: float,
+        plot_xmax: float,
+    ):
+        self.update_dataset(
+            dataset_type,
+            function,
+            data_xmin,
+            data_xmax,
+            sigma,
+            nsample,
+            sample_method,
+            csv_path,
+            has_header,
+            xcol,
+            ycol,
+        )
+
+        true_dataset = self._build_true_dataset(
+            dataset_type,
+            function,
+            plot_xmin,
+            plot_xmax,
+        )
+
+        self.compute_plot_data(
+            kernel,
+            distribution,
+            plot_xmin,
+            plot_xmax,
+        )
+
+        return self.generate_plot(true_dataset)
+
+    def generate_plot(self, true_dataset: Dataset):
+        if self.plots_data is None:
+            raise ValueError("Plot data has not been computed.")
+
+        fig, ax = plt.subplots(figsize=(12, 9))
+        cmap = plt.get_cmap("tab20")
+
+        ax.scatter(self.dataset.x, self.dataset.y, color=cmap(0), label="Data Points")
+
+        if true_dataset.y is not None and len(true_dataset.y) > 0:
+            ax.plot(true_dataset.x, true_dataset.y, color=cmap(1), label="True Function")
+
+        ax.plot(self.plots_data.x, self.plots_data.pred_mean, color=cmap(2), label="Mean Prediction")
+
+        ax.fill_between(
+            self.plots_data.x,
+            self.plots_data.pred_mean - 1.96 * self.plots_data.pred_std,
+            self.plots_data.pred_mean + 1.96 * self.plots_data.pred_std,
+            color=cmap(3),
+            alpha=0.2,
+            label="95% Confidence Interval",
+        )
+
+        ax.legend()
+        return fig
+
+    def _build_true_dataset(
+        self,
+        dataset_type: Literal["Generate", "CSV"],
+        function: str,
+        xmin: float,
+        xmax: float,
+    ) -> Dataset:
+        if dataset_type == "CSV":
+            return Dataset(x=[], y=[])
+
+        try:
+            parsed_function = sympify(function)
+        except Exception as exc:
+            raise ValueError(f"Invalid function: {exc}") from exc
+
+        return generate_dataset(
+            parsed_function,
+            (xmin, xmax),
+            DataGenerationOptions(
+                method="grid",
+                num_samples=1000,
+                noise=0.0,
+            ),
+        )
+
+    def _normalize_csv_path(self, csv_path: str | Path | None) -> str | None:
+        if csv_path is None:
+            return None
+
+        if isinstance(csv_path, Path):
+            return str(csv_path)
+
+        if isinstance(csv_path, str):
+            return csv_path
+
+        name = getattr(csv_path, "name", None)
+        if name:
+            return str(name)
+
+        return None

frontends/gradio/main.py

@@ -1,24 +1,16 @@
 from typing import Literal
 
 import gradio as gr
-import matplotlib.pyplot as plt
-from sympy import sympify
 
 import sys
 from pathlib import Path
+
 root_dir = Path(__file__).resolve().parent.parent.parent
 backend_src = root_dir / "backend" / "src"
 if str(backend_src) not in sys.path:
     sys.path.append(str(backend_src))
 
-from backend import (
-    Dataset, 
-    DataGenerationOptions,
-    PlotData,
-    compute_plot_values,
-    generate_dataset, 
-    load_dataset_from_csv, 
-)
+from manager import Manager
 
 
 CSS = """
@@ -28,107 +20,8 @@ CSS = """
 """
 
 
-def get_dataset(
-    dataset_type: Literal["Generate", "CSV"],
-    function: str,
-    xmin: float,
-    xmax: float,
-    sigma: float,
-    nsample: int,
-    sample_method: Literal["Grid", "Random"],
-    csv_path: str,
-    has_header: bool,
-    xcol: int,
-    ycol: int,
-) -> tuple[Dataset, Dataset]:
-    if dataset_type == "Generate":
-        try:
-            function = sympify(function)
-        except Exception as e:
-            raise ValueError(f"Invalid function: {e}")
-
-        sample_method = sample_method.lower()
-        if sample_method not in ['grid', 'random']:
-            raise ValueError(f"Unknown sampling method: {sample_method}")
-        
-        dataset = generate_dataset(
-            function,
-            (xmin, xmax),
-            DataGenerationOptions(
-                sample_method,
-                nsample,
-                sigma,
-            ),
-        )
-
-    else:  # CSV
-        dataset = load_dataset_from_csv(
-            csv_path,
-            has_header,
-            xcol,
-            ycol,
-        )
-
-    return dataset
-
-
-def get_true_dataset(
-    dataset_type: Literal["Generate", "CSV"],
-    function: str,
-    xmin: float,
-    xmax: float,
-) -> Dataset:
-    if dataset_type == "CSV":
-        # No true dataset for CSV uploads
-        return Dataset(x=[], y=[])
-    else:
-        try:
-            function = sympify(function)
-        except Exception as e:
-            raise ValueError(f"Invalid function: {e}")
-
-        true_dataset = generate_dataset(
-            function,
-            (xmin, xmax),
-            DataGenerationOptions(
-                method='grid',
-                num_samples=1000,
-                noise=0.,
-            ),
-        )
-        return true_dataset
-
-
-def generate_plot(
-    plot_data: PlotData,
-    dataset: Dataset,
-    true_dataset: Dataset,
-):
-    fig, ax = plt.subplots(figsize=(12, 9))
-    cmap = plt.get_cmap("tab20")
-
-    ax.scatter(dataset.x, dataset.y, color=cmap(0), label='Data Points')
-
-    if true_dataset.y is not None:
-        ax.plot(true_dataset.x, true_dataset.y, color=cmap(1), label='True Function')
-
-    ax.plot(plot_data.x, plot_data.pred_mean, color=cmap(2), label='Mean Prediction')
-
-    ax.fill_between(
-        plot_data.x,
-        plot_data.pred_mean - 1.96 * plot_data.pred_std,
-        plot_data.pred_mean + 1.96 * plot_data.pred_std,
-        color=cmap(3),
-        alpha=0.2,
-        label='95% Confidence Interval'
-    )
-
-    ax.legend()
-    return fig
-
-
 def handle_dataset_type_change(
-    dataset_type: Literal["Generate", "CSV"]
+    dataset_type: Literal["Generate", "CSV"],
 ):
     if dataset_type == "Generate":
         return (
@@ -143,27 +36,65 @@ def handle_dataset_type_change(
             gr.update(visible=False),  # xcol
             gr.update(visible=False),  # ycol
         )
-    else:  # CSV
-        return (
-            gr.update(visible=False),  # function
-            gr.update(visible=False),  # xmin
-            gr.update(visible=False),  # xmax
-            gr.update(visible=False),  # sigma
-            gr.update(visible=False),  # nsample
-            gr.update(visible=False),  # sample method
-            gr.update(visible=True),  # csv upload
-            gr.update(visible=True),  # has header
-            gr.update(visible=True),  # xcol
-            gr.update(visible=True),  # ycol
-        )
+
+    return (
+        gr.update(visible=False),  # function
+        gr.update(visible=False),  # xmin
+        gr.update(visible=False),  # xmax
+        gr.update(visible=False),  # sigma
+        gr.update(visible=False),  # nsample
+        gr.update(visible=False),  # sample method
+        gr.update(visible=True),  # csv upload
+        gr.update(visible=True),  # has header
+        gr.update(visible=True),  # xcol
+        gr.update(visible=True),  # ycol
+    )
+
+
+def handle_generate(
+    manager: Manager,
+    dataset_type: Literal["Generate", "CSV"],
+    function: str,
+    data_xmin: float,
+    data_xmax: float,
+    sigma: float,
+    nsample: int,
+    sample_method: Literal["Grid", "Random"],
+    csv_upload,
+    has_header: bool,
+    xcol: int,
+    ycol: int,
+    kernel: str,
+    distribution: Literal["Prior", "Posterior"],
+    plot_xmin: float,
+    plot_xmax: float,
+):
+    plot = manager.handle_generate_plots(
+        dataset_type,
+        function,
+        data_xmin,
+        data_xmax,
+        sigma,
+        nsample,
+        sample_method,
+        csv_upload,
+        has_header,
+        xcol,
+        ycol,
+        kernel,
+        distribution,
+        plot_xmin,
+        plot_xmax,
+    )
+    return manager, plot
 
 
 def launch():
     default_dataset_type = "Generate"
 
     default_function = "sin(2 * pi * x)"
-    default_data_xmin = -1.
-    default_data_xmax = 1.
+    default_data_xmin = -1.0
+    default_data_xmax = 1.0
     default_sigma = 0
     default_nsample = 100
     default_sample_method = "Grid"
@@ -175,56 +106,36 @@ def launch():
 
     default_kernel = "RBF() + WhiteKernel()"
     default_distribution = "Posterior"
-    default_plot_xmin = -2.
-    default_plot_xmax = 2.
+    default_plot_xmin = -2.0
+    default_plot_xmax = 2.0
+
+    manager = Manager()
+    initial_plot = manager.handle_generate_plots(
+        default_dataset_type,
+        default_function,
+        default_data_xmin,
+        default_data_xmax,
+        default_sigma,
+        default_nsample,
+        default_sample_method,
+        default_csv_file,
+        default_has_header,
+        default_xcol,
+        default_ycol,
+        default_kernel,
+        default_distribution,
+        default_plot_xmin,
+        default_plot_xmax,
+    )
 
     with gr.Blocks() as demo:
         gr.HTML("<div style='text-align:left; font-size:40px; font-weight: bold;'>Gaussian process visualizer</div>")
-        
-        dataset = gr.State(
-            get_dataset(
-                default_dataset_type,
-                default_function,
-                default_data_xmin,
-                default_data_xmax,
-                default_sigma,
-                default_nsample,
-                default_sample_method,
-                default_csv_file,
-                default_has_header,
-                default_xcol,
-                default_ycol,
-            )
-        )
 
-        true_dataset = gr.State(
-            get_true_dataset(
-                default_dataset_type,
-                default_function,
-                default_plot_xmin,
-                default_plot_xmax,
-            )
-        )
-
-        plot_data = gr.State(
-            compute_plot_values(
-                dataset.value,
-                default_kernel,
-                default_distribution,
-                default_plot_xmin,
-                default_plot_xmax,
-            )
-        )
+        manager_state = gr.State(manager)
 
         with gr.Row():
             with gr.Column(scale=2):
-                plot = gr.Plot(
-                    value=generate_plot(
-                        plot_data.value, 
-                        dataset.value,
-                        true_dataset.value,
-                    ),
-                )
+                plot = gr.Plot(value=initial_plot)
 
             with gr.Column(scale=1):
                 with gr.Tab("Data"):
@@ -236,22 +147,21 @@ def launch():
                             interactive=True,
                         )
 
-                    # Generated data options
                     with gr.Row():
                         function = gr.Textbox(
-                            label="Function (in terms of x)", 
+                            label="Function (in terms of x)",
                             value=default_function,
                             interactive=True,
                         )
-                    
+
                     with gr.Row():
                         data_xmin = gr.Number(
-                            label="Data x min", 
+                            label="Data x min",
                             value=default_data_xmin,
                             interactive=True,
                         )
                         data_xmax = gr.Number(
-                            label="Data x max", 
+                            label="Data x max",
                             value=default_data_xmax,
                             interactive=True,
                         )
@@ -269,37 +179,36 @@ def launch():
                         )
 
                         nsample = gr.Slider(
-                            label="Number of data points", 
-                            minimum=0, 
-                            maximum=100, 
-                            step=1, 
+                            label="Number of data points",
+                            minimum=0,
+                            maximum=100,
+                            step=1,
                             value=default_nsample,
                             interactive=True,
                         )
 
-                    # CSV options
                     with gr.Row():
                         csv_upload = gr.File(
-                            label="Upload CSV file", 
-                            file_types=['.csv'],
+                            label="Upload CSV file",
+                            file_types=[".csv"],
                             visible=False,
                         )
                     with gr.Row():
                         has_header = gr.Checkbox(
-                            label="CSV has header row", 
+                            label="CSV has header row",
                             value=default_has_header,
                             visible=False,
                         )
-                    
+
                     with gr.Row():
                         xcol = gr.Number(
-                            label="X column index", 
+                            label="X column index",
                             value=default_xcol,
                             precision=0,
                             visible=False,
                         )
                         ycol = gr.Number(
-                            label="Y column index", 
+                            label="Y column index",
                             value=default_ycol,
                             precision=0,
                             visible=False,
@@ -324,7 +233,7 @@ def launch():
 
                 with gr.Tab("Model"):
                     kernel = gr.Textbox(
-                        label="Kernel", 
+                        label="Kernel",
                         value=default_kernel,
                         interactive=True,
                     )
@@ -336,21 +245,22 @@ def launch():
                     )
                     with gr.Row():
                         plot_xmin = gr.Number(
-                            label="Plot x min", 
+                            label="Plot x min",
                             value=default_plot_xmin,
                             interactive=True,
                         )
                         plot_xmax = gr.Number(
-                            label="Plot x max", 
+                            label="Plot x max",
                             value=default_plot_xmax,
                             interactive=True,
                         )
 
-                    generate_button = gr.Button("Regenerate Plot")
+                generate_button = gr.Button("Regenerate Plot")
 
                 generate_button.click(
-                    fn=get_dataset,
+                    fn=handle_generate,
                     inputs=[
+                        manager_state,
                         dataset_type,
                         function,
                         data_xmin,
@@ -362,39 +272,16 @@ def launch():
                         has_header,
                         xcol,
                         ycol,
-                    ],
-                    outputs=[dataset],
-                ).then(
-                    fn=get_true_dataset,
-                    inputs=[
-                        dataset_type,
-                        function,
-                        plot_xmin,
-                        plot_xmax,
-                    ],
-                    outputs=[true_dataset],
-                ).then(
-                    fn=compute_plot_values,
-                    inputs=[
-                        dataset,
                         kernel,
                         distribution,
                         plot_xmin,
                         plot_xmax,
                     ],
-                    outputs=[plot_data],
-                ).then(
-                    fn=generate_plot,
-                    inputs=[
-                        plot_data,
-                        dataset,
-                        true_dataset,
-                    ],
-                    outputs=[plot],
+                    outputs=[manager_state, plot],
                 )
 
     demo.launch(css=CSS)
-                
 
-if  __name__ == "__main__":
+
+if __name__ == "__main__":
     launch()