Merge main

Author: d-j-hatton

.github/workflows/ci.yml

@@ -41,6 +41,7 @@ jobs:
 
     steps:
       - uses: actions/checkout@v4
+
       - name: Use Python ${{ matrix.python-version }}
         uses: actions/setup-python@v5
         with:
@@ -67,13 +68,13 @@ jobs:
           docker run --detach --name rabbitmq -p 127.0.0.1:5672:5672 -p 127.0.0.1:15672:15672 test-rabbitmq
           docker container list -a
 
-      - name: Get ispyb database
+      - name: Get ISPyB database
         uses: actions/download-artifact@v4
         with:
           name: database
           path: database/
 
-      - name: Install package
+      - name: Install Murfey
         run: |
           set -eux
           pip install --disable-pip-version-check -e "."[cicd,client,server,developer]
@@ -84,7 +85,7 @@ jobs:
           mysql-version: "11.3"
           auto-start: false
 
-      - name: Set up test ipsyb database
+      - name: Set up test ISPyB database
         run: |
           set -eu
           cp ".github/workflows/config/my.cnf" .my.cnf
@@ -101,9 +102,14 @@ jobs:
                    schemas/ispyb/routines.sql \
                    grants/ispyb_processing.sql \
                    grants/ispyb_import.sql; do
-              echo Importing ${f}...
+
+              echo "Patching ${f} in SQL files to fix CLI escape issues..."
+              sed -i 's/\\-/-/g' "$f"
+
+              echo "Importing ${f}..."
               mariadb --defaults-file=.my.cnf < $f
           done
+
           mariadb --defaults-file=.my.cnf -e "CREATE USER ispyb_api@'%' IDENTIFIED BY 'password_1234'; GRANT ispyb_processing to ispyb_api@'%'; GRANT ispyb_import to ispyb_api@'%'; SET DEFAULT ROLE ispyb_processing FOR ispyb_api@'%';"
           mariadb --defaults-file=.my.cnf -e "CREATE USER ispyb_api_future@'%' IDENTIFIED BY 'password_4321'; GRANT SELECT ON ispybtest.* to ispyb_api_future@'%';"
           mariadb --defaults-file=.my.cnf -e "CREATE USER ispyb_api_sqlalchemy@'%' IDENTIFIED BY 'password_5678'; GRANT SELECT ON ispybtest.* to ispyb_api_sqlalchemy@'%'; GRANT INSERT ON ispybtest.* to ispyb_api_sqlalchemy@'%'; GRANT UPDATE ON ispybtest.* to ispyb_api_sqlalchemy@'%';"
@@ -112,18 +118,17 @@ jobs:
       - name: Check RabbitMQ is alive
         run: wget -t 10 -w 1 http://127.0.0.1:15672 -O -
 
-      - name: Run tests
+      - name: Run Murfey tests
         env:
           POSTGRES_HOST: localhost
           POSTGRES_PORT: 5432
           POSTGRES_DB: murfey_test_db
           POSTGRES_PASSWORD: psql_pwd
           POSTGRES_USER: psql_user
         run: |
-          export ISPYB_CREDENTIALS=".github/workflows/config/ispyb.cfg"
           PYTHONDEVMODE=1 pytest -v -ra --cov=murfey --cov-report=xml --cov-branch
 
-      - name: Upload to Codecov
+      - name: Upload test results to Codecov
         uses: codecov/codecov-action@v5
         with:
           name: ${{ matrix.python-version }}

.github/workflows/test.yml

@@ -3,11 +3,12 @@ name: Build and test
 on: [push, pull_request]
 
 env:
-  DATABASE_SCHEMA: 4.2.1 # released 2024-08-19
+  ISPYB_DATABASE_SCHEMA: 4.6.0
   # Installs from GitHub
   # Versions: https://github.com/DiamondLightSource/ispyb-database/tags
   # Previous version(s):
-  # 4.1.0
+  # 4.2.1 # released 2024-08-19
+  # 4.1.0 # released 2024-03-26
 
 permissions:
   contents: read
@@ -53,10 +54,10 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4
-      - name: Download ISPyB DB schema v${{ env.DATABASE_SCHEMA }} for tests
+      - name: Download ISPyB DB schema v${{ env.ISPYB_DATABASE_SCHEMA }} for tests
         run: |
           mkdir database
