Upload scripts for Palmas.

src/pynxtools_spm/UploadScripts/dataconverter_script.py

@@ -0,0 +1,96 @@
+import importlib.metadata
+from pynxtools.dataconverter.convert import convert
+from pathlib import Path
+from typing import Optional
+from dataclasses import dataclass, asdict
+import zipfile
+
+import importlib
+
+package_path = Path(importlib.metadata.PackagePath("pynxtools_spm"))
+
+
+afm_path = package_path / "nomad" / "afm"
+stm_path = package_path / "nomad" / "stm"
+sts_path = package_path / "nomad" / "sts"
+
+afm_reader_inputs = (afm_path / "AFMExampleWithCustomization" / "eln_data.yaml",)
+stm_reader_inputs = (stm_path / "STMExampleWithCustomization" / "eln_data.yaml",)
+sts_reader_inputs = sts_path / "STSExampleWithCustomization" / "eln_data.yaml"
+
+
+@dataclass
+class SPMConvertInputParameters:
+    input_files: tuple
+    eln: str | Path
+    expriement_type: str
+    reader: str = "spm"
+    output: str = None
+    nxdl: Optional[str] = None
+    create_zip: Optional[bool] = True
+    skip_verify: Optional[bool] = False
+    config: Optional[str | Path] = None
+    raw_extension: Optional[str] = None
+
+
+def convert_spm_experiments(
+    input_params: SPMConvertInputParameters,
+):
+    """Convert SPM experirments."""
+
+    def get_base_raw__path(input_files: tuple, ext):
+        for file in input_files:
+            f_str = str(file)
+            if f_str.endswith(ext):
+                return f_str.split(".", maxsplit=1)[0]
+
+    base_path = None
+    if not isinstance(input_params, SPMConvertInputParameters):
+        raise ValueError(
+            "Input parameters must be an instance of SPMConvertInputParameters"
+        )
+
+    if not input_params.input_files:
+        raise ValueError("Input files are required to run an SPM experiment")
+    if not input_params.eln:
+        raise ValueError(
+            f"ELN is required to run an {input_params.expriement_type} experiment."
+        )
+    if not input_params.expriement_type:
+        raise ValueError("Experiment type is required to run an SPM experiment")
+
+    input_params.input_files = (input_params.input_files, input_params.eln)
+    if input_params.config is not None:
+        input_params.input_files = (
+            input_params.input_files,
+            input_params.config,
+        )
+
+    if input_params.expriement_type.lower() == "sts":
+        input_params.nxdl = input_params.nxdl | "NXsts"
+        if input_params.raw_extension is None:
+            input_params.raw_extension = ".dat"
+
+    elif input_params.expriement_type.lower() == "stm":
+        input_params.nxdl = input_params.nxdl | "NXstm"
+        if input_params.raw_extension is None:
+            input_params.raw_extension = ".sxm"
+
+    elif input_params.expriement_type.lower() == "afm":
+        input_params.nxdl = input_params.nxdl | "NXafm"
+        if input_params.raw_extension is None:
+            input_params.raw_extension = ".sxm"
+
+    base_path = get_base_raw__path(input_params.input_files, input_params.raw_extension)
+    if input_params.output is None:
+        input_params.output = base_path + ".nxs"
+    zip_file = base_path + ".zip"
+
+    convert(**asdict(input_params))
+
+    if input_params.create_zip:
+        with zipfile.ZipFile(zip_file, "w") as zipf:
+            zipf.write(input_params.output)
+            for file in input_params.input_files:
+                zipf.write(file)
+        return zip_file

src/pynxtools_spm/UploadScripts/files.py

@@ -0,0 +1,77 @@
+#
+# Copyright The NOMAD Authors.
+#
+# This file is part of NOMAD. See https://nomad-lab.eu for further info.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""Modules to handle files."""
+
+from dataclasses import dataclass
+from pathlib import Path
+import os
+import shutil
+from typing import Optional
+
+
+@dataclass
+class Dir:
+    full_path: Path
+    parent_dir: str
+
+
+@dataclass
+class File:
+    full_path: Path
+    file: str
+    ext: str
+    base: str
+    parent_dir: Dir
+
+
+# Copy create an script that copies a directory structure from source to destination directory
+def copy_directory_structure(
+    src: Path, dst: Path, extension: Optional[str], run_action_on_files=None
+):
+    """
+    Copies a directory structure from source to destination directory
+    only with the files having a specific extension.
+
+    :param src: Source directory path
+    :param dst: Destination directory path
+    :param extension: File extension to filter by
+    """
+    if not src.is_dir():
+        raise ValueError(f"Source {src} is not a directory")
+    if not dst.is_dir():
+        raise ValueError(f"Destination {dst} is not a directory")
+
+    for root, dirs, files in os.walk(src):
+        if files:
+            for file in files:
+                if extension:
+                    if not file.endswith(extension):
+                        continue
+
+                src_file = Path(root) / file
+                dest_file = dst / file
+                shutil.copy2(src_file, dest_file)
+                if run_action_on_files is not None:
+                    run_action_on_files(dest_file)
+
+        if dirs:
+            for dir_ in dirs:
+                nested_dir = dst / dir_
+                os.makedirs(nested_dir, exist_ok=True)
+                copy_directory_structure(src / dir_, nested_dir, extension)