Core API

The central classes that form the backbone of Hera: the toolkit registry, base toolkit class, and project data interface.

ToolkitHome

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

__init__(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:

Name	Type	Description	Default
projectName	str	The project name. If None, loaded from caseConfiguration.	None
filesDirectory	str	Directory for file outputs.	None

Source code in hera/toolkit.py

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

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}'."
    )

auto_register_and_get(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.

Source code in hera/toolkit.py

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)

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

getExperimentToolkitDocuments(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.

Source code in hera/toolkit.py

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

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

setDefaultRepository(*, 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'.

Source code in hera/toolkit.py

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,
    )

getDefaultRepository(*, projectName: str) -> str

Read the saved default repository name (if exists). Returns '' if missing.

Source code in hera/toolkit.py

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 ""

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
    )

import_toolkits_from_json(*, 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.)

Source code in hera/toolkit.py

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

import_experiments_from_json(*, projectName: str, json_path: str) -> list

Load experiments from JSON into the given project as measurements documents.

Parameters:

Name	Type	Description	Default
projectName	str	The project to load experiments into.	required
json_path	str	Path to the JSON file containing experiment definitions.	required

Returns:

Type	Description
list of str	Names of the loaded experiments.

Source code in hera/toolkit.py

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

abstractToolkit

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.")

presentation property

Access the presentation layer.

Returns:

Type	Description
object or None

analysis property

Access the analysis (data) layer.

Returns:

Type	Description
object or None

toolkitName property

The name of the toolkit.

Returns:

Type	Description
str

projectName property

The name of the project.

Returns:

Type	Description
str

classLoggerName property

The logger name for the current class and method context.

Returns:

Type	Description
str

__init__(toolkitName: str, projectName: Optional[str] = None, connectionName: Optional[str] = None, filesDirectory: Optional[str] = None)

Initialize a new toolkit.

Parameters:

Name	Type	Description	Default
toolkitName	str	The name of the toolkit.	required
projectName	str	The project that the toolkit works in. If None, Project's automatic project-name mechanism is used.	None
connectionName	str	The name of the DB connection. If None, uses the current OS username.	None
filesDirectory	str	Directory to save datasource files.	None

Source code in hera/toolkit.py

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

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

Add a cache document, automatically tagging it with the toolkit name.

See Project.addCacheDocument for parameter details.

Source code in hera/toolkit.py

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)

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

Add a measurements document, automatically tagging it with the toolkit name.

See Project.addMeasurementsDocument for parameter details.

Source code in hera/toolkit.py

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)

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

Add a simulations document, automatically tagging it with the toolkit name.

See Project.addSimulationsDocument for parameter details.

Source code in hera/toolkit.py

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)

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]

getDataSourceMap(**filters) -> List[Dict[str, Any]]

Return a list of all data sources and their versions for this toolkit.

Parameters:

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

Returns:

Type	Description
list of dict	Each dict contains dataFormat, resource, and all desc fields.

Source code in hera/toolkit.py

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

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)

getDataSourceDocumentsList(**kwargs)

Return all data source documents associated with this toolkit.

Parameters:

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

Returns:

Type	Description
list	List of mongoengine documents.

Source code in hera/toolkit.py

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)

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

getToolkitDocument(toolkit_name: str)

Find a dynamic toolkit document by name.

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

Parameters:

Name	Type	Description	Default
toolkit_name	str	The toolkit name to search for.	required

Returns:

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

Source code in hera/toolkit.py

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

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

Return a list containing the data source document (for API symmetry).

Parameters:

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

Returns:

Type	Description
list	A list containing the document, or an empty list if not found.

Source code in hera/toolkit.py

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]

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()

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

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

setDataSourceDefaultVersion(datasourceName: str, version: tuple)

Set the default version for a given data source.

Parameters:

Name	Type	Description	Default
datasourceName	str	The name of the data source.	required
version	tuple	The version to set as default.	required

Raises:

Type	Description
ValueError	If no data source with the given name and version exists.

Source code in hera/toolkit.py

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.")

Project

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()))

FilesDirectory property

The directory to save files (when creating files).

:return:

filesDirectory property

The directory to save files (when creating files).

:return:

measurements: Measurements_Collection property

    Access the measurement type documents.

Returns:

Type	Description
hera.datalayer.collection.Measurements_Collection

cache: Cache_Collection property

    Access the Cache type documents.

Returns:

Type	Description
hera.datalayer.collection.Cache_Collection

all: AbstractCollection property

    Access to all document types.

Returns:

Type	Description
hera.datalayer.collection.AbstractCollection

projectName property

The name of the current project.

Returns:

Type	Description
str

simulations: Simulations_Collection property

    Access the simulation type documents.

Returns:

Type	Description
hera.datalayer.collection.Simulation_Collection

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)

__init__(projectName=None, connectionName=None, configurationPath=None, filesDirectory=None)

Initialize the project class.

Parameters:

