Merge pull request #805 from qiboteam/dataclass_web_report

Simplifying plot generation

src/qibocal/auto/execute.py

@@ -26,8 +26,6 @@ class Executor:
     """Qubits/Qubit Pairs to be calibrated."""
     platform: Platform
     """Qubits' platform."""
-    max_iterations: int
-    """Maximum number of iterations."""
     update: bool = True
     """Runcard update mechanism."""
 
@@ -46,7 +44,6 @@ def load(
         return cls(
             actions=card.actions,
             history=History({}),
-            max_iterations=card.max_iterations,
             output=output,
             platform=platform,
             targets=targets,
@@ -62,12 +59,8 @@ def run(self, mode):
         """
         for action in self.actions:
             task = Task(action)
-            task.iteration = self.history.iterations(task.id)
-            log.info(
-                f"Executing mode {mode.name} on {task.id} iteration {task.iteration}."
-            )
+            log.info(f"Executing mode {mode.name} on {task.id}.")
             completed = task.run(
-                max_iterations=self.max_iterations,
                 platform=self.platform,
                 targets=self.targets,
                 folder=self.output,
@@ -78,4 +71,4 @@ def run(self, mode):
             if mode.name in ["autocalibration", "fit"] and self.platform is not None:
                 completed.update_platform(platform=self.platform, update=self.update)
 
-            yield completed.task.uid
+            yield completed.task.id

src/qibocal/auto/history.py

@@ -5,8 +5,8 @@
 
 
 def add_timings_to_meta(meta, history):
-    for task_id, iteration in history:
-        completed = history[(task_id, iteration)]
+    for task_id in history:
+        completed = history[task_id]
         if task_id not in meta:
             meta[task_id] = {}
 
@@ -33,14 +33,6 @@ class History(dict[tuple[Id, int], Completed]):
 
     def push(self, completed: Completed):
         """Adding completed task to history."""
-        self[completed.task.uid] = completed
-
-    def iterations(self, task_id: Id):
-        """Count task id present in history."""
-        counter = 0
-        for task, _ in self:
-            if task == task_id:
-                counter += 1
-        return counter
+        self[completed.task.id] = completed
 
     # TODO: implemet time_travel()

src/qibocal/auto/runcard.py

@@ -17,9 +17,6 @@
 Targets = Union[list[QubitId], list[QubitPairId], list[tuple[QubitId, ...]]]
 """Elements to be calibrated by a single protocol."""
 
-MAX_ITERATIONS = 5
-"""Default max iterations."""
-
 
 @dataclass(config=dict(smart_union=True))
 class Action:
@@ -55,8 +52,6 @@ class Runcard:
     """Qibo backend."""
     platform: str = os.environ.get("QIBO_PLATFORM", "dummy")
     """Qibolab platform."""
-    max_iterations: int = MAX_ITERATIONS
-    """Maximum number of iterations."""
 
     def __post_init__(self):
         if self.targets is None and self.platform_obj is not None:

src/qibocal/auto/task.py

@@ -9,7 +9,7 @@
 from qibolab.platform import Platform
 from qibolab.serialize import dump_platform
 
-from ..config import log, raise_error
+from ..config import log
 from ..protocols.characterization import Operation
 from .mode import ExecutionMode
 from .operation import Data, DummyPars, Results, Routine, dummy_operation
@@ -29,8 +29,6 @@
 class Task:
     action: Action
     """Action object parsed from Runcard."""
-    iteration: int = 0
-    """Task iteration."""
 
     @property
     def targets(self) -> Targets:
@@ -42,11 +40,6 @@ def id(self) -> Id:
         """Task Id."""
         return self.action.id
 
-    @property
-    def uid(self) -> TaskId:
-        """Task unique Id."""
-        return (self.action.id, self.iteration)
-
     @property
     def operation(self):
         """Routine object from Operation Enum."""
@@ -67,17 +60,11 @@ def update(self):
 
     def run(
         self,
-        max_iterations: int,
         platform: Platform = None,
         targets: Targets = list,
         mode: ExecutionMode = None,
         folder: Path = None,
     ):
-        if self.iteration > max_iterations:
-            raise_error(
-                ValueError,
-                f"Maximum number of iterations {max_iterations} reached!",
-            )
 
         if self.targets is None:
             self.action.targets = targets
@@ -150,7 +137,7 @@ def __post_init__(self):
     @property
     def datapath(self):
         """Path contaning data and results file for task."""
-        path = self.folder / "data" / f"{self.task.id}_{self.task.iteration}"
+        path = self.folder / "data" / f"{self.task.id}"
         if not path.is_dir():
             path.mkdir(parents=True)
         return path

src/qibocal/cli/acquisition.py

@@ -38,7 +38,6 @@ def acquire(runcard, folder, force):
     (path / META).write_text(json.dumps(meta, indent=4))
 
     executor = Executor.load(runcard, path, platform, runcard.targets)
-
     # connect and initialize platform
     if platform is not None:
         platform.connect()

src/qibocal/cli/autocalibration.py

@@ -8,7 +8,7 @@
 from ..auto.execute import Executor
 from ..auto.history import add_timings_to_meta
 from ..auto.mode import ExecutionMode
-from ..cli.report import ReportBuilder
+from .report import report
 from .utils import (
     META,
     PLATFORM,
@@ -59,9 +59,8 @@ def autocalibrate(runcard, folder, force, update):
         meta["end-time"] = e.strftime("%H:%M:%S")
         # dump updated meta
         meta = add_timings_to_meta(meta, executor.history)
-        report = ReportBuilder(path, runcard.targets, executor, meta, executor.history)
-        report.run(path)
         (path / META).write_text(json.dumps(meta, indent=4))
+        report(path, executor)
 
     # stop and disconnect platform
     if platform is not None:

src/qibocal/cli/report.py

@@ -1,26 +1,34 @@
+import io
 import json
-import tempfile
-from functools import cached_property
-from pathlib import Path
+import pathlib
+from typing import Optional, Union
 
+import plotly.graph_objects as go
 import yaml
+from jinja2 import Environment, FileSystemLoader
 from qibo.backends import GlobalBackend
-from qibolab.qubits import QubitId
+from qibolab.qubits import QubitId, QubitPairId
 
 from qibocal.auto.execute import Executor
 from qibocal.auto.mode import ExecutionMode
 from qibocal.auto.runcard import Runcard
-from qibocal.auto.task import TaskId
+from qibocal.auto.task import Completed
+from qibocal.cli.utils import META, RUNCARD
 from qibocal.config import log
+from qibocal.web.report import STYLES, TEMPLATES, Report
 
-META = "meta.json"
-RUNCARD = "runcard.yml"
-UPDATED_PLATFORM = "new_platform.yml"
-PLATFORM = "platform.yml"
+ReportOutcome = tuple[str, list[go.Figure]]
+"""Report produced by protocol."""
 
 
-def generate_figures_and_report(node, target):
-    """Returns figures and table for report."""
+def generate_figures_and_report(
+    node: Completed, target: Union[QubitId, QubitPairId, list[QubitId]]
+) -> ReportOutcome:
+    """Calling protocol plot by checking if fit has been performed.
+
+    It operates on a completed `node` and a specific protocol `target`, generating
+    a report outcome (cf. `ReportOutcome`).
+    """
 
     if node.results is None:
         # plot acquisition data
@@ -33,14 +41,34 @@ def generate_figures_and_report(node, target):
     return node.task.operation.report(data=node.data, fit=node.results, target=target)
 
 
-def report(path):
-    """Report generation
-
-    Arguments:
+def plotter(
+    node: Completed, target: Union[QubitId, QubitPairId, list[QubitId]]
+) -> tuple[str, str]:
+    """Run plotly pipeline for generating html.
 
-    - FOLDER: input folder.
+    Performs conversions of plotly figures in html rendered code for completed
+    node on specific target.
 
     """
+    figures, fitting_report = generate_figures_and_report(node, target)
+    buffer = io.StringIO()
+    html_list = []
+    for figure in figures:
+        figure.write_html(buffer, include_plotlyjs=False, full_html=False)
+        buffer.seek(0)
+        html_list.append(buffer.read())
+    buffer.close()
+    all_html = "".join(html_list)
+    return all_html, fitting_report
+
+
+def report(path: pathlib.Path, executor: Optional[Executor] = None):
+    """Report generation.
+
+    Generates the report for protocol dumped in `path`.
+    Executor can be passed to generate report on the fly.
+    """
+
     if path.exists():
         log.warning(f"Regenerating {path}/index.html")
     # load meta
@@ -50,61 +78,29 @@ def report(path):
 
     # set backend, platform and qubits
     GlobalBackend.set_backend(backend=meta["backend"], platform=meta["platform"])
-    backend = GlobalBackend()
-    platform = backend.platform
 
     # load executor
-    executor = Executor.load(runcard, path, targets=runcard.targets)
-    # produce html
-    builder = ReportBuilder(path, runcard.targets, executor, meta)
-    builder.run(path)
-
-
-class ReportBuilder:
-    """Builder to produce html report."""
-
-    def __init__(self, path: Path, targets, executor: Executor, metadata, history=None):
-        self.path = self.title = path
-        self.targets = targets
-        self.executor = executor
-        self.metadata = metadata
-        self._history = history
-
-    @cached_property
-    def history(self):
-        if self._history is None:
-            list(self.executor.run(mode=ExecutionMode.report))
-            return self.executor.history
-        else:
-            return self._history
-
-    def routine_name(self, routine, iteration):
-        """Prettify routine's name for report headers."""
-        name = routine.replace("_", " ").title()
-        return f"{name} - {iteration}"
-
-    def routine_targets(self, task_id: TaskId):
-        """Get local targets parameter from Task if available otherwise use global one."""
-        local_targets = self.history[task_id].task.targets
-        return local_targets if len(local_targets) > 0 else self.targets
-
-    def single_qubit_plot(self, task_id: TaskId, qubit: QubitId):
-        """Generate single qubit plot."""
-        node = self.history[task_id]
-        figures, fitting_report = generate_figures_and_report(node, qubit)
-        with tempfile.NamedTemporaryFile(delete=False) as temp:
-            html_list = []
-            for figure in figures:
-                figure.write_html(temp.name, include_plotlyjs=False, full_html=False)
-                temp.seek(0)
-                fightml = temp.read().decode("utf-8")
-                html_list.append(fightml)
-
-        all_html = "".join(html_list)
-        return all_html, fitting_report
-
-    def run(self, path):
-        """Generation of html report."""
-        from qibocal.web.report import create_report
-
-        create_report(path, self)
+    if executor is None:
+        executor = Executor.load(runcard, path, targets=runcard.targets)
+        # produce html
+        list(executor.run(mode=ExecutionMode.report))
+
+    css_styles = f"<style>\n{pathlib.Path(STYLES).read_text()}\n</style>"
+
+    env = Environment(loader=FileSystemLoader(TEMPLATES))
+    template = env.get_template("template.html")
+    html = template.render(
+        is_static=True,
+        css_styles=css_styles,
+        path=path,
+        title=path.name,
+        report=Report(
+            path=path,
+            targets=executor.targets,
+            history=executor.history,
+            meta=meta,
+            plotter=plotter,
+        ),
+    )
+
+    (path / "index.html").write_text(html)

src/qibocal/web/report.py

@@ -1,29 +1,36 @@
-import os
 import pathlib
+from dataclasses import dataclass
+from typing import Callable
 
-from jinja2 import Environment, FileSystemLoader
-
-from qibocal import __version__
-from qibocal.cli.report import ReportBuilder
+from qibocal.auto.history import History
+from qibocal.auto.task import TaskId
 
 WEB_DIR = pathlib.Path(__file__).parent
 STYLES = WEB_DIR / "static" / "styles.css"
 TEMPLATES = WEB_DIR / "templates"
 
 
-def create_report(path, report: ReportBuilder):
-    """Creates an HTML report for the data in the given path."""
-    with open(STYLES) as file:
-        css_styles = f"<style>\n{file.read()}\n</style>"
+@dataclass
+class Report:
+    """Report generation class."""
+
+    path: pathlib.Path
+    """Path with calibration data."""
+    targets: list
+    """Global targets."""
+    history: History
+    """History of protocols."""
+    meta: dict
+    """Meta data."""
+    plotter: Callable
+    """Plotting function to generate html."""
 
-    env = Environment(loader=FileSystemLoader(TEMPLATES))
-    template = env.get_template("template.html")
-    html = template.render(
-        is_static=True,
-        css_styles=css_styles,
-        version=__version__,
-        report=report,
-    )
+    @staticmethod
+    def routine_name(routine):
+        """Prettify routine's name for report headers."""
+        return routine.title()
 
-    with open(os.path.join(path, "index.html"), "w") as file:
-        file.write(html)
+    def routine_targets(self, task_id: TaskId):
+        """Get local targets parameter from Task if available otherwise use global one."""
+        local_targets = self.history[task_id].task.targets
+        return local_targets if len(local_targets) > 0 else self.targets