Core Concepts

This page provides a deep technical reference for the three foundational classes in Hera: Project, ToolkitHome, and abstractToolkit. Understanding these is essential for working with any part of the system.

---

Class Hierarchy Overview

The diagram below shows the core hierarchy — Project at the base, abstractToolkit adding datasource management on top, and ToolkitHome acting as the registry/factory.

Project <|-- abstractToolkit : "inherits data layer"
ToolkitHome ..> abstractToolkit : "instantiates via getToolkit()"

--> --> Project <|-- abstractToolkit : "inherits data layer" ToolkitHome ..> abstractToolkit : "instantiates via getToolkit()" --> -->

Class	Role	Key methods
Project	Central data access layer — CRUD for measurements, simulations, cache	addMeasurementsDocument, getMeasurementsDocuments, setConfig, getCounterAndAdd, saveData, export
abstractToolkit	Base for all domain toolkits — adds datasource management, analysis/presentation layers	addDataSource, getDataSourceData, getDataSourceList, deleteDataSource, setDataSourceDefaultVersion
ToolkitHome	Registry/factory — resolves toolkit names to classes	getToolkit, getToolkitTable, registerToolkit, getToolkitDocuments

Concrete Toolkit Hierarchy

All domain toolkits extend abstractToolkit. The diagram below shows the full inheritance tree including the special dataToolkit that manages repositories.

t : "simulations" abstractToolkit <|-- LSMToolkit : "simulations" abstractToolkit <|-- RiskToolkit : "risk assessment"

-->
-->t : "simulations"
    abstractToolkit <|-- LSMToolkit : "simulations"
    abstractToolkit <|-- RiskToolkit : "risk assessment"

--> -->

Toolkit class	Domain	Key capabilities
dataToolkit	Data	Repository management, loading data sources into projects
TopographyToolkit	GIS (raster)	Elevation data, STL generation, CRS conversion
BuildingsToolkit	GIS (vector)	Building footprints, 3D STL for CFD
DemographyToolkit	GIS (vector)	Population calculations within polygons
LandCoverToolkit	GIS (raster)	Land cover classification, roughness estimation
lowFreqToolKit	Meteorology	Hourly/daily station data with analysis + presentation
HighFreqToolKit	Meteorology	High-frequency sonic anemometer data
OFToolkit	Simulations	OpenFOAM lifecycle (templates, running, post-processing)
LSMToolkit	Simulations	Lagrangian Stochastic Model
RiskToolkit	Risk	Agent-based hazard and casualty modeling

---

Project

Source: hera/datalayer/project.py

The Project class is the central data access layer. Every interaction with stored data — measurements, simulations, cached results — goes through a Project instance.

Responsibilities

1. Document CRUD — Add, query, and delete documents across three MongoDB collections: Measurements, Simulations, and Cache.
2. Configuration — Per-project key-value config stored as a special Cache document (type = <projectName>__config__).
3. Counters — Atomic counters for generating sequential IDs (stored inside the config document under desc.counters).
4. File Management — Each project has a filesDirectory where physical files (parquet, netcdf, etc.) are stored on disk.
5. Import/Export — Projects can be exported to a zip file and loaded into another MongoDB instance.

Initialization Flow

["Save filesDirectory\nto project config\n(if not already saved)"] SaveConfig --> Ready["Project instance\nready for use"]

-->
-->["Save filesDirectory\nto project config\n(if not already saved)"]
    SaveConfig --> Ready["Project instance\nready for use"]

--> -->

The Default Project

When projectName=None and no caseConfiguration.json exists, Hera uses a special "defaultProject" which is read-only. This is used by the dataToolkit to store repository metadata without polluting user projects.

Key Methods

Method	Description
addMeasurementsDocument(resource, dataFormat, type, desc)	Insert a new measurement document into MongoDB
getMeasurementsDocuments(**filters)	Query measurement documents by any combination of fields
deleteMeasurementsDocuments(**kwargs)	Remove matching measurement documents
setConfig(**kwargs)	Update project configuration (atomic per-key updates)
getConfig()	Return the full config dict
setCounter(name, default)	Define a named counter
getCounterAndAdd(name, addition)	Atomically read and increment a counter
saveData(name, data, desc, kind)	Auto-detect format, save to disk, and create a document
export(path)	Export all project documents to a zip file
Project.load(proj, path, is_hard_import)	Import documents from an exported zip

Document Structure

Every document in Hera has the following top-level fields:

{
    "projectName": "MY_PROJECT",
    "_cls": "Metadata.Measurements",   # or .Simulations / .Cache
    "type": "ToolkitDataSource",        # application-defined type tag
    "resource": "/path/to/data.parquet",
    "dataFormat": "parquet",
    "desc": {                           # free-form metadata dict
        "toolkit": "MeteoLowFreq",
        "datasourceName": "YAVNEEL",
        "version": [0, 0, 1],
        ...
    }
}

---

ToolkitHome

Source: hera/toolkit.py (class ToolkitHome)

ToolkitHome is the central registry for all available toolkits. A singleton instance toolkitHome is created at import time in hera/__init__.py.

Responsibilities

1. Static Registry — A dict mapping toolkit names (e.g., "MeteoLowFreq") to their full Python class paths.
2. Dynamic Registry — Toolkits registered at runtime via registerToolkit() are stored as ToolkitDataSource documents in MongoDB.
3. Instantiation — getToolkit(name, projectName) resolves the class, imports it, and returns a new instance.
4. Auto-Registration — auto_register_and_get() can discover toolkit classes from repository JSON hints or DB documents.

Toolkit Resolution Flow

mport"] RegisterAndImport --> Instantiate

AutoRegister -- "No" --> ReturnNone["Return None\n(toolkit not available)"]

--> -->mport"] RegisterAndImport --> Instantiate AutoRegister -- "No" --> ReturnNone["Return None\n(toolkit not available)"] --> -->

Singleton Pattern

ToolkitHome is instantiated once at hera/__init__.py as toolkitHome = ToolkitHome(). All code should use this singleton rather than creating new instances. The static registry dict is populated in __init__ and shared across the entire process.

Static Toolkit Registry

The _toolkits dict is structured as:

{
    "GIS_Raster_Topography": {
        "cls": "hera.measurements.GIS.raster.topography.TopographyToolkit",
        "desc": None,
        "type": "measurements"
    },
    "MeteoLowFreq": {
        "cls": "hera.measurements.meteorology.lowfreqdata.toolkit.lowFreqToolKit",
        "desc": None,
        "type": "measurements"
    },
    # ... 16 built-in toolkits total
}

Save Modes

ToolkitHome defines standard save-mode constants used throughout the system:

Constant	Value	Meaning
TOOLKIT_SAVEMODE_NOSAVE	None	Do not persist
TOOLKIT_SAVEMODE_ONLYFILE	"File"	Save to disk only
TOOLKIT_SAVEMODE_ONLYFILE_REPLACE	"File_overwrite"	Save to disk, overwrite
TOOLKIT_SAVEMODE_FILEANDDB	"DB"	Save to disk + MongoDB document
TOOLKIT_SAVEMODE_FILEANDDB_REPLACE	"DB_overwrite"	Save to disk + MongoDB, overwrite

---

abstractToolkit

Source: hera/toolkit.py (class abstractToolkit)

abstractToolkit is the base class for all domain toolkits. It inherits from Project, meaning every toolkit instance is a project — it has full access to measurements, simulations, cache, config, and counters.

Design Decision: Inheritance over Composition

abstractToolkit inherits from Project rather than wrapping it. This means every toolkit call like toolkit.getMeasurementsDocuments(...) goes directly to the Project implementation. The toolkit adds a datasource management layer on top, where each datasource is a measurement document tagged with type="ToolkitDataSource" and the toolkit's name.

What It Adds Over Project

1. Toolkit Identity — toolkitName property, automatically injected into every document created by the toolkit.
2. Datasource Versioning — Each datasource has a name + version tuple (major, minor, patch). The system supports multiple versions and a default-version mechanism.
3. Analysis & Presentation Layers — self._analysis and self._presentation hold domain-specific logic (e.g., statistical calculations, plotting).
4. Automatic Tagging — addMeasurementsDocument, addSimulationsDocument, and addCacheDocument are overridden to inject the toolkit name into every document's desc.

Datasource Lifecycle

-> UseDefault UseVersion --> GetData UseDefault --> GetData GetData --> CallGetData CallGetData --> ReturnData

-->
-->-> UseDefault
    UseVersion --> GetData
    UseDefault --> GetData
    GetData --> CallGetData
    CallGetData --> ReturnData

--> -->

Key Methods

Method	Description
addDataSource(name, resource, dataFormat, version)	Register a new datasource (versioned)
getDataSourceDocument(name, version)	Get the MongoDB document for a datasource
getDataSourceData(name, version)	Load and return the actual data
getDataSourceList(**filters)	List all datasource names
getDataSourceTable(**filters)	DataFrame of all datasources with metadata
deleteDataSource(name, version)	Remove a datasource document
setDataSourceDefaultVersion(name, version)	Set which version is returned by default

Concrete Toolkit Example

A concrete toolkit like lowFreqToolKit extends abstractToolkit and provides:

• analysis — Methods like addDatesColumns(), calcHourlyDist(), resampleSecondMoments()
• presentation — Plot generators like dailyPlots.plotScatter(), seasonalPlots.plotProbContourf_bySeason()
• docType — A string identifying the data type this toolkit handles

from hera import toolkitHome

# Instantiate via the registry
lf = toolkitHome.getToolkit(toolkitHome.METEOROLOGY_LOWFREQ, projectName="MY_PROJECT")

# Access data
df = lf.getDataSourceData("YAVNEEL")

# Use analysis
enriched = lf.analysis.addDatesColumns(df, datecolumn="datetime")

# Use presentation
ax = lf.presentation.dailyPlots.plotScatter(enriched, plotField="RH")

---

dataToolkit

Source: hera/utils/data/toolkit.py

dataToolkit extends abstractToolkit and serves as the repository management layer. It operates on the defaultProject (read-only for normal users) and manages the mapping between repository JSON files and project datasources.

Responsibilities

1. Repository Registration — addRepository(name, path) stores a reference to a repository JSON file as a datasource in the default project.
2. Repository Loading — loadAllDatasourcesInRepositoryJSONToProject() iterates through a parsed repository JSON and, for each toolkit section, dispatches to specialized handlers.
3. Static Helpers — resolveDataSourcePaths() and loadRepositoryFromPath() allow working with repository JSONs without MongoDB.

Repository JSON Loading Pipeline

on HandleSim --> NextSection HandleCache --> NextSection HandleFunc --> NextSection NextSection --> SectionType

-->
-->on
    HandleSim --> NextSection
    HandleCache --> NextSection
    HandleFunc --> NextSection
    NextSection --> SectionType

--> -->

Handler Dispatch Table

JSON Section Key	Handler Method	Action
Config	_handle_Config	Calls toolkit.setConfig(**docTypeDict)
Datasource	_handle_DataSource	Adds each item as a versioned datasource
Measurements	_DocumentHandler	Adds raw measurement documents
Simulations	_DocumentHandler	Adds simulation documents
Cache	_DocumentHandler	Adds cache documents
Function	_handle_Function	Calls a named function with parameters

---

API Reference

hera.toolkit.ToolkitHome

Bases: abstractToolkit

Central registry for available toolkits (static + dynamic).

Responsibilities: - getToolkit(toolkitName, ...): locate & instantiate a toolkit class. - getToolkitTable(projectName): table of all toolkits (static + DB). - registerToolkit(...): register a toolkit class as a ToolkitDataSource using the abstractToolkit data source interface.

Source code in hera/toolkit.py

