Qblox

flake.nix

@@ -49,12 +49,15 @@
 
               env = {
                 QIBOLAB_PLATFORMS = (dirOf config.env.DEVENV_ROOT) + "/qibolab_platforms_qrc";
+                LD_LIBRARY_PATH = builtins.concatStringsSep ":" (map (p: "${p}/lib") (with pkgs; [
+                  stdenv.cc.cc.lib
+                  zlib
+                ]));
               };
 
               languages.python = {
                 enable = true;
-                libraries = with pkgs; [zlib];
-                version = "3.11";
+                version = "3.12";
                 poetry = {
                   enable = true;
                   install = {

pyproject.toml

@@ -18,15 +18,15 @@ classifiers = [
   "Topic :: Scientific/Engineering :: Physics",
 ]
 packages = [{ include = "qibolab", from = "src" }]
-include = ["*.out", "*.yml"]
+include = ["src/*.lark"]
 
 [tool.poetry.dependencies]
 python = ">=3.9,<3.13"
 qibo = "^0.2.8"
 numpy = "^1.26.4"
 scipy = "^1.13.0"
 pydantic = "^2.6.4"
-qblox-instruments = { version = "0.12.0", optional = true }
+qblox-instruments = { version = "^0.12.0", optional = true }
 lark = { version = "^1.1.9", optional = true }
 qcodes = { version = "^0.37.0", optional = true }
 qcodes_contrib_drivers = { version = "0.18.0", optional = true }
@@ -58,7 +58,7 @@ nbsphinx = "^0.9.1"
 ipython = "^8.12.0"
 sphinx-copybutton = "^0.5.1"
 # extras
-qblox-instruments = "0.12.0"
+qblox-instruments = "^0.12.0"
 qcodes = "^0.37.0"
 qcodes_contrib_drivers = "0.18.0"
 qibosoq = { version = "^0.1.2", python = "<3.12" }

src/qibolab/_core/components/channels.py

@@ -38,6 +38,23 @@ class Channel(Model):
     def port(self) -> int:
         return int(self.path)
 
+    def iqout(self, id_: ChannelId) -> Optional[ChannelId]:
+        """Extract associated IQ output channel.
+
+        This is the identity for each IQ output channel identifier, while it retrieves the
+        associated probe channel for acquisition ones, and :obj:`None` for any other one
+        (essentially, non-RF channels).
+
+        The argument is the identifier of the present channel, since it is not stored within
+        the objec itself, as it is only relevant to address it in a collection (and so,
+        out of the scope of the object itself).
+        """
+        return (
+            id_
+            if isinstance(self, IqChannel)
+            else (self.probe if isinstance(self, AcquisitionChannel) else None)
+        )
+
 
 class DcChannel(Channel):
     """Channel that can be used to send DC pulses."""
@@ -46,21 +63,21 @@ class DcChannel(Channel):
 class IqChannel(Channel):
     """Channel that can be used to send IQ pulses."""
 
-    mixer: Optional[str]
+    mixer: Optional[str] = None
     """Name of the IQ mixer component corresponding to this channel.
 
     None, if the channel does not have a mixer, or it does not need
     configuration.
     """
-    lo: Optional[str]
+    lo: Optional[str] = None
     """Name of the local oscillator component corresponding to this channel.
 
     None, if the channel does not have an LO, or it is not configurable.
     """
 
 
 class AcquisitionChannel(Channel):
-    twpa_pump: Optional[str]
+    twpa_pump: Optional[str] = None
     """Name of the TWPA pump component.
 
     None, if there is no TWPA, or it is not configurable.

src/qibolab/_core/components/configs.py

@@ -7,6 +7,7 @@
 configuration defined by these classes.
 """
 
+from pathlib import Path
 from typing import Annotated, Literal, Optional, Union
 
 from pydantic import Field
@@ -20,14 +21,20 @@
     "IqMixerConfig",
     "OscillatorConfig",
     "Config",
+    "Configs",
     "ChannelConfig",
+    "LogConfig",
 ]
 
 
 class Config(Model):
     """Configuration values depot."""
 
 
+Configs = dict[str, Config]
+"""Configuration database."""
+
+
 class DcConfig(Config):
     """Configuration for a channel that can be used to send DC pulses (i.e.
     just envelopes without modulation)."""
@@ -120,6 +127,14 @@ def __eq__(self, other) -> bool:
         )
 
 
+class LogConfig(Config):
+    """Configuration for logging."""
+
+    kind: Literal["log"] = "log"
+
+    path: Path
+
+
 ChannelConfig = Union[
-    DcConfig, IqMixerConfig, OscillatorConfig, IqConfig, AcquisitionConfig
+    DcConfig, IqMixerConfig, OscillatorConfig, IqConfig, AcquisitionConfig, LogConfig
 ]

src/qibolab/_core/execution_parameters.py

@@ -1,11 +1,18 @@
 from enum import Enum, auto
-from typing import Any, Optional
+from math import prod
+from typing import Any, Optional, TypeVar
+
+from pydantic import Field
+
+from qibolab._core.sequence import PulseSequence
 
 from .serialize import Model
-from .sweeper import ParallelSweepers
+from .sweeper import ParallelSweepers, iteration_length
 
 __all__ = ["AcquisitionType", "AveragingMode"]
 
+NS_TO_SEC = 1e-9
+
 
 class AcquisitionType(Enum):
     """Data acquisition from hardware."""
@@ -37,13 +44,24 @@ def average(self) -> bool:
         return self is not AveragingMode.SINGLESHOT
 
 
-ConfigUpdate = dict[str, dict[str, Any]]
+Update = dict[str, Any]
+
+ConfigUpdate = dict[str, Update]
 """Update for component configs.
 
 Maps component name to corresponding update, which in turn is a map from
 config property name that needs an update to its new value.
 """
 
+# TODO: replace with https://docs.python.org/3/reference/compound_stmts.html#type-params
+T = TypeVar("T")
+
+
+# TODO: lift for general usage in Qibolab
+def default(value: Optional[T], default: T) -> T:
+    """None replacement shortcut."""
+    return value if value is not None else default
+
 
 class ExecutionParameters(Model):
     """Data structure to deal with execution parameters."""
@@ -65,28 +83,47 @@ class ExecutionParameters(Model):
     """Data acquisition type."""
     averaging_mode: AveragingMode = AveragingMode.SINGLESHOT
     """Data averaging mode."""
-    updates: list[ConfigUpdate] = []
+    updates: list[ConfigUpdate] = Field(default_factory=list)
     """List of updates for component configs.
 
     Later entries in the list take precedence over earlier ones (if they
     happen to update the same thing). These updates will be applied on
     top of platform defaults.
     """
 
-    def results_shape(
-        self, sweepers: list[ParallelSweepers], samples: Optional[int] = None
-    ) -> tuple[int, ...]:
-        """Compute the expected shape for collected data."""
-
+    def bins(self, sweepers: list[ParallelSweepers]) -> tuple[int, ...]:
+        assert self.nshots is not None
         shots = (
             (self.nshots,) if self.averaging_mode is AveragingMode.SINGLESHOT else ()
         )
         sweeps = tuple(
             min(len(sweep.values) for sweep in parsweeps) for parsweeps in sweepers
         )
+        return shots + sweeps
+
+    def results_shape(
+        self, sweepers: list[ParallelSweepers], samples: int = -1
+    ) -> tuple[int, ...]:
+        """Compute the expected shape for collected data."""
         inner = {
             AcquisitionType.DISCRIMINATION: (),
             AcquisitionType.INTEGRATION: (2,),
             AcquisitionType.RAW: (samples, 2),
         }[self.acquisition_type]
-        return shots + sweeps + inner
+        return self.bins(sweepers) + inner
+
+    def estimate_duration(
+        self,
+        sequences: list[PulseSequence],
+        sweepers: list[ParallelSweepers],
+    ) -> float:
+        """Estimate experiment duration."""
+        duration = sum(seq.duration for seq in sequences)
+        relaxation = default(self.relaxation_time, 0)
+        nshots = default(self.nshots, 0)
+        return (
+            (duration + len(sequences) * relaxation)
+            * nshots
+            * NS_TO_SEC
+            * prod(iteration_length(s) for s in sweepers)
+        )

src/qibolab/_core/instruments/qblox/__init__.py

@@ -0,0 +1,3 @@
+from . import mock
+
+__all__ = ["mock"]