added metadata retrieve from beamtime folder

Author: kutnyakhov

src/sed/core/config_model.py

@@ -26,6 +26,7 @@ class PathsModel(BaseModel):
 
     raw: DirectoryPath
     processed: Optional[Union[DirectoryPath, NewPath]] = None
+    meta: Optional[Union[DirectoryPath, NewPath]] = None
 
 
 class CopyToolModel(BaseModel):

src/sed/loader/cfel/loader.py

@@ -56,8 +56,10 @@ def __init__(self, config: dict, verbose: bool = True) -> None:
         set_verbosity(logger, self._verbose)
 
         self.instrument: str = self._config["core"].get("instrument", "hextof")  # default is hextof
+        self.beamtime_dir: str = None
         self.raw_dir: str = None
         self.processed_dir: str = None
+        self.meta_dir: str = None
 
     @property
     def verbose(self) -> bool:
@@ -94,9 +96,14 @@ def _initialize_dirs(self) -> None:
         # Only raw_dir is necessary, processed_dir can be based on raw_dir, if not provided
         if "paths" in self._config["core"]:
             raw_dir = Path(self._config["core"]["paths"].get("raw", ""))
+            print(raw_dir)
             processed_dir = Path(
                 self._config["core"]["paths"].get("processed", raw_dir.joinpath("processed")),
             )
+            meta_dir = Path(
+                self._config["core"]["paths"].get("meta", raw_dir.joinpath("meta")),
+            )
+            beamtime_dir = Path(raw_dir).parent
 
         else:
             try:
@@ -130,11 +137,14 @@ def _initialize_dirs(self) -> None:
             raw_dir = raw_paths[0].resolve()
 
             processed_dir = beamtime_dir.joinpath("processed")
+            meta_dir = beamtime_dir.joinpath("meta/fabtrack/")
 
         processed_dir.mkdir(parents=True, exist_ok=True)
 
+        self.beamtime_dir = str(beamtime_dir)
         self.raw_dir = str(raw_dir)
         self.processed_dir = str(processed_dir)
+        self.meta_dir = str(meta_dir)
 
     @property
     def available_runs(self) -> list[int]:
@@ -209,7 +219,7 @@ def get_files_from_run_id(  # type: ignore[override]
         # Return the list of found files
         return [str(file.resolve()) for file in files]
 
-    def parse_metadata(self, token: str = None) -> dict:
+    def parse_scicat_metadata(self, token: str = None) -> dict:
         """Uses the MetadataRetriever class to fetch metadata from scicat for each run.
 
         Returns:
@@ -225,6 +235,23 @@ def parse_metadata(self, token: str = None) -> dict:
 
         return metadata
 
+    def parse_local_metadata(self) -> dict:
+        """Uses the MetadataRetriever class to fetch metadata from local folder for each run.
+
+        Returns:
+            dict: Metadata dictionary
+        """
+        metadata_retriever = MetadataRetriever(self._config["metadata"])
+        metadata = metadata_retriever.get_local_metadata(
+            beamtime_id=self._config["core"]["beamtime_id"],
+            beamtime_dir=self.beamtime_dir,
+            meta_dir=self.meta_dir,
+            runs=self.runs,
+            metadata=self.metadata,
+        )
+
+        return metadata
+
     def get_count_rate(
         self,
         fids: Sequence[int] = None,  # noqa: ARG002
@@ -403,7 +430,12 @@ def read_dataframe(
             filter_timed_by_electron=filter_timed_by_electron,
         )
 
-        self.metadata.update(self.parse_metadata(token) if collect_metadata else {})
+        if len(self.parse_scicat_metadata(token)) == 0:
+            print("No SciCat metadata available, checking local folder")
+            self.metadata.update(self.parse_local_metadata())
+        else:
+            print("Metadata taken from SciCat")
+            self.metadata.update(self.parse_scicat_metadata(token) if collect_metadata else {})
         self.metadata.update(bh.metadata)
 
         print(f"loading complete in {time.time() - t0: .2f} s")

src/sed/loader/flash/metadata.py

@@ -5,6 +5,8 @@
 from __future__ import annotations
 
 import requests
+import json
+import yaml
 
 from sed.core.config import read_env_var
 from sed.core.config import save_env_var
@@ -144,3 +146,93 @@ def _create_new_dataset_url(self, pid: str) -> str:
     def _reformat_pid(self, pid: str) -> str:
         """SciCat adds a pid-prefix + "/"  but at DESY prefix = "" """
         return (pid).replace("/", "%2F")
+
+    def get_local_metadata(
+            self,
+            beamtime_id: str,
+            beamtime_dir: str,
+            meta_dir: str,
+            runs: list,
+            metadata: dict = None,
+    ) -> dict:
+        """
+        Retrieves metadata for a given beamtime ID and list of runs from local meta folder and yaml file.
+
+        Args:
+            beamtime_id (str): The ID of the beamtime.
+            runs (list): A list of run IDs.
+            metadata (dict, optional): The existing metadata dictionary.
+            Defaults to None.
+
+        Returns:
+            Dict: The updated metadata dictionary.
+
+        Raises:
+            Exception: If the request to retrieve metadata fails.
+        """
+        if metadata is None:
+            metadata = {}
+
+        beamtime_metadata = self._get_beamtime_metadata(beamtime_dir,beamtime_id)
+        metadata.update(beamtime_metadata)
+        for run in runs:
+            logger.debug(f"Retrieving metadata for PID: {run}")
+            local_metadata_per_run = self._get_local_metadata_per_run(meta_dir,run)
+            local_metadata_per_run.update(local_metadata_per_run)  # TODO: Not correct for multiple runs
+
+        metadata.update({'scientificMetadata': local_metadata_per_run['_data']})
+
+        logger.debug(f"Retrieved metadata with {len(metadata)} entries")
+        return metadata
+
+    def _get_beamtime_metadata(
+            self,
+            beamtime_dir: str,
+            beamtime_id: str,
+    ) -> dict:
+        """
+        Retrieves general metadata for a given beamtime ID from beamtime-metadata-{beamtime_id}.json file
+
+        Args:
+            beamtime_id (str): The ID of the beamtime.
+            meta_dir(str): The existing local metadata folder.
+
+        Returns:
+            Dict: The retrieved metadata dictionary.
+
+        Raises:
+            Exception: If the request to retrieve metadata fails.
+        """     
+        try:
+            f = open(f'{beamtime_dir}/beamtime-metadata-{beamtime_id}.json', "r")
+            beamtime_metadata = json.loads(f.read())
+            return beamtime_metadata
+
+        except Exception as exception:
+            logger.warning(f"Failed to retrieve metadata for beamtime ID {beamtime_id}: {str(exception)}")
+            return {}  # Return an empty dictionary for this beamtime ID
+
+
+    def _get_local_metadata_per_run(self, meta_dir: str, run: str) -> dict:
+        """
+        Retrieves metadata for a specific run based on the PID from yaml file in the local beamtime folder.
+
+            Args:
+                pid (str): The PID of the run.
+
+            Returns:
+                dict: The retrieved metadata.
+
+            Raises:
+                Exception: If the request to retrieve metadata fails.
+        """
+        try:
+            run = str(run)
+            with open(f"{meta_dir}/{run}_1.yaml", 'r') as stream:
+                print("Getting metadata from local folder")
+                run_metadata = yaml.safe_load(stream)
+            return run_metadata
+
+        except Exception as exception:
+            logger.warning(f"Failed to retrieve metadata for PID {run}: {str(exception)}")
+            return {"_data":{}}  # Return an empty dictionary for this run