class ToolkitHome(abstractToolkit):
    """
    Central registry for available toolkits (static + dynamic).

    Responsibilities:
      - getToolkit(toolkitName, ...): locate & instantiate a toolkit class.
      - getToolkitTable(projectName): table of all toolkits (static + DB).
      - registerToolkit(...): register a toolkit class as a ToolkitDataSource
        using the abstractToolkit data source interface.
    """

    # Save modes (kept for compatibility)
    TOOLKIT_SAVEMODE_NOSAVE = TOOLKIT_SAVEMODE_NOSAVE
    TOOLKIT_SAVEMODE_ONLYFILE = TOOLKIT_SAVEMODE_ONLYFILE
    TOOLKIT_SAVEMODE_ONLYFILE_REPLACE = TOOLKIT_SAVEMODE_ONLYFILE_REPLACE
    TOOLKIT_SAVEMODE_FILEANDDB = TOOLKIT_SAVEMODE_FILEANDDB
    TOOLKIT_SAVEMODE_FILEANDDB_REPLACE = TOOLKIT_SAVEMODE_FILEANDDB_REPLACE

    # Static toolkit identifiers
    GIS_BUILDINGS = "GIS_Buildings"
    GIS_TILES = "GIS_Tiles"
    GIS_LANDCOVER = "GIS_LandCover"
    GIS_VECTOR_TOPOGRAPHY = "GIS_Vector_Topography"
    GIS_RASTER_TOPOGRAPHY = "GIS_Raster_Topography"
    GIS_DEMOGRAPHY = "GIS_Demography"
    RISKASSESSMENT = "RiskAssessment"
    LSM = "LSM"

    DATA = "heraData"

    SIMULATIONS_WORKFLOWS = "hermesWorkflows"
    SIMULATIONS_OPENFOAM = "OpenFOAM"

    METEOROLOGY_HIGHFREQ = "MeteoHighFreq"
    METEOROLOGY_LOWFREQ = "MeteoLowFreq"

    EXPERIMENT = "experiment"

    WINDPROFILE = "WindProfile"
    GAUSSIANDISPERSION = "GaussianDispersion"
    MACHINELEARNING_DEEPLEARNING = "machine_deep_learning"

    _toolkits: Dict[str, Dict[str, Any]] = None

    def __init__(self, projectName: Optional[str] = None, filesDirectory: Optional[str] = None):
        """
        Initialize the ToolkitHome registry.

        ToolkitHome is itself a toolkit (inherits from abstractToolkit):

        - ``projectName`` is loaded automatically from caseConfiguration if None.
        - All dynamic ToolkitDataSource entries are registered under
          toolkitName="ToolkitHome".

        Parameters
        ----------
        projectName : str, optional
            The project name. If None, loaded from caseConfiguration.
        filesDirectory : str, optional
            Directory for file outputs.
        """
        super().__init__(toolkitName="ToolkitHome", projectName=projectName, filesDirectory=filesDirectory)

        # Static built-in toolkits (internal source)
        self._toolkits = dict(
            GIS_Buildings=dict(
                cls="hera.measurements.GIS.vector.buildings.toolkit.BuildingsToolkit",
                desc=None,
                type="measurements",
            ),
            GIS_Tiles=dict(
                cls="hera.measurements.GIS.raster.tiles.TilesToolkit",
                desc=None,
                type="measurements",
            ),
            GIS_Vector_Topography=dict(
                cls="hera.measurements.GIS.vector.topography.TopographyToolkit",
                desc=None,
                type="measurements",
            ),
            GIS_Raster_Topography=dict(
                cls="hera.measurements.GIS.raster.topography.TopographyToolkit",
                desc=None,
                type="measurements",
            ),
            GIS_Demography=dict(
                cls="hera.measurements.GIS.vector.demography.DemographyToolkit",
                desc=None,
                type="measurements",
            ),
            GIS_LandCover=dict(
                cls="hera.measurements.GIS.raster.landcover.LandCoverToolkit",
                desc=None,
                type="measurements",
            ),
            RiskAssessment=dict(
                cls="hera.riskassessment.riskToolkit.RiskToolkit",
                desc=None,
                type="riskassessment",
            ),
            LSM=dict(
                cls="hera.simulations.LSM.toolkit.LSMToolkit",
                desc=None,
                type="simulations",
            ),
            OF_LSM=dict(
                cls="hera.simulations.openFoam.LSM.toolkit.OFLSMToolkit",
                desc=None,
                type="simulations",
            ),
            MeteoHighFreq=dict(
                cls="hera.measurements.meteorology.highfreqdata.toolkit.HighFreqToolKit",
                desc=None,
                type="measurements",
            ),
            MeteoLowFreq=dict(
                cls="hera.measurements.meteorology.lowfreqdata.toolkit.lowFreqToolKit",
                desc=None,
                type="measurements",
            ),
            hermesWorkflows=dict(
                cls="hera.simulations.hermesWorkflowToolkit.hermesWorkflowToolkit",
                desc=None,
                type="simulations",
            ),
            OpenFOAM=dict(
                cls="hera.simulations.openFoam.toolkit.OFToolkit",
                desc=None,
                type="simulations",
            ),
            WindProfile=dict(
                cls="hera.simulations.windProfile.toolkit.WindProfileToolkit",
                desc=None,
                type="simulations",
            ),
            GaussianDispersion=dict(
                cls="hera.simulations.gaussian.toolkit.gaussianToolkit",
                desc=None,
                type="simulations",
            ),
            machine_deep_learning=dict(
                cls="hera.simulations.machineLearningDeepLearning.toolkit.machineLearningDeepLearningToolkit",
                desc=None,
                type="simulations",
            ),
            experiment=dict(
                cls="hera.measurements.experiment.experiment.experimentHome",
                desc=None,
                type="measurements",
            ),
        )

        # Optional: keep a handle to the experiment toolkit (if available)
        self.experimentTK = None
        try:
            self.experimentTK = self.getToolkit(self.EXPERIMENT)
        except Exception:
            self.experimentTK = None

    # ------------------------------------------------------------------
    # Internal helper for repository config (uses generic dataToolkit)
    # ------------------------------------------------------------------

    def _get_data_toolkit(self, projectName: str = None):
        """
        Helper that returns a dataToolkit instance.

        We import dataToolkit lazily here to avoid circular imports
        between hera.toolkit and hera.utils.data.toolkit.
        dataToolkit itself works on the DEFAULT project internally.
        """
        from hera.utils.data.toolkit import dataToolkit
        return dataToolkit()

    # ------------------------------------------------------------------
    # Main API: getToolkit
    # ------------------------------------------------------------------

    def getToolkit(self, toolkitName: str, filesDirectory: Optional[str] = None, **kwargs):
        """
        Locate and instantiate a toolkit by name.

        This is the main public entry point for accessing toolkits in Hera.
        The method resolves the requested toolkit using the following order:
          1. Static built-in toolkit registry.
          2. Dynamically registered ToolkitDataSource documents (DB-backed).
          3. Experiment toolkits exposed via the experiment toolkit.

        Args:
            toolkitName (str): Logical name of the toolkit to load.
            filesDirectory (Optional[str]): Optional directory for toolkit file outputs.
            **kwargs: Additional keyword arguments forwarded to the toolkit constructor.
                      Commonly includes `projectName`.

        Returns:
            abstractToolkit: An initialized toolkit instance.

        Raises:
            ImportError: If the toolkit class cannot be imported.
            ValueError: If the toolkit cannot be found in any registry.
        """
        from hera.utils.data.toolkit import dataToolkit

        projectName = kwargs.pop("projectName", None) or self.projectName

        # 1) Static built-in toolkits
        if toolkitName in (self._toolkits or {}):
            info = self._toolkits[toolkitName]
            cls_path = info.get("cls")
            toolkit_cls = pydoc.locate(cls_path)
            if toolkit_cls is None:
                raise ImportError(f"Cannot locate class: {cls_path}")

            # Instantiate the toolkit in the resolved project context
            return toolkit_cls(
                projectName=projectName,
                filesDirectory=filesDirectory,
                **kwargs,
            )

        # 2) Dynamic toolkits registered as ToolkitDataSource of ToolkitHome
        dtk = dataToolkit()
        doc = dtk.getDataSourceDocument(datasourceName=toolkitName)
        if doc is not None:
            # self.logger.info(f"Found dynamic toolkit {toolkitName}. Loading")

            toolkitPath_raw = doc.getData()

            # ------------------------------------------------------------
            # Robust path resolution (same spirit as getExperiment)
            # ------------------------------------------------------------
            candidates = []

            # 1) Raw value
            candidates.append(toolkitPath_raw)

            # 2) Absolute relative to CWD
            try:
                candidates.append(os.path.abspath(toolkitPath_raw))
            except Exception:
                pass

            # 3) Relative to repo root if provided
            repo_root = os.environ.get("HERA_REPO_ROOT")
            if repo_root:
                candidates.append(os.path.join(repo_root, toolkitPath_raw))

            toolkitPath = None
            for cand in candidates:
                try:
                    cand_abs = os.path.abspath(cand)

                    # Valid if directory exists
                    if os.path.isdir(cand_abs):
                        toolkitPath = cand_abs
                        break

                    # Or if a single-file toolkit exists: <toolkitName>.py
                    py_file = os.path.join(cand_abs, f"{toolkitName}.py")
                    if os.path.isfile(py_file):
                        toolkitPath = cand_abs
                        break
                except Exception:
                    continue

            # Best-effort fallback (keeps error messages informative)
            if toolkitPath is None:
                toolkitPath = os.path.abspath(toolkitPath_raw)

            # ------------------------------------------------------------
            # Add toolkit path to sys.path (highest priority)
            # ------------------------------------------------------------
            if toolkitPath in sys.path:
                try:
                    sys.path.remove(toolkitPath)
                except ValueError:
                    pass
            sys.path.insert(0, toolkitPath)

            # self.logger.debug(f"Toolkit path (raw): {toolkitPath_raw}")
            # self.logger.debug(f"Toolkit path (resolved): {toolkitPath}")
            # self.logger.debug(f"Adding path {toolkitPath} to sys.path (priority)")

            # ------------------------------------------------------------
            # Import convention: <toolkitName>.<toolkitName>
            # ------------------------------------------------------------
            import_path = f"{toolkitName}.{toolkitName}"
            # self.logger.debug(f"Loading toolkit class: {import_path}")

            toolkit_cls = pydoc.locate(import_path)
            if toolkit_cls is None:
                raise ImportError(
                    f"Cannot find toolkit class {import_path} in {toolkitPath}"
                )

            return toolkit_cls(
                projectName=projectName,
                filesDirectory=filesDirectory,
                **kwargs,
            )
        # 3) Experiment toolkits fallback (experimentHome)
        if self.experimentTK is not None:
            try:
                return self.experimentTK.getExperiment(
                    experimentName=toolkitName,
                    filesDirectory=filesDirectory,
                )
            except Exception:
                # experimentHome does not recognize this experiment name
                pass

        # Nothing found in any registry
        raise ValueError(
            f"Toolkit '{toolkitName}' not found in static registry, ToolkitDataSource, "
            f"or experiment toolkit in project '{projectName}'."
        )

    # ------------------------------------------------------------------
    # Auto-register + get (kept for compatibility – but uses datasource API)
    # ------------------------------------------------------------------

    def auto_register_and_get(
        self,
        toolkitName: str,
        repositoryJSON: dict = None,
        repositoryName: Optional[str] = None,
        params: Optional[dict] = None,
        version: tuple = (0, 0, 1),
    ):
        """
           Automatically register a toolkit (if missing) and return an instance.

           This helper attempts to locate a toolkit class using repository metadata
           or JSON configuration, registers it as a ToolkitDataSource, and then
           returns an initialized toolkit instance.

           Args:
               toolkitName (str): Name of the toolkit to load.
               repositoryJSON (Optional[dict]): Optional repository metadata.
               repositoryName (Optional[str]): Target repository name.
               params (Optional[dict]): Toolkit initialization parameters.
               version (tuple): Toolkit version.

           Returns:
               abstractToolkit: An initialized toolkit instance.

           Raises:
               ValueError: If registration information is incomplete.
               ImportError: If the toolkit class cannot be imported.
           """
        from importlib import import_module

        params = params or {}
        classpath_hint = None
        projectName = self.projectName

        # 1) Classpath hint in the repository JSON
        if repositoryJSON:
            try:
                tk_section = repositoryJSON.get(toolkitName, {})
                reg = tk_section.get("Registry", {})
                classpath_hint = reg.get("classpath") or reg.get("cls")
            except Exception:
                pass

        if not classpath_hint:
            raise ValueError(
                f"auto_register_and_get: no classpath hint found for toolkit '{toolkitName}'. "
                f"Provide a 'Registry.classpath'/'cls' in repository JSON or seed a Toolkit document."
            )

        # Import the class
        mod_name, _, cls_name = classpath_hint.rpartition(".")
        if not mod_name or not cls_name:
            raise ValueError(f"Invalid classpath hint: '{classpath_hint}'")
        try:
            mod = import_module(mod_name)
            toolkit_cls = getattr(mod, cls_name)
        except Exception as exc:
            raise ImportError(f"Failed to import '{classpath_hint}' for toolkit '{toolkitName}'") from exc

        # Decide target repository for registration
        repo_to_use = (repositoryName or "").strip()
        if not repo_to_use:
            if projectName is None:
                raise ValueError(
                    "auto_register_and_get: projectName is None and no repositoryName provided. "
                    "Cannot resolve default repository."
                )
            repo_to_use = self.getDefaultRepository(projectName=projectName)
        if not repo_to_use:
            raise ValueError(
                f"auto_register_and_get: no target repository for project '{projectName}'. "
                f"Set a default repository or pass repositoryName explicitly."
            )

        # Register (idempotent if overwrite=True)
        self.registerToolkit(
            toolkitclass=toolkit_cls,
            datasource_name=toolkitName,
            params=params,
            version=version,
            overwrite=True,
            repositoryName=repo_to_use,
        )

        # Return an instance
        return self.getToolkit(toolkitName=toolkitName)

    # ------------------------------------------------------------------
    # Listing toolkits (static + dynamic)
    # ------------------------------------------------------------------

    from typing import Optional, List, Dict

    def getToolkitDocuments(self, name: Optional[str] = None) -> List[Dict]:
        """
        Single source of truth for listing toolkits.

        This method returns a uniform list of "document-like" dictionaries coming from:
          1) The static in-memory registry (self._toolkits).
          2) Dynamic DB-backed toolkit records (type='ToolkitDataSource').
          3) Experiments that are exposed via the 'experiment' toolkit
             (so that experiments appear as toolkits in the same view).

        Each returned dict has the general shape:
            {
                "toolkit": <name>,
                "desc": {
                    "classpath": <string>,
                    "type": <string>,
                    "source": <string>,
                    "repositoryName": <string>,
                    "version": <any>
                }
            }
        """
        from hera.utils.data.toolkit import dataToolkit

        docs: List[Dict] = []

        # ------------------------------------------------------------------
        # 1) Static toolkits from the in-memory registry
        # ------------------------------------------------------------------
        for tk_name, info in (self._toolkits or {}).items():
            if name and tk_name != name:
                continue

            docs.append(
                {
                    "toolkit": tk_name,
                    "desc": {
                        # Fully-qualified classpath of the toolkit implementation
                        "classpath": info.get("cls", ""),
                        # Logical type of the toolkit (e.g. 'measurements', 'simulations', ...)
                        "type": info.get("type", "measurements"),
                        # Static entries are considered 'internal'
                        "source": "internal",
                        # Static toolkits do not come from a specific repository
                        "repositoryName": "",
                        # No explicit version for static entries
                        "version": "",
                    },
                }
            )
        # ------------------------------------------------------------------
        # 2) Dynamic toolkits stored in the DB as ToolkitDataSource documents
        # ------------------------------------------------------------------
        dtk = dataToolkit()
        try:
            # The dataToolkit is used as a generic interface to measurements
            # and to the underlying MongoDB-backed documents.
            dyn_docs = dtk.getMeasurementsDocuments(type="ToolkitDataSource") or []
            for d in dyn_docs:
                try:
                    desc = getattr(d, "desc", {}) or {}
                    tk_name = desc.get("datasourceName") or getattr(d, "datasourceName", None)
                    if not tk_name:
                        continue
                    if name and tk_name != name:
                        continue
                    docs.append(
                        {
                            "toolkit": tk_name,
                            "desc": {
                                # Dynamic entries may carry a classpath for dynamic import
                                "classpath": getattr(d, "resource", ""),
                                # Toolkit logical type; default to 'measurements' if missing
                                "type": desc.get("type", "") or "measurements",
                                # DB-backed entries are marked as coming from the DB
                                "source": desc.get("source", "") or "db",
                                # Repository is taken from the desc or from the document itself
                                "repositoryName": desc.get("repository", "") or getattr(d, "repository", ""),
                                # Version may be saved as a list or any other structure
                                "version": tuple(desc.get("version", ())) or getattr(d, "version", ""),
                            },
                        }
                    )
                except Exception:
                    # Be forgiving in case some records are partially formed
                    continue
        except Exception:
            # If the project or DB is not available, we still return the static list.
            pass

        # ------------------------------------------------------------------
        # 3) Experiments exposed as toolkits (via the 'experiment' toolkit)
        # ------------------------------------------------------------------
        docs.extend(self.getExperimentToolkitDocuments(name=name))

        return docs

    def getExperimentToolkitDocuments(self, name: Optional[str] = None) -> List[Dict]:
        """
        Return experiment definitions as "toolkit-like" documents.

        The 'experiment' toolkit (experimentHome) exposes all experiments
        of the project via getExperimentsMap(). This helper converts them
        into the same normalized "document-like" shape used by
        getToolkitDocuments(), so that experiments appear in the unified
        toolkits table and can be discovered via the same CLI.

        Notes:
        - Experiments do not have a direct classpath here; they are
          instantiated via experimentHome.getExperiment(...), so the
          'classpath' field is left empty.
        - The 'type' is marked as 'experiment'.
        - The 'source' is marked as 'experiment' to distinguish them from
          static or DB-backed toolkits.
        """
        # If the experiment toolkit is not available, return an empty list
        if self.experimentTK is None:
            return []

        try:
            # experimentHome.getExperimentsMap() returns a dictionary where:
            #   keys   = experiment names
            #   values = experiment metadata / configuration
            exp_map = self.experimentTK.getExperimentsMap()
        except Exception:
            # If anything goes wrong while querying experiments, do not
            # break the unified toolkit listing.
            return []

        docs: List[Dict] = []
        for exp_name in exp_map.keys():
            if name and exp_name != name:
                continue

            docs.append(
                {
                    "toolkit": exp_name,
                    "desc": {
                        # Experiments are not imported via a direct classpath
                        "classpath": "",
                        # Logical type to mark this as an experiment
                        "type": "experiment",
                        # Source tag to differentiate experiments from static/DB toolkits
                        "source": "experiment",
                        # Experiments are not associated with a repository name here
                        "repositoryName": "",
                        # No explicit version is tracked at this layer
                        "version": "",
                    },
                }
            )

        return docs


    def getToolkitTable(self, projectName: Optional[str] = None):
        """
        Build a DataFrame listing all available toolkits (static, dynamic, and experiments).

        Parameters
        ----------
        projectName : str, optional
            The project name. If None, uses the current project.

        Returns
        -------
        pandas.DataFrame
            Table with columns: toolkit, cls, source, type, repositoryName, version.
        """
        docs = self.getToolkitDocuments(name=None, projectName=projectName) or []
        rows = []
        for d in docs:
            desc = d.get("desc", {})
            rows.append(
                {
                    "toolkit": d.get("toolkit", ""),
                    "cls": desc.get("classpath", ""),
                    "source": desc.get("source", ""),
                    "type": desc.get("type", ""),
                    "repositoryName": desc.get("repositoryName", ""),
                    "version": desc.get("version", ""),
                }
            )

        if not rows:
            return pd.DataFrame(
                columns=["toolkit", "cls", "source", "type", "repositoryName", "version"]
            )

        df = pd.DataFrame(rows).drop_duplicates(subset=["toolkit", "source"], keep="first")
        return df

    # ------------------------------------------------------------------
    # Registration helpers (default repository config)
    # ------------------------------------------------------------------

    def setDefaultRepository(self, *, projectName: str, repositoryName: str, overwrite: bool = True):
        """
        Persist default repository name for a project so future calls can omit --repository.
        The configuration is stored as a measurement document with type='RepositoryConfig'.
        """
        if not projectName:
            raise ValueError("setDefaultRepository: 'projectName' is required")
        if not repositoryName:
            raise ValueError("setDefaultRepository: 'repositoryName' is required")

        dt = self._get_data_toolkit(projectName=projectName)

        # delete previous config if exists (by type)
        if overwrite:
            old = dt.getMeasurementsDocuments(type="RepositoryConfig")
            for d in old:
                d.delete()

        desc = {"defaultRepository": repositoryName}

        df_arg = {}
        try:
            from hera.datalayer import datatypes as _dt
            dfmt = getattr(_dt, "JSON", None) or getattr(_dt, "json", None) or getattr(
                _dt, "TEXT", None
            )
            if dfmt is not None:
                df_arg["dataFormat"] = dfmt
        except Exception:
            pass

        return dt.addMeasurementsDocument(
            type="RepositoryConfig",
            resource=".",
            desc=desc,
            **df_arg,
        )

    def getDefaultRepository(self, *, projectName: str) -> str:
        """
        Read the saved default repository name (if exists). Returns '' if missing.
        """
        if not projectName:
            raise ValueError("getDefaultRepository: 'projectName' is required")

        dt = self._get_data_toolkit(projectName=projectName)
        docs = dt.getMeasurementsDocuments(type="RepositoryConfig")
        if not docs:
            return ""
        return docs[0].desc.get("defaultRepository", "") or ""

    # ------------------------------------------------------------------
    # Registration of toolkits using datasource interface
    # ------------------------------------------------------------------

    def registerToolkit(
            self,
            toolkit_name: str,
            toolkit_path: str,
            params: Optional[dict] = None,
            version = (0, 0, 1),
            overwrite: bool = False,
            **kwargs
    ):
        """
        Register a toolkit by name and filesystem path.

        The toolkit is stored as a ToolkitDataSource measurement document
        via the dataToolkit interface. When loaded later via ``getToolkit``,
        the path is added to ``sys.path`` and the toolkit class is imported
        using the convention ``<toolkit_name>.<toolkit_name>``.

        Parameters
        ----------
        toolkit_name : str
            The name of the toolkit to register.
        toolkit_path : str
            Path to the toolkit directory or file.
        params : dict, optional
            Additional parameters stored with the toolkit.
        version : tuple of int
            Version tuple (major, minor, patch). Default is (0, 0, 1).
        overwrite : bool
            If True, overwrite an existing registration.
        kwargs : dict
            Additional keyword arguments passed to ``addDataSource``.

        Returns
        -------
        document
            The created data source document.
        """
        from hera.utils.data.toolkit import dataToolkit

        if not toolkit_name:
            raise ValueError("toolkit_name must be provided")

        if not toolkit_path:
            raise ValueError("toolkit_path must be provided")

        # If a file is given, store its parent directory
        toolkit_path = os.path.abspath(toolkit_path)
        if os.path.isfile(toolkit_path):
            toolkit_path = os.path.dirname(toolkit_path)

        params = params or {}
        dtk = dataToolkit()
        dtk._allowWritingToDefaultProject = True
        return dtk.addDataSource(
            dataSourceName=toolkit_name,
            resource = toolkit_path,  # directory only
            dataFormat = "string",
            version=version,
            overwrite = overwrite,
            params = params
        )
    # ------------------------------------------------------------------
    # JSON import helpers
    # ------------------------------------------------------------------

    def import_toolkits_from_json(
        self,
        *,
        projectName: str,
        json_path: str,
        default_repository: str = None,
        overwrite: bool = True,
    ) -> list:
        """
        Read a JSON file and register all Toolkits it declares into the given project.
        (Uses registerToolkit -> datasource interface.)
        """
        import json
        from pydoc import locate

        if not projectName:
            raise ValueError("import_toolkits_from_json: projectName is required")

        with open(json_path, "r") as f:
            payload = json.load(f) or {}

        repo_from_json = (payload.get("repository") or "").strip()
        repo_to_use = repo_from_json or (default_repository or self.getDefaultRepository(projectName=projectName))
        if not repo_to_use:
            raise ValueError(
                f"No repository provided in JSON and no default repository set for project '{projectName}'. "
                f"Set one via ToolkitHome.setDefaultRepository(...) or pass default_repository."
            )

        toolkits = payload.get("toolkits") or []
        if not isinstance(toolkits, list):
            raise ValueError("import_toolkits_from_json: 'toolkits' must be a list")

        registered = []
        for item in toolkits:
            name = item.get("name")
            classpath = item.get("classpath")
            version = item.get("version", [0, 0, 1])
            params = item.get("parameters", {})

            if not name or not classpath:
                raise ValueError(f"Toolkit entry missing name/classpath: {item}")

            tk_class = locate(classpath)
            if tk_class is None:
                raise ImportError(f"Cannot locate class by classpath: {classpath}")

            self.registerToolkit(
                toolkitclass=tk_class,
                repositoryName=repo_to_use,
                datasource_name=name,
                params=params,
                version=tuple(int(x) for x in version),
                overwrite=overwrite,
            )
            registered.append(name)

        return registered

    def import_experiments_from_json(self, *, projectName: str, json_path: str) -> list:
        """
        Load experiments from JSON into the given project as measurements documents.

        Parameters
        ----------
        projectName : str
            The project to load experiments into.
        json_path : str
            Path to the JSON file containing experiment definitions.

        Returns
        -------
        list of str
            Names of the loaded experiments.
        """
        import json
        from hera.datalayer import Project

        if not projectName:
            raise ValueError("import_experiments_from_json: projectName is required")

        with open(json_path, "r") as f:
            payload = json.load(f) or {}

        experiments = payload.get("experiments") or []
        if not isinstance(experiments, list):
            raise ValueError("'experiments' must be a list in the JSON file")

        proj = Project(projectName=projectName)
        loaded = []
        base_dir = os.path.dirname(os.path.abspath(json_path))

        for exp in experiments:
            exp_name = exp.get("name")
            data_items = exp.get("data", [])

            for di in data_items:
                typ = di.get("type")
                resource = di.get("resource")
                data_fmt = di.get("dataFormat")
                desc = di.get("desc", {})
                is_rel = bool(di.get("isRelativePath", False))

                if not typ or not resource or not data_fmt:
                    continue

                res_path = resource
                if is_rel:
                    res_path = os.path.abspath(os.path.join(base_dir, resource))

                proj.addMeasurementsDocument(
                    type=typ,
                    resource=res_path,
                    dataFormat=data_fmt,
                    desc=desc,
                )

            if exp_name and exp_name not in loaded:
                loaded.append(exp_name)

        return loaded