-          wget -t 3 --waitretry=20 https://github.com/DiamondLightSource/ispyb-database/releases/download/v${{ env.DATABASE_SCHEMA }}/ispyb-database-${{ env.DATABASE_SCHEMA }}.tar.gz -O database/ispyb-database.tar.gz
+          wget -t 3 --waitretry=20 https://github.com/DiamondLightSource/ispyb-database/releases/download/v${{ env.ISPYB_DATABASE_SCHEMA }}/ispyb-database-${{ env.ISPYB_DATABASE_SCHEMA }}.tar.gz -O database/ispyb-database.tar.gz
       - name: Store database artifact
         uses: actions/upload-artifact@v4
         with:

src/murfey/cli/spa_ispyb_messages.py

@@ -19,7 +19,7 @@
 
 from murfey.client.contexts.spa import _get_xml_list_index
 from murfey.server import _murfey_id, _register
-from murfey.server.ispyb import Session, TransportManager, get_session_id
+from murfey.server.ispyb import ISPyBSession, TransportManager, get_session_id
 from murfey.server.murfey_db import url
 from murfey.util import db
 from murfey.util.config import get_machine_config, get_microscope, get_security_config
@@ -256,7 +256,7 @@ def run():
             proposal_code=args.visit[:2],
             proposal_number=args.visit[2:].split("-")[0],
             visit_number=args.visit.split("-")[1],
-            db=Session(),
+            db=ISPyBSession(),
         ),
     )
 

src/murfey/client/__init__.py

@@ -11,8 +11,9 @@
 import webbrowser
 from datetime import datetime
 from pathlib import Path
+from pprint import pprint
 from queue import Queue
-from typing import List, Literal
+from typing import Literal
 from urllib.parse import ParseResult, urlparse
 
 import requests
@@ -53,7 +54,7 @@ def write_config(config: configparser.ConfigParser):
 
 
 def main_loop(
-    source_watchers: List[murfey.client.watchdir.DirWatcher],
+    source_watchers: list[murfey.client.watchdir.DirWatcher],
     appearance_time: float,
     transfer_all: bool,
 ):
@@ -104,6 +105,7 @@ def _check_for_updates(
 
 
 def run():
+    # Load client config and server information
     config = read_config()
     instrument_name = config["Murfey"]["instrument_name"]
     try:
@@ -122,6 +124,7 @@ def run():
     else:
         known_server = config["Murfey"].get("server")
 
+    # Set up argument parser with dynamic defaults based on client config
     parser = argparse.ArgumentParser(description="Start the Murfey client")
     parser.add_argument(
         "--server",
@@ -207,23 +210,23 @@ def run():
         default=False,
         help="Do not trigger processing for any data directories currently on disk (you may have started processing for them in a previous murfey run)",
     )
-
     args = parser.parse_args()
 
+    # Logic to exit early based on parsed args
     if not args.server:
         exit("Murfey server not set. Please run with --server host:port")
     if not args.server.startswith(("http://", "https://")):
         if "://" in args.server:
             exit("Unknown server protocol. Only http:// and https:// are allowed")
         args.server = f"http://{args.server}"
-
     if args.remove_files:
         remove_prompt = Confirm.ask(
             f"Are you sure you want to remove files from {args.source or Path('.').absolute()}?"
         )
         if not remove_prompt:
             exit("Exiting")
 
+    # If a new server URL is provided, save info to config file
     murfey_url = urlparse(args.server, allow_fragments=False)
     if args.server != known_server:
         # New server specified. Verify that it is real
@@ -245,8 +248,7 @@ def run():
     if args.no_transfer:
         log.info("No files will be transferred as --no-transfer flag was specified")
 
-    from pprint import pprint
-
+    # Check ISPyB (if set up) for ongoing visits
     ongoing_visits = []
     if args.visit:
         ongoing_visits = [args.visit]
@@ -263,35 +265,38 @@ def run():
 
     _enable_webbrowser_in_cygwin()
 
+    # Set up additional log handlers
     log.setLevel(logging.DEBUG)
     log_queue = Queue()
     input_queue = Queue()
 
-    # rich_handler = DirectableRichHandler(log_queue, enable_link_path=False)
+    # Rich-based console handler
     rich_handler = DirectableRichHandler(enable_link_path=False)
     rich_handler.setLevel(logging.DEBUG if args.debug else logging.INFO)
 
+    # Set up websocket app and handler
     client_id = requests.get(f"{murfey_url.geturl()}/new_client_id/").json()
     ws = murfey.client.websocket.WSApp(
         server=args.server,
         id=client_id["new_id"],
     )
+    ws_handler = CustomHandler(ws.send)
 
+    # Add additional handlers and set logging levels
     logging.getLogger().addHandler(rich_handler)
-    handler = CustomHandler(ws.send)
-    logging.getLogger().addHandler(handler)
+    logging.getLogger().addHandler(ws_handler)
     logging.getLogger("murfey").setLevel(logging.INFO)
     logging.getLogger("websocket").setLevel(logging.WARNING)
 
     log.info("Starting Websocket connection")
 
-    status_bar = StatusBar()
-
+    # Load machine data for subsequent sections
     machine_data = requests.get(
         f"{murfey_url.geturl()}/instruments/{instrument_name}/machine"
     ).json()
     gain_ref: Path | None = None
 
+    # Set up Murfey environment instance and map it to websocket app
     instance_environment = MurfeyInstanceEnvironment(
         url=murfey_url,
         client_id=ws.id,
@@ -308,9 +313,10 @@ def run():
             else ""
         ),
     )
