Manage .keep files

logbook2mouse/experiment.py

@@ -1,5 +1,6 @@
 import attrs
 from typing import List
+from pathlib import Path
 
 import epics
 
@@ -9,6 +10,7 @@ class ExperimentVariables:
     eiger_prefix: str = "detector_eiger"
     parrot_prefix: str = "pa0"
     image_processing_prefix: str = "image"
+    data_dir: Path = Path("~/data/")
 
     def __attrs_post_init__(self):
         epics.caput(f"{self.eiger_prefix}:Initialize", True)

logbook2mouse/file_management.py

@@ -3,14 +3,19 @@
 import numpy as np
 from pathlib import Path
 
-def work_directory(entry, basedir = Path("/home/ws8665-epics/data/")) -> Path:
+from logbook2mouse.logbook_reader import Logbook2MouseEntry
+from logbook2mouse.experiment import ExperimentVariables
+
+def work_directory(entry: Logbook2MouseEntry, experiment: ExperimentVariables) -> Path:
     timestamp = entry.date
     ymd = timestamp.strftime("%Y%m%d")
-    work_directory = basedir / str(timestamp.year) / ymd
+    work_directory = experiment.data_dir / str(timestamp.year) / ymd
     if os.path.exists(work_directory):
         logging.info(f"Measurement directory {ymd} already in use. Data will be added to this directory. ")
     else:
         os.mkdir(work_directory)
+    # ensure .keep file is present
+    (work_directory / ".keep").touch(exist_ok = True)
     return work_directory
 
 def scan_counter_next(scan_counter, work_directory, entry):

logbook2mouse/measure_config.py

@@ -213,7 +213,7 @@ def measure_at_config(
 
     epics.caput(f"{experiment.parrot_prefix}:exp:nrep", repetitions)
     scan_counter = 0
-    work_directory = filemanagement.work_directory(entry)
+    work_directory = filemanagement.work_directory(entry, experiment)
     ymd = entry.date.strftime("%Y%m%d")
     epics.caput(f"{experiment.parrot_prefix}:exp:logbook_date", ymd)
 

logbook2mouse/measurement_script.py

@@ -9,6 +9,7 @@
 logger = logging.getLogger("logbook2mouse")
 
 from logbook2mouse.logbook_reader import Logbook2MouseEntry
+from logbook2mouse.file_management import work_directory
 from logbook2mouse.measure_config import standard_configurations, default_repetitions
 
 @attrs.define
@@ -24,6 +25,16 @@ def __attrs_post_init__(self):
         assert self.output_directory.exists(), f"Output directory does not exist: {self.output_directory}"
         assert self.protocols_directory.exists(), f"Protocols directory does not exist: {self.protocols_directory}"
 
+    def identify_ymd_subdirs(self):
+        """
+        Identify what YMDs are written to based on the list of Logbook2MouseEntries.
+        """
+
+        ymd_directories_used = set()
+        for entry in self.entries:
+            ymd_directories_used = ymd_directories_used | {Path(str(entry.date.year)) / Path(entry.ymd)}
+        return ymd_directories_used
+
     def collate_configurations(self):
         """
         Generate configurations to be measured and collect in dataframe.
@@ -142,6 +153,7 @@ def generate_script(self):
         measurements_overall = self.calculate_number_of_measurements(entry_df)
         logger.info(f"{measurements_overall} measurements are planned overall.")
         script_lines += r"caput(f'{experiment.parrot_prefix}:exp:progress:measurements_overall'," + f"{measurements_overall}" + ")\n"
+
         for r, row in entry_df.iterrows():
             for e, entry in row.items():
                 if type(entry) == Logbook2MouseEntry:
@@ -152,6 +164,11 @@ def generate_script(self):
 
         # Script shutdown
         script_lines += self.load_protocol_template(protocol_path=self.protocols_directory/'teardown.py')
+        # allow removal of directories used
+        ymd_directories_used = self.identify_ymd_subdirs()
+        for ymd_dir in ymd_directories_used:
+            script_lines += r"Path(f'{experiment.data_dir}" + f"/{ymd_dir}/.keep').unlink()" + "\n"
+
 
         return "".join(script_lines)
 

protocols/20250109_alignment.py

@@ -34,7 +34,7 @@ def get_float_parameter(entry, keyword, default):
 
 # define where to save scans
 ymd = entry.date.strftime("%Y%m%d")
-scan_dir = Path(work_directory(entry)) / f"{ymd}_{entry.batchnum}_{0}" / "scans"
+scan_dir = Path(work_directory(entry, experiment)) / f"{ymd}_{entry.batchnum}_{0}" / "scans"
 os.makedirs(scan_dir / "scan_0", exist_ok=True)
 
 move_to_sampleposition(experiment, entry.sampleposition)

protocols/20250116_gisaxs.py

@@ -29,7 +29,7 @@ def get_float_parameter(entry, keyword, default):
     repetitions = int(float(repetitions))
 
 ymd = entry.date.strftime("%Y%m%d")
-wd = Path(work_directory(entry))
+wd = Path(work_directory(entry, experiment))
 aligned_data =  wd / f"{ymd}_{alignment_batch}_{1}"
 scan_dir = wd / f"{ymd}_{entry.batchnum}_{scan_counter_next(0, wd, entry)}"
 

protocols/20250116_reflectivity.py

@@ -17,7 +17,7 @@
 alignment_batch = int(float(entry.additional_parameters.get('alignment_batch', entry.batchnum)))
 
 ymd = entry.date.strftime("%Y%m%d")
-wd = Path(work_directory(entry))
+wd = Path(work_directory(entry, experiment))
 aligned_data =  wd / f"{ymd}_{alignment_batch}_{1}"
 scan_dir = wd / f"{ymd}_{entry.batchnum}_{scan_counter_next(0, wd, entry)}"
 

protocols/20250207_high_resolution_reflectivity.py

@@ -17,7 +17,7 @@
 alignment_batch = int(float(entry.additional_parameters.get('alignment_batch', entry.batchnum)))
 
 ymd = entry.date.strftime("%Y%m%d")
-wd = Path(work_directory(entry))
+wd = Path(work_directory(entry, experiment))
 aligned_data =  wd / f"{ymd}_{alignment_batch}_{1}"
 scan_dir = wd / f"{ymd}_{entry.batchnum}_{scan_counter_next(0, wd, entry)}"
 

protocols/20250826_roll_alignment.py

@@ -34,7 +34,7 @@ def get_float_parameter(entry, keyword, default):
 
 # define where to save scans
 ymd = entry.date.strftime("%Y%m%d")
-scan_dir = Path(work_directory(entry)) / f"{ymd}_{entry.batchnum}_{0}" / "scans"
+scan_dir = Path(work_directory(entry, experiment)) / f"{ymd}_{entry.batchnum}_{0}" / "scans"
 os.makedirs(scan_dir / "scan_0", exist_ok=True)
 
 move_to_sampleposition(experiment, entry.sampleposition)

protocols/setup.py

@@ -3,6 +3,7 @@
 import logging
 logger = logging.getLogger("measurement")
 logger.setLevel(logging.INFO)
+from pathlib import Path
 from epics import caget, caput
 from pandas import Timestamp
 import logbook2mouse.detector as detector
@@ -52,7 +53,7 @@
     if value is None:
         raise ConnectionError(f"PV '{pv}' is not reachable or has no value.")
 
-experiment = ExperimentVariables(required_pvs)
+experiment = ExperimentVariables(required_pvs, data_dir = Path("/home/ws8665-epics/data/"))
 
 # ensure motor positions are at least checked at the beginning of a measurement
 caput("pa0:config:config_id", 999)