from copy import deepcopy
from dataclasses import dataclass, field
from enum import StrEnum
from io import StringIO
from pathlib import Path
from typing import Any
from urllib.parse import quote
from ruamel.yaml import YAML
from semi_cr.core.helpers import deprecated
from semi_cr.core.lab.utils.yaml_utils import _read_yaml, deep_merge
[docs]
class GraphRelation(StrEnum):
CONTAINS = "contains"
DEPENDENCY = "dependency"
CONNECTION = "connection"
REPRESENTS = "represents"
INTERNAL_CONNECTION = "internal_connection"
PHYSICAL = "physical"
DEVICE = "device"
[docs]
@dataclass(frozen=True)
class InstrumentSpec:
"""
Static description of an instrument.
Does not represent an active connection and does not contain
a QCoDeS Instrument instance.
"""
name: str
type: str
init: dict[str, Any] = field(default_factory=dict)
@property
def runtime_name(self) -> str:
return self.init.get("name", self.name)
[docs]
def static_instrument_node_id(spec: InstrumentSpec) -> str:
return f"lab://instrument/{quote(spec.name, safe='')}"
# @dataclass(frozen=True)
# class InstrumentBinding:
# spec: InstrumentSpec
# instrument: Instrument
# @property
# def name(self) -> str:
# return self.spec.name
# @property
# def runtime_name(self) -> str:
# return self.instrument.name
[docs]
class ConfigBundle:
"""
A container class for QCoDes configuration that has been loaded from one or more yaml files. Instantiate a
ConfigBundle with or without files, and add more using `import_yaml`.
Jens has a suspicion that we can avoid some complexity by encapsulating the configuration merging logic properly in
here and having it here only. I believe now the ConfifBundle does that properly.
"""
def __init__(
self,
files: str | Path | tuple[str | Path, ...] | None = None,
):
# Recreate the dataclass functionality by setting the local member `self.files` but directly import the provided
# files too. We may do this because the `load` function (which is deprecated) starts by wiping the existing
# config, so if people opt for that path the class will continue to work as expected.
self._config: dict[str, Any] = {}
self.files: tuple[str | Path, ...] = ()
if files is not None:
if isinstance(files, tuple):
self.files = files
else:
self.files = (files,)
self.import_yaml(files)
[docs]
@deprecated("Loads whatever is in `config` but that may be out of sync.")
def load(self) -> dict[str, Any]:
"""
Loads a dict of paths to configuration files into a merged dict. A
Maintaining parity by initializing self.config to the empty dict. Would
be nice if the ConfigBundle would maintain state.
DEPRECATED: Get rid of this method and `self.config` in order to maintain atomicity.
:return:
"""
# The inner state of the class may already be initialized, so if somebody uses this API it means they'd expected
# this to be a dataclass still, which isn't the case. Therefore we get rid of the state to simulate that
# behavior.
self.config = {}
for file in self.files:
deep_merge(self.config, _read_yaml(file))
return self.config
[docs]
def import_yaml(
self,
yamlfile: str | Path | tuple[str | Path, ...],
) -> None:
"""
Takes a reference to a file or a tuple of such references and augments the
current set of configurations with it.
"""
if isinstance(yamlfile, tuple):
for path in yamlfile:
self.import_yaml(path)
return
if not isinstance(yamlfile, (str, Path)):
raise TypeError("Expected str, Path, or tuple of those.")
path = Path(yamlfile)
loaded = _read_yaml(path)
deep_merge(
self._config,
loaded,
)
self.files = (*self.files, path)
[docs]
def merge(
self,
other: "ConfigBundle",
) -> "ConfigBundle":
result = ConfigBundle()
deep_merge(
result._config,
deepcopy(self._config),
)
deep_merge(
result._config,
deepcopy(other._config),
)
result.files = (
*self.files,
*other.files,
)
return result
[docs]
def dump_yaml(self, stream) -> None:
"""
Dumps a YaML structure from the current ConfigBundle contents into a provided stream.
"""
yaml_rt = YAML()
yaml_rt.default_flow_style = False
yaml_rt.dump(self._config, stream)
[docs]
def get_yaml(self) -> str:
"""
:return: a string containing a YaML document containing the current state of the ConfigBundle.
"""
buffer = StringIO()
self.dump_yaml(buffer)
return buffer.getvalue()
@property
def config(self) -> dict[str, Any]:
return self._config
@config.setter
def config(self, value: dict[str, Any]):
self._config = value
[docs]
@classmethod
def combine(
cls,
*bundles: "ConfigBundle",
) -> "ConfigBundle":
result = cls()
for bundle in bundles:
deep_merge(
result._config,
deepcopy(bundle.config),
)
result.files += bundle.files
return result