getToolkit(toolkitName: str, filesDirectory: Optional[str] = None, **kwargs)

Locate and instantiate a toolkit by name.

This is the main public entry point for accessing toolkits in Hera. The method resolves the requested toolkit using the following order: 1. Static built-in toolkit registry. 2. Dynamically registered ToolkitDataSource documents (DB-backed). 3. Experiment toolkits exposed via the experiment toolkit.

Args: toolkitName (str): Logical name of the toolkit to load. filesDirectory (Optional[str]): Optional directory for toolkit file outputs. **kwargs: Additional keyword arguments forwarded to the toolkit constructor. Commonly includes projectName.

Returns: abstractToolkit: An initialized toolkit instance.

Raises: ImportError: If the toolkit class cannot be imported. ValueError: If the toolkit cannot be found in any registry.

Source code in hera/toolkit.py

def getToolkit(self, toolkitName: str, filesDirectory: Optional[str] = None, **kwargs):
    """
    Locate and instantiate a toolkit by name.

    This is the main public entry point for accessing toolkits in Hera.
    The method resolves the requested toolkit using the following order:
      1. Static built-in toolkit registry.
      2. Dynamically registered ToolkitDataSource documents (DB-backed).
      3. Experiment toolkits exposed via the experiment toolkit.

    Args:
        toolkitName (str): Logical name of the toolkit to load.
        filesDirectory (Optional[str]): Optional directory for toolkit file outputs.
        **kwargs: Additional keyword arguments forwarded to the toolkit constructor.
                  Commonly includes `projectName`.

    Returns:
        abstractToolkit: An initialized toolkit instance.

    Raises:
        ImportError: If the toolkit class cannot be imported.
        ValueError: If the toolkit cannot be found in any registry.
    """
    from hera.utils.data.toolkit import dataToolkit

    projectName = kwargs.pop("projectName", None) or self.projectName

    # 1) Static built-in toolkits
    if toolkitName in (self._toolkits or {}):
        info = self._toolkits[toolkitName]
        cls_path = info.get("cls")
        toolkit_cls = pydoc.locate(cls_path)
        if toolkit_cls is None:
            raise ImportError(f"Cannot locate class: {cls_path}")

        # Instantiate the toolkit in the resolved project context
        return toolkit_cls(
            projectName=projectName,
            filesDirectory=filesDirectory,
            **kwargs,
        )

    # 2) Dynamic toolkits registered as ToolkitDataSource of ToolkitHome
    dtk = dataToolkit()
    doc = dtk.getDataSourceDocument(datasourceName=toolkitName)
    if doc is not None:
        # self.logger.info(f"Found dynamic toolkit {toolkitName}. Loading")

        toolkitPath_raw = doc.getData()

        # ------------------------------------------------------------
        # Robust path resolution (same spirit as getExperiment)
        # ------------------------------------------------------------
        candidates = []

        # 1) Raw value
        candidates.append(toolkitPath_raw)

        # 2) Absolute relative to CWD
        try:
            candidates.append(os.path.abspath(toolkitPath_raw))
        except Exception:
            pass

        # 3) Relative to repo root if provided
        repo_root = os.environ.get("HERA_REPO_ROOT")
        if repo_root:
            candidates.append(os.path.join(repo_root, toolkitPath_raw))

        toolkitPath = None
        for cand in candidates:
            try:
                cand_abs = os.path.abspath(cand)

                # Valid if directory exists
                if os.path.isdir(cand_abs):
                    toolkitPath = cand_abs
                    break

                # Or if a single-file toolkit exists: <toolkitName>.py
                py_file = os.path.join(cand_abs, f"{toolkitName}.py")
                if os.path.isfile(py_file):
                    toolkitPath = cand_abs
                    break
            except Exception:
                continue

        # Best-effort fallback (keeps error messages informative)
        if toolkitPath is None:
            toolkitPath = os.path.abspath(toolkitPath_raw)

        # ------------------------------------------------------------
        # Add toolkit path to sys.path (highest priority)
        # ------------------------------------------------------------
        if toolkitPath in sys.path:
            try:
                sys.path.remove(toolkitPath)
            except ValueError:
                pass
        sys.path.insert(0, toolkitPath)

        # self.logger.debug(f"Toolkit path (raw): {toolkitPath_raw}")
        # self.logger.debug(f"Toolkit path (resolved): {toolkitPath}")
        # self.logger.debug(f"Adding path {toolkitPath} to sys.path (priority)")

        # ------------------------------------------------------------
        # Import convention: <toolkitName>.<toolkitName>
        # ------------------------------------------------------------
        import_path = f"{toolkitName}.{toolkitName}"
        # self.logger.debug(f"Loading toolkit class: {import_path}")

        toolkit_cls = pydoc.locate(import_path)
        if toolkit_cls is None:
            raise ImportError(
                f"Cannot find toolkit class {import_path} in {toolkitPath}"
            )

        return toolkit_cls(
            projectName=projectName,
            filesDirectory=filesDirectory,
            **kwargs,
        )
    # 3) Experiment toolkits fallback (experimentHome)
    if self.experimentTK is not None:
        try:
            return self.experimentTK.getExperiment(
                experimentName=toolkitName,
                filesDirectory=filesDirectory,
            )
        except Exception:
            # experimentHome does not recognize this experiment name
            pass

    # Nothing found in any registry
    raise ValueError(
        f"Toolkit '{toolkitName}' not found in static registry, ToolkitDataSource, "
        f"or experiment toolkit in project '{projectName}'."
    )