Name	Type	Description	Default
projectName		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] } ```	None
connectionName		The name of the DB connection. If not specified, use the connection with the linux user name.	None

Source code in hera/datalayer/project.py

    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

export(path, export_chunk_size=1024, show_progressbar=True)

exports the project in chunks contained to one zip file

Parameters:

Name	Type	Description	Default
path		the path to the zip file to be created, overrides file if already exists	required
export_chunk_size		the number of documents in each chunk	1024
show_progressbar			True

Source code in hera/datalayer/project.py

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

updateProjectNameOnDoc(doc_son)

updates the projectName field of a document to be assigned to this project

Parameters:

Name	Type	Description	Default
doc_son		the document to be updated	required

Source code in hera/datalayer/project.py

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

load(proj, exported_project_path, is_hard_import, show_progressbar=True) staticmethod

loads an exported project's documents to a project, either hard import with the original ids or not

Parameters:

Name	Type	Description	Default
proj		either name of the project to import or an existing Project instance, creates a new project if it doesn't exist	required
is_hard_import		should the original ids be used when importing or generate new ones per document(using original ids might fail due to duplicate ids)	required
show_progressbar			True

Source code in hera/datalayer/project.py

@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)

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})

defineCounter(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:

Name	Type	Description	Default
counterName	str	The counter name.	required

Source code in hera/datalayer/project.py

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

getCounter(counterName)

Return the value of the counter if doesn't exist returning None

Parameters:

Name	Type	Description	Default
counterName	str	The name of the counter.	required

Source code in hera/datalayer/project.py

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]

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]

addCounter(counterName, addition=1)

Deprecated. Use getCounterAndAdd instead.

Source code in hera/datalayer/project.py

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

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"])

initConfig(**kwargs)

Sets the value of the config, if the keys does not exist. If they exist, leave the old value.

Parameters:

Name	Type	Description	Default
kwargs			{}

Source code in hera/datalayer/project.py

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]})

getMetadata()

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:

Type	Description
pandas.DataFrame

Source code in hera/datalayer/project.py

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)

getDocumentByID(id)

Returns a document by its ID from any collection.

Parameters:

Name	Type	Description	Default
id	str	The document ID.	required

Returns:

Type	Description
document	The document with the given ID.

Source code in hera/datalayer/project.py

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)

getMeasurementsDocumentsAsDict(with_id=False, **kwargs)

Querying the DB for measurements.old documents and return the results as a list of dict

Parameters:

Name	Type	Description	Default
with_id	bool	rather or not should the 'id' key be in the documents.	False
kwargs		Filters for the query	{}

Returns:

Type	Description
List of dicts

Source code in hera/datalayer/project.py

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)

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)

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

Runs getXDocuments for measurements, cache and simulations aggregating all to one list

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 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

addDocumentFromDict(documentDict)

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

Parameters:

Name	Type	Description	Default
documentDict	dict	The dictionary with the data of the document	required

Source code in hera/datalayer/project.py

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)

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)

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)

getSimulationsDocumentsAsDict(with_id=False, **kwargs)

Querying the DB for simulation documents and return the results as a list of dict

Parameters:

Name	Type	Description	Default
with_id	bool	rather or not should the 'id' key be in the documents.	False
kwargs		Filters for the query	{}

Returns:

Type	Description
List of dicts

Source code in hera/datalayer/project.py

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)

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

Query simulation 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 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)

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

Adds a new simulations.old 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 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)

deleteSimulationsDocuments(**kwargs)

Delete the simulations.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 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)

getCacheDocumentsAsDict(with_id=False, **kwargs)

Querying the DB for cache documents and return the results as a list of dict

Parameters:

Name	Type	Description	Default
with_id	bool	rather or not should the 'id' key be in the documents.	False
kwargs		Filters for the query	{}

Returns:

Type	Description
List of dicts

Source code in hera/datalayer/project.py

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)

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

Querying the DB for cache documents and return the results as a list of dict

Parameters:

Name	Type	Description	Default
with_id	bool	rather or not should the 'id' key be in the documents.	False
kwargs		Filters for the query	required

Returns:

Type	Description
List of dicts

Source code in hera/datalayer/project.py

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)

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

Adds a new cache 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 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)

deleteCacheDocuments(**kwargs)

Delete the cache 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 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)

saveData(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:

Name	Type	Description	Default
data	a data that can be dataframe, xarray, numpy ....		required
desc	dict	A dict with the metadata to save	required
type	str	If None, then set the type to be the name of the function that called this method.	None

Returns:

Type	Description
The new document

Source code in hera/datalayer/project.py

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

saveMeasurementData(name, data, desc, type=None, dataFormat=None, **kwargs)

Save data as a measurement document. See saveData for parameter details.

Source code in hera/datalayer/project.py

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)

saveCacheData(name, data, desc, type=None, dataFormat=None, **kwargs)

Save data as a cache document. See saveData for parameter details.

Source code in hera/datalayer/project.py

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)

saveSimulationData(name, data, desc, type=None, dataFormat=None, **kwargs)

Save data as a simulation document. See saveData for parameter details.

Source code in hera/datalayer/project.py

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)

getProjectList(user=None) staticmethod

Return the list with the names of the existing projects .

:param user: str The name of the database.

Returns:

Type	Description
list.

Source code in hera/datalayer/project.py

@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()))