from datetime import datetime
from typing import Optional, Iterable, Union
from xml.etree.ElementTree import Element
from SDF.data_model._helper_functions import pop_child_element, element_is_empty, pop_all_child_elements
from SDF.data_model.abstract import XMLWritable
from SDF.data_model.comment import Comment
from SDF.data_model.dataset import Dataset
from SDF.data_model.date import Date
from SDF.data_model.element_set import ElementSet
from SDF.data_model.instrument import Instrument
from SDF.data_model.name import NameElement
from SDF.data_model.owner import Owner
from SDF.data_model.parameter import ParameterType
from SDF.data_model.sample import Sample
from SDF.data_model.sdf_object import SDFObject
[docs]class Workspace(SDFObject, XMLWritable):
"""Represents an SDF <workspace> element"""
def __init__(self, name: Union[str, NameElement], *,
owner: Optional[Union[str, Owner]] = None,
date: Optional[Union[datetime, Date]] = None,
comment: Optional[Union[str, Comment]] = None,
samples: Optional[Iterable[Sample]] = None,
parameters: Optional[Iterable[ParameterType]] = None,
instruments: Optional[Iterable[Instrument]] = None,
datasets: Optional[Iterable[Dataset]] = None,
workspaces: Optional[Iterable["Workspace"]] = None):
super().__init__(name=name, owner=owner, date=date, comment=comment, samples=samples, parameters=parameters,
instruments=instruments)
self.__datasets = ElementSet(key_func=lambda item: item.name,
check_func=lambda item: isinstance(item, Dataset))
if datasets is not None:
self.datasets.update(datasets)
self.__workspaces = ElementSet(key_func=lambda item: item.name,
check_func=lambda item: isinstance(item, Workspace))
if workspaces is not None:
self.workspaces.update(workspaces)
@property
def workspaces(self) -> ElementSet["Workspace"]:
return self.__workspaces
@property
def datasets(self) -> ElementSet[Dataset]:
return self.__datasets
[docs] def to_xml_element(self) -> Element:
element = Element("workspace")
super()._to_partial_xml_element(element)
for dataset in self.datasets:
element.append(dataset.to_xml_element())
for workspace in self.workspaces:
element.append(workspace.to_xml_element())
return element
[docs] @classmethod
def from_xml_element(cls, element: Element) -> "Workspace":
if element.tag != "workspace":
raise ValueError(f"Expected a <workspace> element, got {element.tag}")
name = NameElement.from_xml_element(pop_child_element(element, 0))
ret = cls(name=name)
ret._from_partial_xml_element(element)
# NOTE: For backwards-compatibility, workspaces and datasets can be read in any order.
for child in pop_all_child_elements(element):
if child.tag == "workspace":
ret.workspaces.add(Workspace.from_xml_element(child))
elif child.tag == "dataset":
ret.datasets.add(Dataset.from_xml_element(child))
else:
raise ValueError(f"Expected a <dataset> or <workspace> element, got {element.tag}")
if not element_is_empty(element):
raise ValueError("Element is not empty")
return ret
def __iter__(self) -> Iterable[SDFObject]:
"""Iterate over child datasets and workspaces"""
for child in self.datasets:
yield child
for child in self.workspaces:
yield child
def __getitem__(self, key: str) -> Union["Workspace", Dataset]:
if key in self.workspaces and key in self.datasets:
raise KeyError(f"Ambigious key '{key}', can be workspace or dataset")
if key in self.workspaces:
return self.workspaces[key]
if key in self.datasets:
return self.datasets[key]
raise KeyError(f"No child dataset or workspace with name '{key}'")
def __contains__(self, item: Union[str, "Workspace", Dataset]) -> bool:
if isinstance(item, str):
return item in self.workspaces or item in self.datasets
if isinstance(item, Workspace):
return item in self.workspaces
if isinstance(item, Dataset):
return item in self.datasets
raise TypeError(f"Expected a string, Dataset or Workspace, got {item}")
def __len__(self) -> int:
return len(self.datasets) + len(self.workspaces)
def __repr__(self) -> str:
return f"{self.__class__.__name__}({self.name!r}, owner={self.owner!r}, date={self.date!r}, " \
f"comment={self.comment!r}, samples={self.samples!r}, parameters={self.parameters!r}, " \
f"instruments={self.instruments!r}, datasets={self.datasets!r}, workspaces={self.workspaces!r})"
def __eq__(self, other) -> bool:
if isinstance(other, Workspace):
return (super().__eq__(other)
and self.datasets == other.datasets
and self.workspaces == other.workspaces)
return False