getToolkitTable(projectName: Optional[str] = None)

Build a DataFrame listing all available toolkits (static, dynamic, and experiments).

Parameters:

Name	Type	Description	Default
projectName	str	The project name. If None, uses the current project.	None

Returns:

Type	Description
DataFrame	Table with columns: toolkit, cls, source, type, repositoryName, version.

Source code in hera/toolkit.py

def getToolkitTable(self, projectName: Optional[str] = None):
    """
    Build a DataFrame listing all available toolkits (static, dynamic, and experiments).

    Parameters
    ----------
    projectName : str, optional
        The project name. If None, uses the current project.

    Returns
    -------
    pandas.DataFrame
        Table with columns: toolkit, cls, source, type, repositoryName, version.
    """
    docs = self.getToolkitDocuments(name=None, projectName=projectName) or []
    rows = []
    for d in docs:
        desc = d.get("desc", {})
        rows.append(
            {
                "toolkit": d.get("toolkit", ""),
                "cls": desc.get("classpath", ""),
                "source": desc.get("source", ""),
                "type": desc.get("type", ""),
                "repositoryName": desc.get("repositoryName", ""),
                "version": desc.get("version", ""),
            }
        )

    if not rows:
        return pd.DataFrame(
            columns=["toolkit", "cls", "source", "type", "repositoryName", "version"]
        )

    df = pd.DataFrame(rows).drop_duplicates(subset=["toolkit", "source"], keep="first")
    return df

registerToolkit(toolkit_name: str, toolkit_path: str, params: Optional[dict] = None, version=(0, 0, 1), overwrite: bool = False, **kwargs)

Register a toolkit by name and filesystem path.

The toolkit is stored as a ToolkitDataSource measurement document via the dataToolkit interface. When loaded later via getToolkit, the path is added to sys.path and the toolkit class is imported using the convention <toolkit_name>.<toolkit_name>.

Parameters:

Name	Type	Description	Default
toolkit_name	str	The name of the toolkit to register.	required
toolkit_path	str	Path to the toolkit directory or file.	required
params	dict	Additional parameters stored with the toolkit.	None
version	tuple of int	Version tuple (major, minor, patch). Default is (0, 0, 1).	(0, 0, 1)
overwrite	bool	If True, overwrite an existing registration.	False
kwargs	dict	Additional keyword arguments passed to addDataSource.	{}

Returns:

Type	Description
document	The created data source document.

Source code in hera/toolkit.py

def registerToolkit(
        self,
        toolkit_name: str,
        toolkit_path: str,
        params: Optional[dict] = None,
        version = (0, 0, 1),
        overwrite: bool = False,
        **kwargs
):
    """
    Register a toolkit by name and filesystem path.

    The toolkit is stored as a ToolkitDataSource measurement document
    via the dataToolkit interface. When loaded later via ``getToolkit``,
    the path is added to ``sys.path`` and the toolkit class is imported
    using the convention ``<toolkit_name>.<toolkit_name>``.

    Parameters
    ----------
    toolkit_name : str
        The name of the toolkit to register.
    toolkit_path : str
        Path to the toolkit directory or file.
    params : dict, optional
        Additional parameters stored with the toolkit.
    version : tuple of int
        Version tuple (major, minor, patch). Default is (0, 0, 1).
    overwrite : bool
        If True, overwrite an existing registration.
    kwargs : dict
        Additional keyword arguments passed to ``addDataSource``.

    Returns
    -------
    document
        The created data source document.
    """
    from hera.utils.data.toolkit import dataToolkit

    if not toolkit_name:
        raise ValueError("toolkit_name must be provided")

    if not toolkit_path:
        raise ValueError("toolkit_path must be provided")

    # If a file is given, store its parent directory
    toolkit_path = os.path.abspath(toolkit_path)
    if os.path.isfile(toolkit_path):
        toolkit_path = os.path.dirname(toolkit_path)

    params = params or {}
    dtk = dataToolkit()
    dtk._allowWritingToDefaultProject = True
    return dtk.addDataSource(
        dataSourceName=toolkit_name,
        resource = toolkit_path,  # directory only
        dataFormat = "string",
        version=version,
        overwrite = overwrite,
        params = params
    )

getToolkitDocuments(name: Optional[str] = None) -> List[Dict]

Single source of truth for listing toolkits.

This method returns a uniform list of "document-like" dictionaries coming from: 1) The static in-memory registry (self._toolkits). 2) Dynamic DB-backed toolkit records (type='ToolkitDataSource'). 3) Experiments that are exposed via the 'experiment' toolkit (so that experiments appear as toolkits in the same view).

Each returned dict has the general shape: { "toolkit": , "desc": { "classpath": , "type": , "source": , "repositoryName": , "version": } }

Source code in hera/toolkit.py

def getToolkitDocuments(self, name: Optional[str] = None) -> List[Dict]:
    """
    Single source of truth for listing toolkits.

    This method returns a uniform list of "document-like" dictionaries coming from:
      1) The static in-memory registry (self._toolkits).
      2) Dynamic DB-backed toolkit records (type='ToolkitDataSource').
      3) Experiments that are exposed via the 'experiment' toolkit
         (so that experiments appear as toolkits in the same view).

    Each returned dict has the general shape:
        {
            "toolkit": <name>,
            "desc": {
                "classpath": <string>,
                "type": <string>,
                "source": <string>,
                "repositoryName": <string>,
                "version": <any>
            }
        }
    """
    from hera.utils.data.toolkit import dataToolkit

    docs: List[Dict] = []

    # ------------------------------------------------------------------
    # 1) Static toolkits from the in-memory registry
    # ------------------------------------------------------------------
    for tk_name, info in (self._toolkits or {}).items():
        if name and tk_name != name:
            continue

        docs.append(
            {
                "toolkit": tk_name,
                "desc": {
                    # Fully-qualified classpath of the toolkit implementation
                    "classpath": info.get("cls", ""),
                    # Logical type of the toolkit (e.g. 'measurements', 'simulations', ...)
                    "type": info.get("type", "measurements"),
                    # Static entries are considered 'internal'
                    "source": "internal",
                    # Static toolkits do not come from a specific repository
                    "repositoryName": "",
                    # No explicit version for static entries
                    "version": "",
                },
            }
        )
    # ------------------------------------------------------------------
    # 2) Dynamic toolkits stored in the DB as ToolkitDataSource documents
    # ------------------------------------------------------------------
    dtk = dataToolkit()
    try:
        # The dataToolkit is used as a generic interface to measurements
        # and to the underlying MongoDB-backed documents.
        dyn_docs = dtk.getMeasurementsDocuments(type="ToolkitDataSource") or []
        for d in dyn_docs:
            try:
                desc = getattr(d, "desc", {}) or {}
                tk_name = desc.get("datasourceName") or getattr(d, "datasourceName", None)
                if not tk_name:
                    continue
                if name and tk_name != name:
                    continue
                docs.append(
                    {
                        "toolkit": tk_name,
                        "desc": {
                            # Dynamic entries may carry a classpath for dynamic import
                            "classpath": getattr(d, "resource", ""),
                            # Toolkit logical type; default to 'measurements' if missing
                            "type": desc.get("type", "") or "measurements",
                            # DB-backed entries are marked as coming from the DB
                            "source": desc.get("source", "") or "db",
                            # Repository is taken from the desc or from the document itself
                            "repositoryName": desc.get("repository", "") or getattr(d, "repository", ""),
                            # Version may be saved as a list or any other structure
                            "version": tuple(desc.get("version", ())) or getattr(d, "version", ""),
                        },
                    }
                )
            except Exception:
                # Be forgiving in case some records are partially formed
                continue
    except Exception:
        # If the project or DB is not available, we still return the static list.
        pass

    # ------------------------------------------------------------------
    # 3) Experiments exposed as toolkits (via the 'experiment' toolkit)
    # ------------------------------------------------------------------
    docs.extend(self.getExperimentToolkitDocuments(name=name))

    return docs

hera.toolkit.abstractToolkit

Bases: Project

Base class for all Toolkits.

• Like Project, it is initialized with a project name. If the toolkit works on data, the data should be present in that project.

• Inherits from Project and therefore exposes all the datalayer functions.

• Holds the toolkit name, and references to the analysis and presentation layers.

• Adds a mechanism (setConfig, getConfig) for saving configuration in the DB. The settings are specific for a project.

• Adds a data source mechanism to list, get, and add data sources.

• A data source is always saved as a measurement document of type TOOLKIT_DATASOURCE_TYPE.

• Each source has the following properties in the description: * datasourceName : str * toolkit : str * version : tuple (major, minor, patch)
• The toolkit can have a default source version per project.

Source code in hera/toolkit.py

