More rsync fixes (#360)

* Replaced 'procrunner' with 'subprocess'
* Fixed rsync subprocesses by prepending them with 'bash -c'

Co-authored by: stephen-riggs <stephen.riggs@diamond.ac.uk>

Author: tieneupin

src/murfey/client/__init__.py

@@ -65,6 +65,31 @@ def read_config() -> configparser.ConfigParser:
 requests.delete = partial(requests.delete, headers={"Authorization": f"Bearer {token}"})
 
 
+def write_config(config: configparser.ConfigParser):
+    mcch = os.environ.get("MURFEY_CLIENT_CONFIG_HOME")
+    murfey_client_config_home = Path(mcch) if mcch else Path.home()
+    with open(murfey_client_config_home / ".murfey", "w") as configfile:
+        config.write(configfile)
+
+
+def main_loop(
+    source_watchers: List[murfey.client.watchdir.DirWatcher],
+    appearance_time: float,
+    transfer_all: bool,
+):
+    log.info(
+        f"Murfey {murfey.__version__} on Python {'.'.join(map(str, sys.version_info[0:3]))} entering main loop"
+    )
+    if appearance_time > 0:
+        modification_time: float | None = time.time() - appearance_time * 3600
+    else:
+        modification_time = None
+    while True:
+        for sw in source_watchers:
+            sw.scan(modification_time=modification_time, transfer_all=transfer_all)
+        time.sleep(15)
+
+
 def _enable_webbrowser_in_cygwin():
     """Helper function to make webbrowser.open() work in CygWin"""
     if "cygwin" in platform.system().lower() and shutil.which("cygstart"):
@@ -325,28 +350,3 @@ def run():
     )
     app.run()
     rich_handler.redirect = False
-
-
-def main_loop(
-    source_watchers: List[murfey.client.watchdir.DirWatcher],
-    appearance_time: float,
-    transfer_all: bool,
-):
-    log.info(
-        f"Murfey {murfey.__version__} on Python {'.'.join(map(str, sys.version_info[0:3]))} entering main loop"
-    )
-    if appearance_time > 0:
-        modification_time: float | None = time.time() - appearance_time * 3600
-    else:
-        modification_time = None
-    while True:
-        for sw in source_watchers:
-            sw.scan(modification_time=modification_time, transfer_all=transfer_all)
-        time.sleep(15)
-
-
-def write_config(config: configparser.ConfigParser):
-    mcch = os.environ.get("MURFEY_CLIENT_CONFIG_HOME")
-    murfey_client_config_home = Path(mcch) if mcch else Path.home()
-    with open(murfey_client_config_home / ".murfey", "w") as configfile:
-        config.write(configfile)

src/murfey/client/multigrid_control.py

@@ -1,5 +1,6 @@
 import json
 import logging
+import subprocess
 import threading
 from dataclasses import dataclass, field
 from datetime import datetime
@@ -8,7 +9,6 @@
 from typing import Dict, List, Optional
 from urllib.parse import urlparse
 
-import procrunner
 import requests
 
 import murfey.client.websocket
@@ -19,7 +19,7 @@
 from murfey.client.rsync import RSyncer, RSyncerUpdate, TransferResult
 from murfey.client.tui.screens import determine_default_destination
 from murfey.client.watchdir import DirWatcher
-from murfey.util import capture_post
+from murfey.util import capture_post, posix_path
 
 log = logging.getLogger("murfey.client.mutligrid_control")
 
