From b0c3c44dbd5232c92444ad1a6f15030d358d65ef Mon Sep 17 00:00:00 2001 From: amyjaynethompson <52806925+amyjaynethompson@users.noreply.github.com> Date: Thu, 30 Apr 2026 16:18:20 +0100 Subject: [PATCH 1/3] update xia2.multiplex file upload for new file structure --- src/dlstbx/wrapper/xia2_multiplex.py | 160 ++++++++++++++++++++------- 1 file changed, 121 insertions(+), 39 deletions(-) diff --git a/src/dlstbx/wrapper/xia2_multiplex.py b/src/dlstbx/wrapper/xia2_multiplex.py index e9f71f262..f5c366dca 100644 --- a/src/dlstbx/wrapper/xia2_multiplex.py +++ b/src/dlstbx/wrapper/xia2_multiplex.py @@ -248,8 +248,8 @@ def run(self): self.log.debug(result.stderr) self.log.info(f"working_directory: {working_directory}") - scaled_unmerged_mtz = working_directory / "scaled_unmerged.mtz" - multiplex_json = working_directory / "xia2.multiplex.json" + scaled_unmerged_mtz = working_directory / "DataFiles/scaled_unmerged.mtz" + multiplex_json = working_directory / "Processing/xia2.multiplex.json" # Placeholder logic to keep existing functionality - TODO - review if this is needed still if not (scaled_unmerged_mtz.is_file() and multiplex_json.is_file()): @@ -257,6 +257,10 @@ def run(self): # Create results directory results_directory.mkdir(parents=True, exist_ok=True) + subdirs = ("DataFiles", "LogFiles", "Processing") + for subdir in subdirs: + new_dir = results_directory / subdir + new_dir.mkdir(parents=True, exist_ok=True) if params.get("create_symlink"): dlstbx.util.symlink.create_parent_symlink( results_directory, params["create_symlink"] @@ -275,6 +279,10 @@ def is_final_result(final_file: pathlib.Path) -> bool: for patt in pipeline_final_params["patterns"] ) + for subdir in subdirs: + new_dir = final_directory / subdir + new_dir.mkdir(parents=True, exist_ok=True) + keep_ext = { ".png": None, ".log": "log", @@ -285,6 +293,8 @@ def is_final_result(final_file: pathlib.Path) -> bool: ".mtz": None, ".html": "log", ".sca": None, + ".bib": "misc", # want these copied, don't need in synchweb + ".phil": "misc", } keep = { "scaled.mtz": "result", @@ -299,8 +309,8 @@ def is_final_result(final_file: pathlib.Path) -> bool: # Record these log files first so they appear at the top of the list # of attachments in SynchWeb primary_log_files = [ - working_directory / "xia2.multiplex.html", - working_directory / "xia2.multiplex.log", + working_directory / "LogFiles/xia2.multiplex.html", + working_directory / "LogFiles/xia2.multiplex.log", ] allfiles = [] @@ -311,33 +321,53 @@ def is_final_result(final_file: pathlib.Path) -> bool: d = json.load(fh) for dataset_name, dataset in d["datasets"].items(): + base_dir_logs = working_directory / "LogFiles" + base_dir_results = working_directory / "DataFiles" + if dataset_name == "All data": - base_dir = working_directory + base_dir_processing = working_directory / "Processing" dimple_symlink = "dimple-xia2.multiplex" cluster_prefix = "" cluster_num = None + cluster_results = None elif "coordinate cluster" in dataset_name: cluster_count += 1 cluster_num = dataset_name.split(" ")[-1] cluster_prefix = f"coordinate_cluster_{cluster_num}_" - base_dir = working_directory / f"coordinate_cluster_{cluster_num}" + base_dir_processing = ( + working_directory + / f"Processing/coordinate_cluster_{cluster_num}" + ) dimple_symlink = ( f"dimple-xia2.multiplex-coordinate_cluster_{cluster_num}" ) + cluster_results = ( + results_directory + / f"Processing/coordinate_cluster_{cluster_num}" + ) + cluster_results.mkdir(parents=True, exist_ok=True) + cluster_final = ( + final_directory / f"Processing/coordinate_cluster_{cluster_num}" + ) + cluster_final.mkdir(parents=True, exist_ok=True) else: self.log.warning( f"Ignoring unrecognised dataset pattern {dataset_name}" ) continue - scaled_unmerged_mtz = base_dir / f"{cluster_prefix}scaled_unmerged.mtz" + scaled_unmerged_mtz = ( + base_dir_results / f"{cluster_prefix}scaled_unmerged.mtz" + ) i_obs = iotbx.merging_statistics.select_data( scaled_unmerged_mtz.as_posix(), data_labels=None ) merging_stats = dataset["merging_stats"] merging_stats_anom = dataset["merging_stats_anom"] - with (base_dir / f"{cluster_prefix}merging-stats.json").open("w") as fh: + with (base_dir_results / f"{cluster_prefix}merging-stats.json").open( + "w" + ) as fh: json.dump(merging_stats, fh) ispyb_d = { @@ -379,7 +409,13 @@ def is_final_result(final_file: pathlib.Path) -> bool: xtriage_results = dataset.get("xtriage") attachments = [] - for filename in set(primary_log_files + list(base_dir.iterdir())): + for filename in set( + primary_log_files + + list(base_dir_processing.iterdir()) + + list(base_dir_results.iterdir()) + + list(base_dir_logs.iterdir()) + + list(working_directory.iterdir()) + ): filetype = None if not filename.is_file(): continue # primary log files may not actually exist @@ -390,14 +426,27 @@ def is_final_result(final_file: pathlib.Path) -> bool: if filetype is None: continue - destination = results_directory / filename.name - if ( - destination.as_posix() in allfiles - and filename not in primary_log_files - ): - destination = ( - results_directory / f"{cluster_prefix}{filename.name}" - ) + # Work out if in Processing, DataFiles or LogFiles (parent_dir) + + parent_dir = None + + for i in subdirs: + if i in filename.parts: + parent_dir = i + + if not parent_dir: + destination = results_directory / filename.name + else: + # Get folder of clustering results (subdir of Processing) + + if cluster_results and parent_dir == "Processing": + if cluster_results.name in filename.parts: + destination = cluster_results / filename.name + else: + self.log.warning(f"Broken paths for {filename}") + continue + else: + destination = results_directory / parent_dir / filename.name if destination.as_posix() not in allfiles: self.log.debug(f"Copying {filename} to {destination}") @@ -405,44 +454,77 @@ def is_final_result(final_file: pathlib.Path) -> bool: allfiles.append(destination.as_posix()) if pipeline_final_params and is_final_result(destination): - destination = final_directory / destination.name + for i, part in enumerate(destination.parts): + if part in subdirs: + destination = final_directory / pathlib.Path( + *destination.parts[i:] + ) if destination not in allfiles: self.log.debug(f"Copying {filename} to {destination}") shutil.copy(filename, destination) allfiles.append(destination.as_posix()) # Files uploaded separately for each cluster - if filetype: - attachments.append( - { - "file_path": destination.parent.as_posix(), - "file_name": destination.name, - "file_type": filetype, - "importance_rank": ( - 1 - if destination.name.endswith( - ( - "scaled.mtz", - "xia2.multiplex.html", - "xia2.multiplex.log", - ) - ) - else 2 - ), - } - ) + + file_data = { + "file_path": destination.parent.as_posix(), + "file_name": destination.name, + "file_type": filetype, + "importance_rank": ( + 1 + if destination.name.endswith( + ( + "scaled.mtz", + "xia2.multiplex.html", + "xia2.multiplex.log", + ) + ) + else 2 + ), + } + + if "DataFiles" in destination.parts and filetype: + # if it is a cluster, only append cluster files + if cluster_results and "coordinate_cluster" in destination.name: + attachments.append(file_data) + # if it is not a cluster, only append non-cluster files + elif ( + not cluster_results + and "coordinate_cluster" not in destination.name + ): + attachments.append(file_data) + + elif "LogFiles" in destination.parts and filetype: + # attach all log files for clusters and non-clusters + attachments.append(file_data) + + elif "Processing" in destination.parts and filetype: + # If cluster, only attach files that are in a cluster folder + if ( + cluster_results + and cluster_results.name in destination.parts + ): + attachments.append(file_data) + # it not a cluster, make sure file is not in a cluster folder + elif ( + not cluster_results + and destination.parent.name == "Processing" + ): + attachments.append(file_data) + # Add parameters to the environment to be picked up downstream by trigger function self.recwrap.environment.update( { "scaled_mtz": ( - results_directory / f"{cluster_prefix}scaled.mtz" + results_directory / f"DataFiles/{cluster_prefix}scaled.mtz" ).as_posix() } ) self.recwrap.environment.update( { "scaled_unmerged_mtz": ( - results_directory / f"{cluster_prefix}scaled_unmerged.mtz" + results_directory + / f"DataFiles/{cluster_prefix}scaled_unmerged.mtz" ).as_posix() } ) From 873d87d76ba6818d142c8fa3356747047f16a1bb Mon Sep 17 00:00:00 2001 From: amyjaynethompson <52806925+amyjaynethompson@users.noreply.github.com> Date: Thu, 30 Apr 2026 16:18:51 +0100 Subject: [PATCH 2/3] update dimple wrapper to put sym links to multiplex in parent folder instead of where mtz is in DataFiles --- src/dlstbx/wrapper/dimple.py | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/src/dlstbx/wrapper/dimple.py b/src/dlstbx/wrapper/dimple.py index 66247561f..23028a191 100644 --- a/src/dlstbx/wrapper/dimple.py +++ b/src/dlstbx/wrapper/dimple.py @@ -317,8 +317,14 @@ def run(self): os.fspath(self.results_directory), symlink ) mtzsymlink = mtz.parent / symlink - if not mtzsymlink.exists(): + if "multiplex" in mtzsymlink.name: + parts = list(mtzsymlink.parts) + parts.remove("DataFiles") + mtzsymlink = pathlib.Path(*parts) + deltapath = os.path.relpath(self.results_directory, mtz.parent.parent) + else: deltapath = os.path.relpath(self.results_directory, mtz.parent) + if not mtzsymlink.exists(): os.symlink(deltapath, mtzsymlink) for f in self.working_directory.iterdir(): if f.name.startswith("."): From 7d9339ba5dd247325e9d0fc4125f93d873dc93b2 Mon Sep 17 00:00:00 2001 From: amyjaynethompson <52806925+amyjaynethompson@users.noreply.github.com> Date: Thu, 30 Jul 2026 16:40:06 +0100 Subject: [PATCH 3/3] update for new mplx file structure --- src/dlstbx/wrapper/xia2_multiplex.py | 43 ++++++++----- .../wrapper/xia2_multiplex_filtering.py | 62 +++++++++++++++---- 2 files changed, 78 insertions(+), 27 deletions(-) diff --git a/src/dlstbx/wrapper/xia2_multiplex.py b/src/dlstbx/wrapper/xia2_multiplex.py index f5c366dca..11751be9e 100644 --- a/src/dlstbx/wrapper/xia2_multiplex.py +++ b/src/dlstbx/wrapper/xia2_multiplex.py @@ -248,8 +248,8 @@ def run(self): self.log.debug(result.stderr) self.log.info(f"working_directory: {working_directory}") - scaled_unmerged_mtz = working_directory / "DataFiles/scaled_unmerged.mtz" - multiplex_json = working_directory / "Processing/xia2.multiplex.json" + scaled_unmerged_mtz = working_directory / "DataFiles" / "scaled_unmerged.mtz" + multiplex_json = working_directory / "Processing" / "xia2.multiplex.json" # Placeholder logic to keep existing functionality - TODO - review if this is needed still if not (scaled_unmerged_mtz.is_file() and multiplex_json.is_file()): @@ -309,8 +309,8 @@ def is_final_result(final_file: pathlib.Path) -> bool: # Record these log files first so they appear at the top of the list # of attachments in SynchWeb primary_log_files = [ - working_directory / "LogFiles/xia2.multiplex.html", - working_directory / "LogFiles/xia2.multiplex.log", + working_directory / "xia2.multiplex.html", + working_directory / "xia2.multiplex.log", ] allfiles = [] @@ -336,18 +336,22 @@ def is_final_result(final_file: pathlib.Path) -> bool: cluster_prefix = f"coordinate_cluster_{cluster_num}_" base_dir_processing = ( working_directory - / f"Processing/coordinate_cluster_{cluster_num}" + / "Processing" + / "coordinate_cluster_{cluster_num}" ) dimple_symlink = ( f"dimple-xia2.multiplex-coordinate_cluster_{cluster_num}" ) cluster_results = ( results_directory - / f"Processing/coordinate_cluster_{cluster_num}" + / "Processing" + / "coordinate_cluster_{cluster_num}" ) cluster_results.mkdir(parents=True, exist_ok=True) cluster_final = ( - final_directory / f"Processing/coordinate_cluster_{cluster_num}" + final_directory + / "Processing" + / "coordinate_cluster_{cluster_num}" ) cluster_final.mkdir(parents=True, exist_ok=True) else: @@ -365,7 +369,7 @@ def is_final_result(final_file: pathlib.Path) -> bool: merging_stats = dataset["merging_stats"] merging_stats_anom = dataset["merging_stats_anom"] - with (base_dir_results / f"{cluster_prefix}merging-stats.json").open( + with (working_directory / f"{cluster_prefix}merging-stats.json").open( "w" ) as fh: json.dump(merging_stats, fh) @@ -454,11 +458,14 @@ def is_final_result(final_file: pathlib.Path) -> bool: allfiles.append(destination.as_posix()) if pipeline_final_params and is_final_result(destination): - for i, part in enumerate(destination.parts): - if part in subdirs: - destination = final_directory / pathlib.Path( - *destination.parts[i:] - ) + if any(i in destination.parts for i in subdirs): + destination = ( + final_directory + / destination.parent.name + / destination.name + ) + else: + destination = final_directory / destination.name if destination not in allfiles: self.log.debug(f"Copying {filename} to {destination}") shutil.copy(filename, destination) @@ -511,12 +518,17 @@ def is_final_result(final_file: pathlib.Path) -> bool: and destination.parent.name == "Processing" ): attachments.append(file_data) + else: + # Anything in the parent folder (ie main logs) should be attached for clusters and non-clusters + attachments.append(file_data) # Add parameters to the environment to be picked up downstream by trigger function self.recwrap.environment.update( { "scaled_mtz": ( - results_directory / f"DataFiles/{cluster_prefix}scaled.mtz" + results_directory + / "DataFiles" + / "{cluster_prefix}scaled.mtz" ).as_posix() } ) @@ -524,7 +536,8 @@ def is_final_result(final_file: pathlib.Path) -> bool: { "scaled_unmerged_mtz": ( results_directory - / f"DataFiles/{cluster_prefix}scaled_unmerged.mtz" + / "DataFiles" + / "{cluster_prefix}scaled_unmerged.mtz" ).as_posix() } ) diff --git a/src/dlstbx/wrapper/xia2_multiplex_filtering.py b/src/dlstbx/wrapper/xia2_multiplex_filtering.py index 66f8b238b..2b1de09d5 100644 --- a/src/dlstbx/wrapper/xia2_multiplex_filtering.py +++ b/src/dlstbx/wrapper/xia2_multiplex_filtering.py @@ -201,14 +201,22 @@ def run(self): self.log.info(f"working_directory: {working_directory}") - filtered_unmerged_mtz = working_directory / "filtered_unmerged.mtz" - multiplex_filtering_json = working_directory / "xia2.multiplex_filtering.json" + filtered_unmerged_mtz = ( + working_directory / "DataFiles" / "filtered_unmerged.mtz" + ) + multiplex_filtering_json = ( + working_directory / "Processing" / "xia2.multiplex_filtering.json" + ) if not (filtered_unmerged_mtz.is_file() and multiplex_filtering_json.is_file()): success = False # Create results directory results_directory.mkdir(parents=True, exist_ok=True) + subdirs = ("DataFiles", "LogFiles", "Processing") + for subdir in subdirs: + new_dir = results_directory / subdir + new_dir.mkdir(parents=True, exist_ok=True) if params.get("create_symlink"): dlstbx.util.symlink.create_parent_symlink( results_directory, params["create_symlink"] @@ -227,6 +235,10 @@ def is_final_result(final_file: pathlib.Path) -> bool: for patt in pipeline_final_params["patterns"] ) + for subdir in subdirs: + new_dir = final_directory / subdir + new_dir.mkdir(parents=True, exist_ok=True) + keep_ext = { ".png": None, ".log": "log", @@ -264,7 +276,9 @@ def is_final_result(final_file: pathlib.Path) -> bool: dataset = d["datasets"]["Filtered"] dimple_symlink = "dimple-xia2.multiplex_filtering" - filtered_unmerged_mtz = working_directory / "filtered_unmerged.mtz" + filtered_unmerged_mtz = ( + working_directory / "DataFiles" / "filtered_unmerged.mtz" + ) i_obs = iotbx.merging_statistics.select_data( filtered_unmerged_mtz.as_posix(), data_labels=None ) @@ -308,7 +322,17 @@ def is_final_result(final_file: pathlib.Path) -> bool: xtriage_results = dataset.get("xtriage") attachments = [] - for filename in set(primary_log_files + list(working_directory.iterdir())): + base_dir_processing = working_directory / "Processing" + base_dir_logs = working_directory / "LogFiles" + base_dir_results = working_directory / "DataFiles" + + for filename in set( + primary_log_files + + list(working_directory.iterdir()) + + list(base_dir_processing.iterdir()) + + list(base_dir_logs.iterdir()) + + list(base_dir_results.iterdir()) + ): filetype = None if not filename.is_file(): continue @@ -319,19 +343,29 @@ def is_final_result(final_file: pathlib.Path) -> bool: if filetype is None: continue - destination = results_directory / filename.name - if ( - destination.as_posix() in allfiles - and filename not in primary_log_files - ): + parent_dir = None + + for i in subdirs: + if i in filename.parts: + parent_dir = i + + if not parent_dir: destination = results_directory / filename.name + else: + destination = results_directory / parent_dir / filename.name if destination.as_posix() not in allfiles: self.log.debug(f"Copying {filename} to {destination}") shutil.copy(filename, destination) allfiles.append(destination.as_posix()) if pipeline_final_params and is_final_result(destination): - destination = final_directory / destination.name + if any(i in destination.parts for i in subdirs): + destination = ( + final_directory / destination.parent.name / destination.name + ) + else: + destination = final_directory / destination.name + if destination not in allfiles: self.log.debug(f"Copying {filename} to {destination}") shutil.copy(filename, destination) @@ -362,12 +396,16 @@ def is_final_result(final_file: pathlib.Path) -> bool: # As using the same downstream triggers, update "scaled_mtz" rather than calling it by "filtered.mtz" self.recwrap.environment.update( - {"scaled_mtz": (results_directory / "filtered.mtz").as_posix()} + { + "scaled_mtz": ( + results_directory / "DataFiles" / "filtered.mtz" + ).as_posix() + } ) self.recwrap.environment.update( { "scaled_unmerged_mtz": ( - results_directory / "filtered_unmerged.mtz" + results_directory / "DataFiles" / "filtered_unmerged.mtz" ).as_posix() } )