class abstractToolkit(Project):
    """
    Base class for all Toolkits.

    *  Like Project, it is initialized with a project name.
       If the toolkit works on data, the data should be present in that project.

    *  Inherits from Project and therefore exposes all the datalayer functions.

    *  Holds the toolkit name, and references to the analysis and presentation layers.

    *  Adds a mechanism (setConfig, getConfig) for saving configuration in the DB.
       The settings are specific for a project.

    *  Adds a data source mechanism to list, get, and add data sources.

       - A data source is always saved as a measurement document of type
         ``TOOLKIT_DATASOURCE_TYPE``.
       - Each source has the following properties in the description:
             * datasourceName : str
             * toolkit : str
             * version : tuple (major, minor, patch)
       - The toolkit can have a default source version per project.
    """

    _toolkitname = None
    _projectName = None

    _analysis = None
    _presentation = None

    @property
    def presentation(self):
        """
        Access the presentation layer.

        Returns
        -------
        object or None
        """
        return self._presentation

    @property
    def analysis(self):
        """
        Access the analysis (data) layer.

        Returns
        -------
        object or None
        """
        return self._analysis

    @property
    def toolkitName(self):
        """
        The name of the toolkit.

        Returns
        -------
        str
        """
        return self._toolkitname

    @property
    def projectName(self):
        """
        The name of the project.

        Returns
        -------
        str
        """
        return self._projectName

    def __init__(self, toolkitName: str, projectName: Optional[str] = None,
                 connectionName: Optional[str] = None, filesDirectory: Optional[str] = None):
        """
        Initialize a new toolkit.

        Parameters
        ----------
        toolkitName : str
            The name of the toolkit.
        projectName : str, optional
            The project that the toolkit works in.
            If None, Project's automatic project-name mechanism is used.
        connectionName : str, optional
            The name of the DB connection. If None, uses the current OS username.
        filesDirectory : str, optional
            Directory to save datasource files.
        """
        super().__init__(projectName=projectName, filesDirectory=filesDirectory, connectionName=connectionName)
        logger = get_classMethod_logger(self, "init")
        self._toolkitname = toolkitName

    @property
    def classLoggerName(self):
        """
        The logger name for the current class and method context.

        Returns
        -------
        str
        """
        return str(get_classMethod_logger(self, "{the_function_name}")).split(" ")[1]

    # ------------------------------------------------------------------
    # Document overrides — automatically tag with toolkit name
    # ------------------------------------------------------------------

    def addCacheDocument(self, resource="", dataFormat="string", type="", desc={}):
        """
        Add a cache document, automatically tagging it with the toolkit name.

        See ``Project.addCacheDocument`` for parameter details.
        """
        if self.toolkitName is not None:
            desc.setdefault(TOOLKIT_TOOLKITNAME_FIELD, self.toolkitName)
        return super().addCacheDocument(resource, dataFormat, type, desc)

    def addMeasurementsDocument(self, resource="", dataFormat="string", type="", desc={}):
        """
        Add a measurements document, automatically tagging it with the toolkit name.

        See ``Project.addMeasurementsDocument`` for parameter details.
        """
        if self.toolkitName is not None:
            desc.setdefault(TOOLKIT_TOOLKITNAME_FIELD, self.toolkitName)
        return super().addMeasurementsDocument(resource, dataFormat, type, desc)

    def addSimulationsDocument(self, resource="", dataFormat="string", type="", desc={}):
        """
        Add a simulations document, automatically tagging it with the toolkit name.

        See ``Project.addSimulationsDocument`` for parameter details.
        """
        if self.toolkitName is not None:
            desc.setdefault(TOOLKIT_TOOLKITNAME_FIELD, self.toolkitName)
        return super().addSimulationsDocument(resource, dataFormat, type, desc)

    # ------------------------------------------------------------------
    # Data sources API
    # ------------------------------------------------------------------

    def getDataSourceList(self, **filters) -> List[str]:
        """
        Return a list of data source names for this toolkit.

        Parameters
        ----------
        filters : dict
            Additional query filters.

        Returns
        -------
        list of str
            Data source names.
        """
        docList = self.getMeasurementsDocuments(
            type=TOOLKIT_DATASOURCE_TYPE,
            toolkit=self.toolkitName,
            **filters,
        )
        return [doc['desc']['datasourceName'] for doc in docList]

    def getDataSourceMap(self, **filters) -> List[Dict[str, Any]]:
        """
        Return a list of all data sources and their versions for this toolkit.

        Parameters
        ----------
        filters : dict
            Additional query filters.

        Returns
        -------
        list of dict
            Each dict contains dataFormat, resource, and all desc fields.
        """
        docList = self.getMeasurementsDocuments(
            type=TOOLKIT_DATASOURCE_TYPE,
            toolkit=self.toolkitName,
            **filters,
        )
        ret = []
        for doc in docList:
            dta = dict(dataFormat=doc['dataFormat'], resource=doc['resource'])
            dta.update(doc.desc)
            ret.append(dta)
        return ret

    def getDataSourceTable(self, **filters) -> pd.DataFrame:
        """
        Build a pandas DataFrame from all data sources of this toolkit.

        Parameters
        ----------
        filters : dict
            Additional query filters.

        Returns
        -------
        pandas.DataFrame
        """
        tables = []
        for sourceMap in self.getDataSourceMap(**filters):
            table = pd.json_normalize(sourceMap)
            tables.append(table)

        if not tables:
            return pd.DataFrame()
        else:
            return pd.concat(tables, ignore_index=True)

    def getDataSourceDocumentsList(self, **kwargs):
        """
        Return all data source documents associated with this toolkit.

        Parameters
        ----------
        kwargs : dict
            Additional query filters.

        Returns
        -------
        list
            List of mongoengine documents.
        """
        queryDict = {
            "type": TOOLKIT_DATASOURCE_TYPE,
            TOOLKIT_TOOLKITNAME_FIELD: self.toolkitName,
        }
        queryDict.update(**kwargs)
        return self.getMeasurementsDocuments(**queryDict)

    def getDataSourceDocument(self, datasourceName: Optional[str], version=None, **filters):
        """
        Return the document of a data source.

        If version is not specified, returns the default version (if configured)
        or the latest version. When multiple versions exist and no default is
        set, the latest version is selected and automatically persisted as the
        default in the project config for stable subsequent calls.

        Parameters
        ----------
        datasourceName : str or None
            The name of the data source. If None, return the default source (if set).
        version : tuple, optional
            The version to retrieve. If None, returns the default or latest.
        filters : dict
            Additional query filters.

        Returns
        -------
        document or None
            The data source document, or None if not found.
        """
        if datasourceName is not None:
            filters[TOOLKIT_DATASOURCE_NAME] = datasourceName
        if version is not None:
            filters[TOOLKIT_DATASOURCE_VERSION] = version
        else:
            try:
                defaultVersion = self.getConfig()[f"{datasourceName}_defaultVersion"]
                filters[TOOLKIT_DATASOURCE_VERSION] = defaultVersion
            except Exception:
                pass

        filters[TOOLKIT_TOOLKITNAME_FIELD] = self.toolkitName

        docList = self.getMeasurementsDocuments(
            type=TOOLKIT_DATASOURCE_TYPE,
            **filters,
        )

        if len(docList) == 0:
            ret = None
        elif len(docList) == 1:
            ret = docList[0]
        else:
            versionsList = [doc['desc']['version'] for doc in docList]
            latestVersion = max(versionsList)
            docList = [doc for doc in docList if doc['desc']['version'] == latestVersion]
            ret = docList[0]

            # No default was set and multiple versions exist — persist the
            # latest version as the default so subsequent calls are stable.
            if version is None and datasourceName is not None:
                try:
                    self.setConfig(**{f"{datasourceName}_defaultVersion": latestVersion})
                except Exception:
                    pass

        return ret

    def getToolkitDocument(self, toolkit_name: str):
        """
        Find a dynamic toolkit document by name.

        Searches by ``desc.datasourceName`` first, then falls back to scanning
        all ToolkitDataSource documents.

        Parameters
        ----------
        toolkit_name : str
            The toolkit name to search for.

        Returns
        -------
        document or None
            The mongoengine document, or None if not found.
        """
        try:
            q = self.getMeasurementsDocuments(
                type="ToolkitDataSource", datasourceName=toolkit_name
            )
            if q and len(q) > 0:
                return q[0]
        except Exception:
            pass

        try:
            q = self.getMeasurementsDocuments(type="ToolkitDataSource")
            for d in q:
                desc = d.desc or {}
                if desc.get("datasourceName") == toolkit_name or desc.get("toolkit") == toolkit_name:
                    return d
        except Exception:
            pass

        try:
            q = self.getDataSourceDocuments(datasourceName=toolkit_name)
            if q and len(q) > 0:
                return q[0]
        except Exception:
            pass

        return None

    def getDataSourceDocuments(self, datasourceName, version=None, **filters):
        """
        Return a list containing the data source document (for API symmetry).

        Parameters
        ----------
        datasourceName : str
            The name of the data source.
        version : tuple, optional
            The version to retrieve.
        filters : dict
            Additional query filters.

        Returns
        -------
        list
            A list containing the document, or an empty list if not found.
        """
        doc = self.getDataSourceDocument(datasourceName=datasourceName, version=version, **filters)
        return [] if doc is None else [doc]

    def getDataSourceData(self, datasourceName=None, version=None, **filters):
        """
        Return the data payload of a data source.

        Parameters
        ----------
        datasourceName : str, optional
            The name of the data source. If None, return the default source.
        version : tuple, optional
            The version to retrieve. If None, returns the latest.
        filters : dict
            Additional query filters.

        Returns
        -------
        object or None
            The loaded data, or None if the data source is not found.
        """
        filters[TOOLKIT_TOOLKITNAME_FIELD] = self.toolkitName
        doc = self.getDataSourceDocument(datasourceName=datasourceName, version=version, **filters)
        return None if doc is None else doc.getData()

    def addDataSource(self, dataSourceName: str, resource: str, dataFormat: str,
                      version=(0, 0, 1), overwrite: bool = False, **kwargs):
        """
        Add a data source to the toolkit.

        The type is always ``TOOLKIT_DATASOURCE_TYPE``.
        The toolkit name is added automatically to the description.

        Parameters
        ----------
        dataSourceName : str
            The name of the data source.
        resource : str
            The resource path or data.
        dataFormat : str
            A data format string from ``datatypes``.
        version : tuple of int
            A 3-tuple version (major, minor, patch). Default is (0, 0, 1).
        overwrite : bool
            If True, overwrite an existing data source with the same name and version.
        kwargs : dict
            Additional metadata fields for the description.

        Returns
        -------
        document
            The newly created data source document.

        Raises
        ------
        ValueError
            If the data source already exists and overwrite is False.
        """
        kwargs[TOOLKIT_TOOLKITNAME_FIELD] = self.toolkitName
        kwargs[TOOLKIT_DATASOURCE_NAME] = dataSourceName
        kwargs[TOOLKIT_DATASOURCE_VERSION] = version

        if (self.getDataSourceDocument(dataSourceName, version=version) is None) or overwrite:
            if self.getDataSourceDocument(dataSourceName, version=version) is not None:
                delargs = {
                    TOOLKIT_DATASOURCE_NAME: dataSourceName,
                    TOOLKIT_DATASOURCE_VERSION: version,
                }
                self.deleteDataSource(**delargs)

            doc = self.addMeasurementsDocument(
                type=TOOLKIT_DATASOURCE_TYPE,
                resource=resource,
                dataFormat=dataFormat,
                desc=kwargs,
            )
        else:
            raise ValueError(
                f"Record {dataSourceName} (version {version}) already exists in project "
                f"{self.projectName}. Use overwrite=True to overwrite the existing document."
            )

        return doc

    def deleteDataSource(self, datasourceName, version=None, **filters):
        """
        Delete a data source document.

        Parameters
        ----------
        datasourceName : str
            The name of the data source.
        version : tuple, optional
            The version to delete.
        filters : dict
            Additional query filters.

        Returns
        -------
        document or None
            The deleted document, or None if not found.
        """
        doc = self.getDataSourceDocument(datasourceName=datasourceName, version=version, **filters)
        if doc is not None:
            doc.delete()
        return doc

    def setDataSourceDefaultVersion(self, datasourceName: str, version: tuple):
        """
        Set the default version for a given data source.

        Parameters
        ----------
        datasourceName : str
            The name of the data source.
        version : tuple
            The version to set as default.

        Raises
        ------
        ValueError
            If no data source with the given name and version exists.
        """
        if len(
            self.getMeasurementsDocuments(
                type=TOOLKIT_DATASOURCE_TYPE,
                **{"datasourceName": datasourceName, "version": version},
            )
        ) == 0:
            raise ValueError(f"No DataSource with name={datasourceName} and version={version}.")

        self.setConfig(**{f"{datasourceName}_defaultVersion": version})
        print(f"{version} for dataSource {datasourceName} is now set to default.")

addDataSource(dataSourceName: str, resource: str, dataFormat: str, version=(0, 0, 1), overwrite: bool = False, **kwargs)

Add a data source to the toolkit.

The type is always TOOLKIT_DATASOURCE_TYPE. The toolkit name is added automatically to the description.

Parameters:

Name	Type	Description	Default
dataSourceName	str	The name of the data source.	required
resource	str	The resource path or data.	required
dataFormat	str	A data format string from datatypes.	required
version	tuple of int	A 3-tuple version (major, minor, patch). Default is (0, 0, 1).	(0, 0, 1)
overwrite	bool	If True, overwrite an existing data source with the same name and version.	False
kwargs	dict	Additional metadata fields for the description.	{}

Returns:

Type	Description
document	The newly created data source document.

Raises:

Type	Description
ValueError	If the data source already exists and overwrite is False.

Source code in hera/toolkit.py