@@ -180,16 +180,23 @@ def _start_rsyncer(
         if self._environment:
             self._environment.default_destinations[source] = destination
             if self._environment.gain_ref and visit_path:
-                gain_rsync = procrunner.run(
-                    [
-                        "rsync",
-                        str(self._environment.gain_ref),
-                        f"{self._environment.url.hostname}::{visit_path}/processing",
-                    ]
-                )
+                # Set up rsync command
+                rsync_cmd = [
+                    "rsync",
+                    f"{posix_path(self._environment.gain_ref)!r}",  # '!r' will print strings in ''
+                    f"{self._environment.url.hostname}::{visit_path}/processing",
+                ]
+                # Wrap in bash shell
+                cmd = [
+                    "bash",
+                    "-c",
+                    " ".join(rsync_cmd),
+                ]
+                # Run rsync subprocess
+                gain_rsync = subprocess.run(cmd)
                 if gain_rsync.returncode:
                     log.warning(
-                        f"Gain reference file {self._environment.gain_ref} was not successfully transferred to {visit_path}/processing"
+                        f"Gain reference file {posix_path(self._environment.gain_ref)!r} was not successfully transferred to {visit_path}/processing"
                     )
         if transfer:
             self.rsync_processes[source] = RSyncer(

src/murfey/client/rsync.py

@@ -18,8 +18,6 @@
 from typing import Callable, List, NamedTuple
 from urllib.parse import ParseResult
 
-import procrunner
-
 from murfey.client.tui.status_bar import StatusBar
 from murfey.util import Observer
 
@@ -380,16 +378,19 @@ def parse_stdout(line: str):
                 return
 
         def parse_stderr(line: str):
-            logger.warning(f"rsync stderr: {line!r}")
+            if line.strip():
+                logger.warning(f"rsync stderr: {line!r}")
 
         # Generate list of relative filenames for this batch of transferred files
+        #   Relative filenames will be safe to use on both Windows and Unix
         relative_filenames: List[Path] = []
         for f in files:
             try:
                 relative_filenames.append(f.relative_to(self._basepath))
             except ValueError:
                 raise ValueError(f"File '{f}' is outside of {self._basepath}") from None
 
+        # Encode files to rsync as bytestring
         if self._remove_files:
             if self._required_substrings_for_removal:
                 rsync_stdin_remove = b"\n".join(
@@ -418,52 +419,88 @@ def parse_stderr(line: str):
             rsync_stdin = b"\n".join(os.fsencode(f) for f in relative_filenames)
 
         # Create and run rsync subprocesses
-        # rsync commands to pass to subprocess
+        # rsync default settings
         rsync_cmd = [
             "rsync",
             "-iiv",
             "--times",
             "--progress",
             "--outbuf=line",
-            "--files-from=-",
-            "-p",  # preserve permissions
-            "--chmod=D0750,F0750",  # 4: Read, 2: Write, 1: Execute | User, Group, Others
+            "--files-from=-",  # '-' indicates reading from standard input
+            # Needed as a pair to trigger permission modifications
+            # Ref: https://serverfault.com/a/796341
+            "-p",
+            "--chmod=D0750,F0750",  # Use extended chmod format
         ]
+        # Add file locations
         rsync_cmd.extend([".", self._remote])
-        result: subprocess.CompletedProcess | None = None
+
+        # Transfer files to destination
+        result: subprocess.CompletedProcess[bytes] | None = None
         success = True
         if rsync_stdin:
-            result = procrunner.run(
-                rsync_cmd,
-                callback_stdout=parse_stdout,
-                callback_stderr=parse_stderr,
-                working_directory=str(self._basepath),
-                stdin=rsync_stdin,
-                print_stdout=False,
-                print_stderr=False,
+            # Wrap rsync command in a bash command
+            cmd = [
+                "bash",
+                "-c",
+                # rsync command passed in as a single string
+                " ".join(rsync_cmd),
+            ]
+            result = subprocess.run(
+                cmd,
+                cwd=self._basepath,  # As-is Path is fine
+                capture_output=True,
+                input=rsync_stdin,
             )
-            success = result.returncode == 0 if result else False
-
+            # Parse outputs
+            for line in result.stdout.decode("utf-8", "replace").split("\n"):
+                parse_stdout(line)
+            for line in result.stderr.decode("utf-8", "replace").split("\n"):
+                parse_stderr(line)
+            success = result.returncode == 0
+
+        # Remove files from source
         if rsync_stdin_remove:
+            # Insert file removal flag before locations
             rsync_cmd.insert(-2, "--remove-source-files")
-            result = procrunner.run(
-                rsync_cmd,
-                callback_stdout=parse_stdout,
-                callback_stderr=parse_stderr,
-                working_directory=str(self._basepath),
-                stdin=rsync_stdin_remove,
-                print_stdout=False,
-                print_stderr=False,
+            # Wrap rsync command in a bash command
+            cmd = [
+                "bash",
+                "-c",
+                # Pass rsync command as single string
+                " ".join(rsync_cmd),
+            ]
+            result = subprocess.run(
+                cmd,
+                cwd=self._basepath,
+                capture_output=True,
+                input=rsync_stdin_remove,
             )
-
+            # Parse outputs
+            for line in result.stdout.decode("utf-8", "replace").split("\n"):
+                parse_stdout(line)
+            for line in result.stderr.decode("utf-8", "replace").split("\n"):
+                parse_stderr(line)
+            # Leave it as a failure if the previous rsync subprocess failed
             if success:
-                success = result.returncode == 0 if result else False
+                success = result.returncode == 0
 
         self.notify(successful_updates, secondary=True)
 
+        # Print out a summary message for each file transfer batch instead of individual messages
+        # List out file paths as stored in memory to see if issue is due to file path mismatch
+        if len(set(relative_filenames) - transfer_success) != 0:
+            logger.debug(
+                f"Files identified for transfer ({len(relative_filenames)}): {relative_filenames!r}"
+            )
+            logger.debug(
+                f"Files successfully transferred ({len(transfer_success)}): {list(transfer_success)!r}"
+            )
+
         # Compare files from rsync stdout to original list to verify transfer
         for f in set(relative_filenames) - transfer_success:
-            logger.warning(f"Transfer of file {f.name!r} considered a failure")
+            # Mute individual file warnings; replace with summarised one above
+            # logger.warning(f"Transfer of file {f.name!r} considered a failure")
             self._files_transferred += 1
             current_outstanding = self.queue.unfinished_tasks - (
                 self._files_transferred - previously_transferred

src/murfey/client/tui/app.py

@@ -1,14 +1,14 @@
 from __future__ import annotations
 
 import logging
+import subprocess
 from datetime import datetime
 from functools import partial
 from pathlib import Path
 from queue import Queue
 from typing import Awaitable, Callable, Dict, List, OrderedDict, TypeVar
 from urllib.parse import urlparse
 
-import procrunner
 import requests
 from textual.app import App
 from textual.reactive import reactive
@@ -36,6 +36,7 @@
 from murfey.util import (
     capture_post,
     get_machine_config_client,
+    posix_path,
     read_config,
     set_default_acquisition_output,
 )
@@ -211,16 +212,23 @@ def _start_rsyncer(
         if self._environment:
             self._environment.default_destinations[source] = destination
             if self._environment.gain_ref and visit_path:
-                gain_rsync = procrunner.run(
-                    [
-                        "rsync",
-                        str(self._environment.gain_ref),
-                        f"{self._url.hostname}::{visit_path}/processing",
-                    ]
-                )
+                # Set up rsync command
+                rsync_cmd = [
+                    "rsync",
+                    f"{posix_path(self._environment.gain_ref)!r}",
+                    f"{self._url.hostname}::{visit_path}/processing",
+                ]
+                # Encase in bash shell
+                cmd = [
+                    "bash",
+                    "-c",
+                    " ".join(rsync_cmd),
+                ]
+                # Run rsync subprocess
+                gain_rsync = subprocess.run(cmd)
                 if gain_rsync.returncode:
                     log.warning(
-                        f"Gain reference file {self._environment.gain_ref} was not successfully transferred to {visit_path}/processing"
+                        f"Gain reference file {posix_path(self._environment.gain_ref)!r} was not successfully transferred to {visit_path}/processing"
                     )
         if transfer:
             self.rsync_processes[source] = RSyncer(

src/murfey/client/tui/screens.py

@@ -2,6 +2,7 @@
 
 # import contextlib
 import logging
+import subprocess
 from datetime import datetime
 from functools import partial
 from pathlib import Path
@@ -17,7 +18,6 @@
     TypeVar,
 )
 
-import procrunner
 import requests
 from pydantic import BaseModel, ValidationError
 from rich.box import SQUARE
@@ -56,7 +56,7 @@
 )
 from murfey.client.rsync import RSyncer
 from murfey.client.tui.forms import FormDependency
-from murfey.util import capture_post, get_machine_config_client, read_config
+from murfey.util import capture_post, get_machine_config_client, posix_path, read_config
 from murfey.util.models import PreprocessingParametersTomo, ProcessingParametersSPA
 
 log = logging.getLogger("murfey.tui.screens")
@@ -906,18 +906,22 @@ def on_button_pressed(self, event):
             if event.button.id == "suggested-gain-ref":
                 self._dir_tree._gain_reference = self._gain_reference
             visit_path = f"data/{datetime.now().year}/{self.app._environment.visit}"
-            cmd = [
+            # Set up rsync command
+            rsync_cmd = [
                 "rsync",
-                str(self._dir_tree._gain_reference),
+                f"{posix_path(self._dir_tree._gain_reference)!r}",
                 f"{self.app._environment.url.hostname}::{visit_path}/processing/{secure_filename(self._dir_tree._gain_reference.name)}",
             ]
+            # Encase in bash shell
+            cmd = ["bash", "-c", " ".join(rsync_cmd)]
             if self.app._environment.demo:
                 log.info(f"Would perform {' '.join(cmd)}")
             else:
-                gain_rsync = procrunner.run(cmd)
+                # Run rsync subprocess
+                gain_rsync = subprocess.run(cmd)
                 if gain_rsync.returncode:
                     log.warning(
-                        f"Gain reference file {self._dir_tree._gain_reference} was not successfully transferred to {visit_path}/processing"
+                        f"Gain reference file {posix_path(self._dir_tree._gain_reference)!r} was not successfully transferred to {visit_path}/processing"
                     )
             process_gain_response = requests.post(
                 url=f"{str(self.app._environment.url.geturl())}/sessions/{self.app._environment.murfey_session}/process_gain",

src/murfey/instrument_server/api.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import secrets
 import time
 from datetime import datetime