-
     ws.environment = instance_environment
 
+    # Set up and run Murfey TUI app
+    status_bar = StatusBar()
     rich_handler.redirect = True
     app = MurfeyTUI(
         environment=instance_environment,

src/murfey/client/multigrid_control.py

@@ -54,6 +54,7 @@ class MultigridController:
     data_collection_parameters: dict = field(default_factory=lambda: {})
     token: str = ""
     _machine_config: dict = field(default_factory=lambda: {})
+    visit_end_time: Optional[datetime] = None
 
     def __post_init__(self):
         if self.token:
@@ -105,6 +106,15 @@ def __post_init__(self):
             register_client=False,
         )
 
+        if self.visit_end_time:
+            current_time = datetime.now()
+            server_timestamp = requests.get(
+                f"{self.murfey_url}{session_router.url_path_for('get_current_timestamp')}"
+            ).json()["timestamp"]
+            self.visit_end_time += current_time - datetime.fromtimestamp(
+                server_timestamp
+            )
+
     def _multigrid_watcher_finalised(self):
         self.multigrid_watcher_active = False
         self.dormancy_check()
@@ -156,7 +166,8 @@ def _start_rsyncer_multigrid(
         tag: str = "",
         limited: bool = False,
     ):
-        log.info(f"starting multigrid rsyncer: {source}")
+        log.info(f"Starting multigrid rsyncer: {source}")
+        log.debug(f"Analysis of {source} is {('enabled' if analyse else 'disabled')}")
         destination_overrides = destination_overrides or {}
         machine_data = requests.get(
             f"{self._environment.url.geturl()}{session_router.url_path_for('machine_info_by_instrument', instrument_name=self.instrument_name)}"
@@ -283,6 +294,7 @@ def _start_rsyncer(
                 stop_callback=self._rsyncer_stopped,
                 do_transfer=self.do_transfer,
                 remove_files=remove_files,
+                end_time=self.visit_end_time,
             )
 
             def rsync_result(update: RSyncerUpdate):

src/murfey/client/rsync.py

@@ -13,6 +13,7 @@
 import subprocess
 import threading
 import time
+from datetime import datetime
 from enum import Enum
 from pathlib import Path
 from typing import Awaitable, Callable, List, NamedTuple
@@ -63,6 +64,7 @@ def __init__(
         remove_files: bool = False,
         required_substrings_for_removal: List[str] = [],
         notify: bool = True,
+        end_time: datetime | None = None,
     ):
         super().__init__()
         self._basepath = basepath_local.absolute()
@@ -76,6 +78,9 @@ def __init__(
         self._server_url = server_url
         self._notify = notify
         self._finalised = False
+        self._end_time = end_time
+
+        self._skipped_files: List[Path] = []
 
         # Set rsync destination
         if local:
@@ -214,6 +219,10 @@ def enqueue(self, file_path: Path):
             absolute_path = self._basepath / file_path
             self.queue.put(absolute_path)
 
+    def flush_skipped(self):
+        for f in self._skipped_files:
+            self.queue.put(f)
+
     def _process(self):
         logger.info("RSync thread starting")
         files_to_transfer: list[Path]
@@ -304,14 +313,23 @@ def _fake_transfer(self, files: list[Path]) -> bool:
 
         return True
 
-    def _transfer(self, files: list[Path]) -> bool:
+    def _transfer(self, infiles: list[Path]) -> bool:
         """
         Transfer files via an rsync sub-process, and parses the rsync stdout to verify
         the success of the transfer.
         """
 
         # Set up initial variables
-        files = [f for f in files if f.is_file()]
+        if self._end_time:
+            files = [
+                f
+                for f in infiles
+                if f.is_file() and f.stat().st_ctime < self._end_time.timestamp()
+            ]
+            self._skipped_files.extend(set(infiles).difference(set(files)))
+        else:
+            files = [f for f in infiles if f.is_file()]
+
         previously_transferred = self._files_transferred
         transfer_success: set[Path] = set()
         successful_updates: list[RSyncerUpdate] = []