def addDataSource(self, dataSourceName: str, resource: str, dataFormat: str,
                  version=(0, 0, 1), overwrite: bool = False, **kwargs):
    """
    Add a data source to the toolkit.

    The type is always ``TOOLKIT_DATASOURCE_TYPE``.
    The toolkit name is added automatically to the description.

    Parameters
    ----------
    dataSourceName : str
        The name of the data source.
    resource : str
        The resource path or data.
    dataFormat : str
        A data format string from ``datatypes``.
    version : tuple of int
        A 3-tuple version (major, minor, patch). Default is (0, 0, 1).
    overwrite : bool
        If True, overwrite an existing data source with the same name and version.
    kwargs : dict
        Additional metadata fields for the description.

    Returns
    -------
    document
        The newly created data source document.

    Raises
    ------
    ValueError
        If the data source already exists and overwrite is False.
    """
    kwargs[TOOLKIT_TOOLKITNAME_FIELD] = self.toolkitName
    kwargs[TOOLKIT_DATASOURCE_NAME] = dataSourceName
    kwargs[TOOLKIT_DATASOURCE_VERSION] = version

    if (self.getDataSourceDocument(dataSourceName, version=version) is None) or overwrite:
        if self.getDataSourceDocument(dataSourceName, version=version) is not None:
            delargs = {
                TOOLKIT_DATASOURCE_NAME: dataSourceName,
                TOOLKIT_DATASOURCE_VERSION: version,
            }
            self.deleteDataSource(**delargs)

        doc = self.addMeasurementsDocument(
            type=TOOLKIT_DATASOURCE_TYPE,
            resource=resource,
            dataFormat=dataFormat,
            desc=kwargs,
        )
    else:
        raise ValueError(
            f"Record {dataSourceName} (version {version}) already exists in project "
            f"{self.projectName}. Use overwrite=True to overwrite the existing document."
        )

    return doc

getDataSourceDocument(datasourceName: Optional[str], version=None, **filters)

Return the document of a data source.

If version is not specified, returns the default version (if configured) or the latest version. When multiple versions exist and no default is set, the latest version is selected and automatically persisted as the default in the project config for stable subsequent calls.

Parameters:

Name	Type	Description	Default
datasourceName	str or None	The name of the data source. If None, return the default source (if set).	required
version	tuple	The version to retrieve. If None, returns the default or latest.	None
filters	dict	Additional query filters.	{}

Returns:

Type	Description
document or None	The data source document, or None if not found.

Source code in hera/toolkit.py

def getDataSourceDocument(self, datasourceName: Optional[str], version=None, **filters):
    """
    Return the document of a data source.

    If version is not specified, returns the default version (if configured)
    or the latest version. When multiple versions exist and no default is
    set, the latest version is selected and automatically persisted as the
    default in the project config for stable subsequent calls.

    Parameters
    ----------
    datasourceName : str or None
        The name of the data source. If None, return the default source (if set).
    version : tuple, optional
        The version to retrieve. If None, returns the default or latest.
    filters : dict
        Additional query filters.

    Returns
    -------
    document or None
        The data source document, or None if not found.
    """
    if datasourceName is not None:
        filters[TOOLKIT_DATASOURCE_NAME] = datasourceName
    if version is not None:
        filters[TOOLKIT_DATASOURCE_VERSION] = version
    else:
        try:
            defaultVersion = self.getConfig()[f"{datasourceName}_defaultVersion"]
            filters[TOOLKIT_DATASOURCE_VERSION] = defaultVersion
        except Exception:
            pass

    filters[TOOLKIT_TOOLKITNAME_FIELD] = self.toolkitName

    docList = self.getMeasurementsDocuments(
        type=TOOLKIT_DATASOURCE_TYPE,
        **filters,
    )

    if len(docList) == 0:
        ret = None
    elif len(docList) == 1:
        ret = docList[0]
    else:
        versionsList = [doc['desc']['version'] for doc in docList]
        latestVersion = max(versionsList)
        docList = [doc for doc in docList if doc['desc']['version'] == latestVersion]
        ret = docList[0]

        # No default was set and multiple versions exist — persist the
        # latest version as the default so subsequent calls are stable.
        if version is None and datasourceName is not None:
            try:
                self.setConfig(**{f"{datasourceName}_defaultVersion": latestVersion})
            except Exception:
                pass

    return ret

getDataSourceData(datasourceName=None, version=None, **filters)

Return the data payload of a data source.

Parameters:

Name	Type	Description	Default
datasourceName	str	The name of the data source. If None, return the default source.	None
version	tuple	The version to retrieve. If None, returns the latest.	None
filters	dict	Additional query filters.	{}

Returns:

Type	Description
object or None	The loaded data, or None if the data source is not found.

Source code in hera/toolkit.py

def getDataSourceData(self, datasourceName=None, version=None, **filters):
    """
    Return the data payload of a data source.

    Parameters
    ----------
    datasourceName : str, optional
        The name of the data source. If None, return the default source.
    version : tuple, optional
        The version to retrieve. If None, returns the latest.
    filters : dict
        Additional query filters.

    Returns
    -------
    object or None
        The loaded data, or None if the data source is not found.
    """
    filters[TOOLKIT_TOOLKITNAME_FIELD] = self.toolkitName
    doc = self.getDataSourceDocument(datasourceName=datasourceName, version=version, **filters)
    return None if doc is None else doc.getData()

getDataSourceList(**filters) -> List[str]

Return a list of data source names for this toolkit.

Parameters:

Name	Type	Description	Default
filters	dict	Additional query filters.	{}

Returns:

Type	Description
list of str	Data source names.

Source code in hera/toolkit.py

def getDataSourceList(self, **filters) -> List[str]:
    """
    Return a list of data source names for this toolkit.

    Parameters
    ----------
    filters : dict
        Additional query filters.

    Returns
    -------
    list of str
        Data source names.
    """
    docList = self.getMeasurementsDocuments(
        type=TOOLKIT_DATASOURCE_TYPE,
        toolkit=self.toolkitName,
        **filters,
    )
    return [doc['desc']['datasourceName'] for doc in docList]

getDataSourceTable(**filters) -> pd.DataFrame

Build a pandas DataFrame from all data sources of this toolkit.

Parameters:

Name	Type	Description	Default
filters	dict	Additional query filters.	{}

Returns:

Type	Description
DataFrame

Source code in hera/toolkit.py

def getDataSourceTable(self, **filters) -> pd.DataFrame:
    """
    Build a pandas DataFrame from all data sources of this toolkit.

    Parameters
    ----------
    filters : dict
        Additional query filters.

    Returns
    -------
    pandas.DataFrame
    """
    tables = []
    for sourceMap in self.getDataSourceMap(**filters):
        table = pd.json_normalize(sourceMap)
        tables.append(table)

    if not tables:
        return pd.DataFrame()
    else:
        return pd.concat(tables, ignore_index=True)

deleteDataSource(datasourceName, version=None, **filters)

Delete a data source document.

Parameters:

Name	Type	Description	Default
datasourceName	str	The name of the data source.	required
version	tuple	The version to delete.	None
filters	dict	Additional query filters.	{}

Returns:

Type	Description
document or None	The deleted document, or None if not found.

Source code in hera/toolkit.py

def deleteDataSource(self, datasourceName, version=None, **filters):
    """
    Delete a data source document.

    Parameters
    ----------
    datasourceName : str
        The name of the data source.
    version : tuple, optional
        The version to delete.
    filters : dict
        Additional query filters.

    Returns
    -------
    document or None
        The deleted document, or None if not found.
    """
    doc = self.getDataSourceDocument(datasourceName=datasourceName, version=version, **filters)
    if doc is not None:
        doc.delete()
    return doc

hera.datalayer.project.Project

Provides a simple interface to the data of a specific project.

The class has all the following functions for the measurements.old, simulations.old and cache.

Measurements

• getMeasurementsDocumentsAsDict"
• getMeasurementsDocuments
• addMeasurementsDocument
• deleteMeasurementsDocuments

Simulations

• getSimulationsDocumentsAsDict"
• getSimulationsDocuments
• addSimulationsDocument
• deleteSimulationsDocuments

Cache

• getCacheDocumentsAsDict"
• getCacheDocuments
• addCacheDocument
• deleteCacheDocuments

Source code in hera/datalayer/project.py

class Project:
    """
        Provides a simple interface to the data of a specific project.

        The class has all the following functions for the measurements.old, simulations.old and cache.

        **Measurements**


        -  getMeasurementsDocumentsAsDict"
        -  getMeasurementsDocuments
        -  addMeasurementsDocument
        -  deleteMeasurementsDocuments

        **Simulations**

        -  getSimulationsDocumentsAsDict"
        -  getSimulationsDocuments
        -  addSimulationsDocument
        -  deleteSimulationsDocuments

        **Cache**

        -  getCacheDocumentsAsDict"
        -  getCacheDocuments
        -  addCacheDocument
        -  deleteCacheDocuments

    """
    _projectName = None
    _all          = None
    _measurements = None
    _cache     = None
    _simulations  = None

    datatypes = datatypes

    DEFAULTPROJECT = "defaultProject"

    _allowWritingToDefaultProject = None # A flag to allow the update of the default project. Used to add the datasources to it.

    _FilesDirectory = None

    @property
    def FilesDirectory(self):
        """
            The directory to save files (when creating files).
        :return:
        """
        return self._FilesDirectory

    @property
    def filesDirectory(self):
        """
            The directory to save files (when creating files).
        :return:
        """
        return self._FilesDirectory



    @property
    def measurements(self) -> Measurements_Collection:
        """
            Access the measurement type documents.

    Returns
    -------
            hera.datalayer.collection.Measurements_Collection
        """
        return self._measurements

    @property
    def cache(self) -> Cache_Collection:
        """
            Access the Cache type documents.

    Returns
    -------
            hera.datalayer.collection.Cache_Collection

        """
        return self._cache

    @property
    def all(self) -> AbstractCollection:
        """
            Access to all document types.

    Returns
    -------
            hera.datalayer.collection.AbstractCollection

        """
        return self._all

    @property
    def projectName(self):
        """
        The name of the current project.

        Returns
        -------
        str
        """
        return self._projectName


    def setConfig(self,keep_old_values=True, **kwargs):
        """
            Create a config document or updates an existing config document.
        """
        if self._projectName == self.DEFAULTPROJECT:
            err = """
                    Default project was initiated and it is cannot use configuration.

                    This error os obtained because you have initiated the project or the toolkit 
                    using projectName=None, and you don't case a caseConfiguration.json in the directory that 
                    specifies the project name. 

                    To solve this project either 
                    - Create a new project in the directory: 
                      run  
                        >> hera-project project create <the directory name>
                      in the parent directory 
                    - Create manually the file caseConfiguration.json in the directory with the key 'projectName' that holds 
                      the name of the project as a string.     
            """
            raise ValueError(err)

        doc = self._getConfigDocument()
        if keep_old_values:
            update_kwargs = {
                f"set__desc__{k}": v
                for k, v in kwargs.items()
            }
            doc.update(**update_kwargs)
        else:
            doc.update(set__desc=kwargs)


    def __init__(self, projectName=None, connectionName=None, configurationPath=None,filesDirectory=None):
        """
            Initialize the project class.

        Parameters
        ----------
        projectName: str default None
                The name of the project.
                If projectName is None, try to load it from the [configurationPath]/caseConfiguration.json
                if configurationPath is None, load it from the current directory.
                the structure of this file is

                ```
                    {
                        "projectName" : [project Name]
                    }
                ```

        connectionName: str
                The name of the DB connection. If not specified, use the
                connection with the linux user name.

        :param configurationPath: str
                The path to the caseConfiguration.json file.
                If not supplied, use the current directory.
        """
        logger = get_classMethod_logger(self,"init")
        self._allowWritingToDefaultProject = False

        if projectName is None:
            configurationPath = os.getcwd() if configurationPath is None else configurationPath
            confFile = os.path.join(configurationPath, "caseConfiguration.json")
            logger.debug(f"projectName is None, try to load file {confFile}")
            if os.path.exists(confFile):
                logger.debug(f"Load as JSON")
                configuration = loadJSON(confFile)
                if 'projectName' not in configuration:
                    err = f"Got projectName=None and the key 'projectName' does not exist in the JSON. "
                    err += """configuration should be :
{
    'projectName' : [project name]
}                                        
"""
                    logger.error(err)
                    raise ValueError(err)
                else:
                    projectName   = configuration['projectName']
            else:
                logger.debug(f"configuration file {confFile} is not found. Using the default project: DEFAULTPROJECT={self.DEFAULTPROJECT}")
                projectName = self.DEFAULTPROJECT

        logger.info(f"Initializing with logger {projectName}")
        self._projectName = projectName

        self._measurements  = Measurements_Collection(connectionName=connectionName)
        self._cache      = Cache_Collection(connectionName=connectionName)
        self._simulations   = Simulations_Collection(connectionName=connectionName)
        self._all           =   AbstractCollection(connectionName=connectionName)

        if self.projectName != self.DEFAULTPROJECT:
            logger.info(f"Attempting to get default directory from the disk")
            savedFilesDirectory = self.getConfig().get("filesDirectory", None)
        else:
            logger.info(f"Default project, setting to current directory")
            savedFilesDirectory = None

        if savedFilesDirectory is None:
            if filesDirectory is None:
                filesDirectory = os.path.join(os.path.abspath(os.getcwd()), projectName)
            else:
                filesDirectory = os.path.abspath(filesDirectory)

            if self.projectName != self.DEFAULTPROJECT:
                logger.info(f"Files directory is not saved for the project, using {filesDirectory}")
                self.setConfig(filesDirectory=filesDirectory)
        else:
            filesDirectory = savedFilesDirectory

        os.makedirs(os.path.abspath(filesDirectory),exist_ok=True)
        self._FilesDirectory = filesDirectory

    @staticmethod
    def _batched_cursor(cur, export_chunk_size):# -> Generator[list, Any, None]:
        """
            turns an iterator to an iterator that returns batches of the original.
            if you cur is a documents list iterator and export_chunk_size=5 then each iteration of  
            _batched_cursor will return a list with 5 documents

        Parameters
        ----------
        cur: Iterator

        export_chunk_size: int

        """
        while True:
            batch = list(itertools.islice(cur, export_chunk_size))
            if not batch:
                break
            else:
                yield batch

    def export(self, path, export_chunk_size=1024, show_progressbar=True):
        """
            exports the project in chunks contained to one zip file

        Parameters
        ----------
        path: str
                the path to the zip file to be created, overrides file if already exists

        export_chunk_size: int
                the number of documents in each chunk

        show_progressbar: bool
        """
        docs_cursor = self._all._metadataCol._get_collection().find({"projectName":self.projectName})
        with zipfile.ZipFile(path, mode='w', compression=zipfile.ZIP_DEFLATED) as zf:
            i = 0
            if show_progressbar:
                docs_iterator= tqdm(self._batched_cursor(docs_cursor, export_chunk_size),
                                    desc="Exporting documents", unit="batchedDocs", unit_scale=True) 
            else:
                docs_iterator = docs_cursor
            for docs_batch in docs_iterator:
                filename = f"chunk_{i}"
                with zf.open(filename, 'w') as zf_archive:
                    pickle.dump(docs_batch, zf_archive, protocol=pickle.HIGHEST_PROTOCOL)
                i+=1

    @staticmethod
    def _iter_pickled_docs(zf, return_batched):
        """
            creates an iterator over the documents in an exported project,
            can either returned the batched data or single documents 

        Parameters
        ----------
        zf: zipfile.ZipFile
                handler for the exported project

        return_batched: str
                should each iteration return an entire batch(with the size it was exported as)
        """
        for name in zf.namelist():
            with zf.open(name) as f:
                depickled_docs_batch=pickle.load(f)
                if return_batched:
                    yield depickled_docs_batch
                else:
                    for depickled_doc in depickled_docs_batch:
                        yield depickled_doc 

    def updateProjectNameOnDoc(self, doc_son):
        """
            updates the projectName field of a document to be assigned to this project 

        Parameters
        ----------
        doc_son: str
                the document to be updated
        """
        doc_son.update({"projectName": self.projectName})
        return doc_son

    @staticmethod
    def load(proj, exported_project_path, is_hard_import, show_progressbar=True):
        """
            loads an exported project's documents to a project, either hard import with the original ids or not 

        Parameters
        ----------
        proj: str or Project
                either name of the project to import or an existing Project instance,
                creates a new project if it doesn't exist

        is_hard_import: bool
                should the original ids be used when importing or generate new ones per document(using original ids might fail due to duplicate ids)

        show_progressbar: bool 
        """
        if isinstance(proj, str):
            proj = Project(proj)

        with zipfile.ZipFile(exported_project_path, 'r') as zf:
            if is_hard_import:
                pickled_docs_iterator = Project._iter_pickled_docs(zf, return_batched=True)
                if show_progressbar:
                    pickled_docs_iterator = tqdm(pickled_docs_iterator, desc="Loading documents", unit="docsBatch", unit_scale=True)
                for pickled_docs_batch in pickled_docs_iterator:
                    proj._all._metadataCol._get_collection().insert_many([proj.updateProjectNameOnDoc(doc_son) for doc_son in pickled_docs_batch])
            else:
                pickled_docs_iterator = Project._iter_pickled_docs(zf, return_batched=False)
                if show_progressbar:
                    pickled_docs_iterator = tqdm(pickled_docs_iterator, desc="Loading documents", unit="docs", unit_scale=True)
                for pickled_doc in pickled_docs_iterator:
                    proj._all._metadataCol.objects.insert(proj._all._metadataCol(**(proj.updateProjectNameOnDoc(pickled_doc))), load_bulk=False)

    @property
    def simulations(self) -> Simulations_Collection:
        """
            Access the simulation type documents.

    Returns
    -------
            hera.datalayer.collection.Simulation_Collection

        """
        return self._simulations

    def _getConfigDocument(self, evaluate=False):
        """
        Returns the document of the config.
        If there is no config document, return empty dictionary.

        Parameters
        -------
        evaluate :  bool
            if False returns a QuerySet otherwise evaluates to a document

        Returns
        -------

         dict
                The configuration of the toolkit.
        """
        config_type = f"{self.projectName}__config__"
        documents = self.getCacheDocuments(type=config_type)
        # keeping lazy so we just return documents which is a QuerySet
        if documents.first() is None:
            self.addCacheDocument(type=config_type,
                                  resource="",
                                  dataFormat=datatypes.STRING,
                                  desc={})

        return documents[0] if evaluate else documents

    def setCounter(self,counterName:str,defaultValue=0):
        """
            Defines a counter in the config of the project.
            The counter is specific to this project.
        Parameters
        ----------
        counterName :  str
            The counter name.

        Returns
        -------
        None
        """
        counterName = self._normalizeCounterName(counterName)
        cnfg_doc = self._getConfigDocument()
        self.defineCounter(counterName,0)
        cnfg_doc.update_one(**{f"set__desc__counter__{counterName}":defaultValue})

    def _normalizeCounterName(self, counterName):
        """Normalize a counter name by replacing dots with underscores."""
        counterName=counterName.replace('.','_')
        # avoiding conflicts with mongodb
        if '__' in counterName:
            raise RuntimeError("a counter's name cannot contain either of __ / ._ / _. / ..")
        return counterName

    def defineCounter(self,counterName,defaultValue=0) -> None:
        """
            Defines a counter in the config of the project, if it does not exist
            The counter is specific to this project.
        Parameters
        ----------
        counterName :  str
            The counter name.

        Returns
        -------

        """
        counterName = self._normalizeCounterName(counterName)
        cnfg_doc = self._getConfigDocument()
        self._enforce_counter_field(cnfg_doc)
        if cnfg_doc.filter(**{f"desc__counters__{counterName}__exists": True}).first() is None: 
            cnfg_doc.update(**{f"set__desc__counters__{counterName}": defaultValue})
            return True
        return False


    def getCounter(self,counterName):
        """
            Return the value of the counter if doesn't exist returning None
        Parameters
        ----------
        counterName :  str
            The name of the counter.

        Returns
        -------

        """
        counterName = self._normalizeCounterName(counterName)
        cnfg_doc = self._getConfigDocument()
        self._enforce_counter_field(cnfg_doc)
        #if it doesn't exist there is nothing to get, returning None
        if cnfg_doc.filter(**{f"desc__counters__{counterName}__exists": True}).first() is None: 
            return None
        doc = cnfg_doc[0]
        return doc['desc']['counters'][counterName]

    def _enforce_counter_field(self, cnfg_doc):
        """Ensure the ``counters`` field exists in the config document."""
        if cnfg_doc.filter(desc__counters__exists=True).first() is None:
            cnfg_doc.update(**{f"set__desc__counters": {}})


    def getCounterAndAdd(self, counterName, addition=1):
        """
            Return the value of the counter and add [addition].
            If the counter is not defined it is initialized to 0.
        Parameters
        ----------
        counterName :  str
            The name of the counter.

        addition : int
            The amount to add to the counter. The default is 1

        Returns
        -------
        int
            The updated counter value, or 0 if the counter was newly created.
        """
        counterName = self._normalizeCounterName(counterName)
        isNew = self.defineCounter(counterName,0)
        if isNew:
            return 0

        cnfg_doc = self._getConfigDocument()
        doc = cnfg_doc.modify(
            upsert=True,
            new=True,
            **{
                f"inc__desc__counters__{counterName}":addition
            })
        return doc['desc']['counters'][counterName]

    @deprecated(reason="Use getCounterAndAdd instead")
    def addCounter(self, counterName, addition=1):
        """Deprecated. Use ``getCounterAndAdd`` instead."""
        return self.getCounterAndAdd(counterName,addition)

    def getConfig(self):
        """
        Returns the config document's description.
        If there is no config document, return empty dictionary.

        Returns
        -------
        dict
                The configuration of the toolkit.
        """
        if self._projectName == self.DEFAULTPROJECT:
            raise ValueError("Default project cannot use configuration")
        doc = self._getConfigDocument(evaluate=True)
        return dict(doc["desc"])


    def initConfig(self,**kwargs):
        """
            Sets the value of the config, if the keys does not exist. If they exist, leave the old value.
        Parameters
        ----------
        kwargs

        Returns
        -------

        """
        if self._projectName == self.DEFAULTPROJECT:
            raise ValueError("Default project cannot use configuration")

        doc = self._getConfigDocument()
        for key in kwargs.keys():
            doc.filter(**{f"desc__{key}__exists":False}).update(**{f"set__desc__{key}":kwargs[key]})



    def getMetadata(self):
        """
        Return the description of all ot the documents in the current project.
        It assumes that description is a key-value format (does not support more complex structures).

    Returns
    -------
             pandas.DataFrame
        """
        descList = [doc.desc for doc in AbstractCollection().getDocuments(projectName=self._projectName)]
        return pandas.DataFrame(descList)

    def getDocumentByID(self, id):
        """
        Returns a document by its ID from any collection.

        Parameters
        ----------
        id : str
            The document ID.

        Returns
        -------
        document
            The document with the given ID.
        """
        return self._all.getDocumentByID(id)

    def getMeasurementsDocumentsAsDict(self, with_id=False, **kwargs):
        """
            Querying the DB for measurements.old documents and return the results as a list of dict

        Parameters
        ----------


        with_id : bool, optional, default False
            rather or not should the 'id' key be in the documents.

        kwargs: parameters
            Filters for the query

        Returns
        -------
            List of dicts
        """
        return self.measurements.getDocumentsAsDict(projectName=self._projectName, with_id=with_id, **kwargs)

    def getMeasurementsDocuments(self, resource=None, dataFormat=None, type=None, **desc):
        """
            Query measurements.old documents.

        Parameters
        ----------
        resource: str
            query by resource, optional.

        dataFormat: str
            query by data format, optional.

        type: str
            query by type

        desc: dict
            query by the measurement document

        Returns
        -------
            List of documents.
        """
        return self.measurements.getDocuments(projectName=self._projectName, resource=resource, dataFormat=dataFormat, type=type, **desc)

    def getAllDocuments(self, resource=None, dataFormat=None, type=None, **desc):
        """
            Runs getXDocuments for measurements, cache and simulations aggregating all to one list

        Parameters
        ----------
        resource: str
            query by resource, optional.

        dataFormat: str
            query by data format, optional.

        type: str
            query by type

        desc: dict
            query by the measurement document

        Returns
        -------
            List of documents.
        """
        docs = []
        docs.extend(self.getSimulationsDocuments(resource=resource, dataFormat=dataFormat, type=type, desc=desc))
        docs.extend(self.getMeasurementsDocuments(resource=resource, dataFormat=dataFormat, type=type, desc=desc))
        docs.extend(self.getCacheDocuments(resource=resource, dataFormat=dataFormat, type=type, desc=desc))
        return docs

    def addDocumentFromDict(self,documentDict):
        """
            Load the document to the project.
            The structure of the dict is:
            {
                _cls : "Metadata.<Cache|Measurements|Simulations>
                desc : {...},
                type : "...",
                resource : ""
                dataFormat : ""
            }

        Parameters
        ----------
        documentDict :  dict
            The dictionary with the data of the document

        Returns
        -------

        """
        addingDict = dict(documentDict)
        if 'projectName' in addingDict:
            del addingDict['projectName']
        docType = addingDict['_cls'].split(".")[1]
        del addingDict['_cls']

        addingFunc = getattr(self,f"add{docType}Document")
        if addingFunc is None:
            raise ValueError(f"The document type {docType} is not found")

        addingFunc(**addingDict)

    def addMeasurementsDocument(self, resource="", dataFormat="string", type="", desc={}):
        """
            Adds a new measurement document.

        Parameters
        ----------
        resource: str
            query by resource, optional.

        dataFormat: str
            query by data format, optional.

        type: str
            query by type

        desc: dict
            query by the measurement document

        Returns
        -------
            The new document
        """
        logger = get_classMethod_logger(self, "init")
        if self.projectName == self.DEFAULTPROJECT and not self._allowWritingToDefaultProject:
            err = f"project {self.projectName} is read-only. "
            logger.error(err)
            raise RuntimeError(err)

        return self.measurements.addDocument(projectName=self._projectName, resource=resource, dataFormat=dataFormat, type=type, desc=desc)

    def deleteMeasurementsDocuments(self, **kwargs):
        """
            Delete the measurements.old documents that fit the query.

        Parameters
        ----------
        kwargs: query dicts.

        Returns
        -------
            The list of documents that was deleted.
        """
        return self.measurements.deleteDocuments(projectName=self._projectName, **kwargs)

    def getSimulationsDocumentsAsDict(self, with_id=False, **kwargs):
        """
            Querying the DB for simulation documents and return the results as a list of dict

        Parameters
        ----------
        with_id : bool, optional, default False
            rather or not should the 'id' key be in the documents.

        kwargs: parameters
            Filters for the query

        Returns
        -------
            List of dicts
        """

        return self.simulations.getDocumentsAsDict(projectName=self._projectName, with_id=with_id, **kwargs)

    def getSimulationsDocuments(self, resource=None, dataFormat=None, type=None, **desc):
        """
            Query simulation documents.

        Parameters
        ----------
        resource: str
            query by resource, optional.

        dataFormat: str
            query by data format, optional.

        type: str
            query by type

        desc: dict
            query by the measurement document

        Returns
        -------
            List of documents.
        """
        return self.simulations.getDocuments(projectName=self._projectName, resource=resource, dataFormat=dataFormat, type=type,
                                **desc)

    def addSimulationsDocument(self, resource="", dataFormat="string", type="", desc={}):
        """
            Adds a new simulations.old document.

        Parameters
        ----------
        resource: str
            query by resource, optional.

        dataFormat: str
            query by data format, optional.

        type: str
            query by type

        desc: dict
            query by the measurement document

        Returns
        -------
            The new document
        """
        logger = get_classMethod_logger(self, "init")
        if self.projectName == self.DEFAULTPROJECT and not self._allowWritingToDefaultProject:
            err = f"project {self.projectName} is read-only. "
            logger.error(err)
            raise RuntimeError(err)

        return self.simulations.addDocument(projectName=self._projectName, resource=resource, dataFormat=dataFormat, type=type,
                               desc=desc)

    def deleteSimulationsDocuments(self, **kwargs):
        """
            Delete the simulations.old documents that fit the query.

        Parameters
        ----------
        kwargs: query dicts.

        Returns
        -------
            The list of documents that was deleted.
        """

        return self.simulations.deleteDocuments(projectName=self._projectName, **kwargs)

    def getCacheDocumentsAsDict(self,  with_id=False, **kwargs):
        """
            Querying the DB for cache documents and return the results as a list of dict

        Parameters
        ----------


        with_id : bool, optional, default False
            rather or not should the 'id' key be in the documents.

        kwargs: parameters
            Filters for the query

        Returns
        -------
            List of dicts
        """

        return self.cache.getDocumentsAsDict(projectName=self._projectName, with_id=with_id, **kwargs)

    def getCacheDocuments(self, resource=None, dataFormat=None, type=None, **desc):
        """
            Querying the DB for cache documents and return the results as a list of dict

        Parameters
        ----------
        with_id : bool, optional, default False
            rather or not should the 'id' key be in the documents.

        kwargs: parameters
            Filters for the query

        Returns
        -------
            List of dicts
        """

        return self.cache.getDocuments(projectName=self._projectName, resource=resource, dataFormat=dataFormat, type=type,
                                       **desc)

    def addCacheDocument(self, resource="", dataFormat="string", type="", desc={}):
        """
            Adds a new cache document.

        Parameters
        ----------
        resource: str
            query by resource, optional.

        dataFormat: str
            query by data format, optional.

        type: str
            query by type

        desc: dict
            query by the measurement document

        Returns
        -------
            The new document
        """
        logger = get_classMethod_logger(self, "init")
        if self.projectName == self.DEFAULTPROJECT and not self._allowWritingToDefaultProject:
            err = f"project {self.projectName} is read-only. "
            logger.error(err)
            raise RuntimeError(err)

        return self.cache.addDocument(projectName=self._projectName, resource=resource, dataFormat=dataFormat, type=type,desc=desc)

    def deleteCacheDocuments(self, **kwargs):
        """
            Delete the cache documents that fit the query.

        Parameters
        ----------
        kwargs: query dicts.

        Returns
        -------
            The list of documents that was deleted.
        """

        return self.cache.deleteDocuments(projectName=self._projectName, **kwargs)

    def saveData(self,name,data,desc,kind,type=None,dataFormat=None,**kwargs):
        """
            Adds a cache document with the data.
            Estimates the dataFormat from the data type.

            The type is
        Parameters
        ----------
        data : a data that can be dataframe, xarray, numpy ....

        desc : dict
            A dict with the metadata to save
        type : str
            If None, then set the type to be the name of the function that called this method.

        Returns
        -------
            The new document
        """
        guessedDataFormat = self.datatypes.getDataFormatName(data) if dataFormat is None else dataFormat

        handler = self.datatypes.getHandler(guessedDataFormat)
        file_extension = self.datatypes.getDataFormatExtension(data)

        cacheDirectory = os.path.join(self.filesDirectory, "cache")
        os.makedirs(cacheDirectory, exist_ok=True)
        fileID = self.getCounterAndAdd(name)
        fileName = os.path.join(cacheDirectory, f"{name}_{fileID}.{file_extension}")

        saveParamsUpdatedDict = handler.saveData(data, fileName,**kwargs)

        funcName = getattr(self,f"add{kind}Document")
        fullType = type if type is not None else name

        qry = ConfigurationToJSON(desc, standardize=True, splitUnits=True, keepOriginalUnits=True)
        storeParamsDict = qry.get("storeParameters",{})
        storeParamsDict.update(saveParamsUpdatedDict)
        qry["storeParameters"] = storeParamsDict

        doc = funcName(type=fullType, dataFormat=guessedDataFormat, resource=fileName, desc=qry)
        return doc

    def saveMeasurementData(self, name, data, desc, type=None, dataFormat=None, **kwargs):
        """
        Save data as a measurement document. See ``saveData`` for parameter details.
        """
        self.saveData(name=name, data=data, desc=desc, kind="Measurement", type=type, dataFormat=dataFormat, **kwargs)

    def saveCacheData(self, name, data, desc, type=None, dataFormat=None, **kwargs):
        """
        Save data as a cache document. See ``saveData`` for parameter details.
        """
        self.saveData(name=name, data=data, desc=desc, kind="Cache", type=type, dataFormat=dataFormat, **kwargs)

    def saveSimulationData(self, name, data, desc, type=None, dataFormat=None, **kwargs):
        """
        Save data as a simulation document. See ``saveData`` for parameter details.
        """
        self.saveData(name=name, data=data, desc=desc, kind="Simulation", type=type, dataFormat=dataFormat, **kwargs)

    def _get_full_func_name(self,func):
        """Returns the full qualified path: module.[class.]function_name"""
        if not callable(func):
            raise TypeError("Provided object is not callable.")

        # Handle bound methods by unwrapping them
        if inspect.ismethod(func):
            # Get the original function and its class
            cls = func.__self__.__class__
            method_name = func.__name__
            class_qualname = cls.__qualname__
            module = func.__module__
            if module == "__main__":
                ret = f"{class_qualname}.{method_name}"
            else:
                ret = f"{module}.{class_qualname}.{method_name}"

        elif inspect.isfunction(func):
            ret = func.__qualname__
        else:

            # Handle unbound class or static methods and plain functions
            qualname = func.__qualname__
            module = func.__module__
            ret = f"{module}.{qualname}"
        return ret



    @staticmethod
    def getProjectList(cls,user=None):
        """
            Return the list with the names of the existing projects .

        :param user: str
            The name of the database.

        Returns
        -------
            list.
        """
        return list(set(AbstractCollection(connectionName=user).getProjectList()))

addMeasurementsDocument(resource='', dataFormat='string', type='', desc={})

Adds a new measurement document.

Parameters:

Name	Type	Description	Default
resource		query by resource, optional.	''
dataFormat		query by data format, optional.	'string'
type		query by type	''
desc		query by the measurement document	{}

Returns:

Type	Description
The new document

Source code in hera/datalayer/project.py

def addMeasurementsDocument(self, resource="", dataFormat="string", type="", desc={}):
    """
        Adds a new measurement document.

    Parameters
    ----------
    resource: str
        query by resource, optional.

    dataFormat: str
        query by data format, optional.

    type: str
        query by type

    desc: dict
        query by the measurement document

    Returns
    -------
        The new document
    """
    logger = get_classMethod_logger(self, "init")
    if self.projectName == self.DEFAULTPROJECT and not self._allowWritingToDefaultProject:
        err = f"project {self.projectName} is read-only. "
        logger.error(err)
        raise RuntimeError(err)

    return self.measurements.addDocument(projectName=self._projectName, resource=resource, dataFormat=dataFormat, type=type, desc=desc)

getMeasurementsDocuments(resource=None, dataFormat=None, type=None, **desc)

Query measurements.old documents.

Parameters:

Name	Type	Description	Default
resource		query by resource, optional.	None
dataFormat		query by data format, optional.	None
type		query by type	None
desc		query by the measurement document	{}

Returns:

Type	Description
List of documents.

Source code in hera/datalayer/project.py

def getMeasurementsDocuments(self, resource=None, dataFormat=None, type=None, **desc):
    """
        Query measurements.old documents.

    Parameters
    ----------
    resource: str
        query by resource, optional.

    dataFormat: str
        query by data format, optional.

    type: str
        query by type

    desc: dict
        query by the measurement document

    Returns
    -------
        List of documents.
    """
    return self.measurements.getDocuments(projectName=self._projectName, resource=resource, dataFormat=dataFormat, type=type, **desc)

deleteMeasurementsDocuments(**kwargs)

Delete the measurements.old documents that fit the query.

Parameters:

Name	Type	Description	Default
kwargs			{}

Returns:

Type	Description
The list of documents that was deleted.

Source code in hera/datalayer/project.py

def deleteMeasurementsDocuments(self, **kwargs):
    """
        Delete the measurements.old documents that fit the query.

    Parameters
    ----------
    kwargs: query dicts.

    Returns
    -------
        The list of documents that was deleted.
    """
    return self.measurements.deleteDocuments(projectName=self._projectName, **kwargs)

setConfig(keep_old_values=True, **kwargs)

Create a config document or updates an existing config document.

Source code in hera/datalayer/project.py

def setConfig(self,keep_old_values=True, **kwargs):
    """
        Create a config document or updates an existing config document.
    """
    if self._projectName == self.DEFAULTPROJECT:
        err = """
                Default project was initiated and it is cannot use configuration.

                This error os obtained because you have initiated the project or the toolkit 
                using projectName=None, and you don't case a caseConfiguration.json in the directory that 
                specifies the project name. 

                To solve this project either 
                - Create a new project in the directory: 
                  run  
                    >> hera-project project create <the directory name>
                  in the parent directory 
                - Create manually the file caseConfiguration.json in the directory with the key 'projectName' that holds 
                  the name of the project as a string.     
        """
        raise ValueError(err)

    doc = self._getConfigDocument()
    if keep_old_values:
        update_kwargs = {
            f"set__desc__{k}": v
            for k, v in kwargs.items()
        }
        doc.update(**update_kwargs)
    else:
        doc.update(set__desc=kwargs)

getConfig()

Returns the config document's description. If there is no config document, return empty dictionary.

Returns:

Type	Description
dict	The configuration of the toolkit.

Source code in hera/datalayer/project.py

def getConfig(self):
    """
    Returns the config document's description.
    If there is no config document, return empty dictionary.

    Returns
    -------
    dict
            The configuration of the toolkit.
    """
    if self._projectName == self.DEFAULTPROJECT:
        raise ValueError("Default project cannot use configuration")
    doc = self._getConfigDocument(evaluate=True)
    return dict(doc["desc"])

setCounter(counterName: str, defaultValue=0)

Defines a counter in the config of the project.
The counter is specific to this project.

Parameters:

Name	Type	Description	Default
counterName	str	The counter name.	required

Returns:

Type	Description
None

Source code in hera/datalayer/project.py

def setCounter(self,counterName:str,defaultValue=0):
    """
        Defines a counter in the config of the project.
        The counter is specific to this project.
    Parameters
    ----------
    counterName :  str
        The counter name.

    Returns
    -------
    None
    """
    counterName = self._normalizeCounterName(counterName)
    cnfg_doc = self._getConfigDocument()
    self.defineCounter(counterName,0)
    cnfg_doc.update_one(**{f"set__desc__counter__{counterName}":defaultValue})

getCounterAndAdd(counterName, addition=1)

Return the value of the counter and add [addition].
If the counter is not defined it is initialized to 0.

Parameters:

Name	Type	Description	Default
counterName	str	The name of the counter.	required
addition	int	The amount to add to the counter. The default is 1	1

Returns:

Type	Description
int	The updated counter value, or 0 if the counter was newly created.

Source code in hera/datalayer/project.py

def getCounterAndAdd(self, counterName, addition=1):
    """
        Return the value of the counter and add [addition].
        If the counter is not defined it is initialized to 0.
    Parameters
    ----------
    counterName :  str
        The name of the counter.

    addition : int
        The amount to add to the counter. The default is 1

    Returns
    -------
    int
        The updated counter value, or 0 if the counter was newly created.
    """
    counterName = self._normalizeCounterName(counterName)
    isNew = self.defineCounter(counterName,0)
    if isNew:
        return 0

    cnfg_doc = self._getConfigDocument()
    doc = cnfg_doc.modify(
        upsert=True,
        new=True,
        **{
            f"inc__desc__counters__{counterName}":addition
        })
    return doc['desc']['counters'][counterName]