Add API endpoints to transfer OTF directory and include OTF files in workflow message

src/murfey/instrument_server/api.py

@@ -425,6 +425,91 @@ def get_possible_otf_dirs(
     return candidates
 
 
+class OTFDirectoryUploadInfo(BaseModel):
+    dir_path: Path  # Full client-side path to OTF directory
+    visit_path: str  # Path fragment from rsync module to visit directory
+    destination_dir: str = "setup"  # Folder in visit directory to save to
+
+
+@router.post("/instruments/{instrument_name}/sessions/{session_id}/upload_otf_dir")
+def upload_otf_dir(
+    instrument_name: str,
+    session_id: MurfeySessionID,
+    otf_dir_info: OTFDirectoryUploadInfo,
+):
+    # Sanitise incoming values
+    otf_dir_path = sanitise(str(otf_dir_info.dir_path))
+    visit_path = sanitise(otf_dir_info.visit_path)
+    destination_dir = sanitise(otf_dir_info.destination_dir)
+
+    # Load machine config and other needed properties
+    machine_config_url_path = url_path_for(
+        "session_control.router",
+        "machine_info_by_instrument",
+        instrument_name=sanitise_nonpath(instrument_name),
+    )
+    machine_config: dict[str, Any] = requests.get(
+        f"{_get_murfey_url()}{machine_config_url_path}",
+        headers={"Authorization": f"Bearer {tokens[session_id]}"},
+    ).json()
+
+    # Validate that file passed is from the gain reference directory
+    otf_dir = machine_config.get("gain_reference_directory", "")
+    if not otf_dir_path.startswith(otf_dir):
+        raise ValueError(
+            "Gain reference file does not originate from the gain reference directory "
+            f"{otf_dir!r}"
+        )
+
+    # Construct the rsync path to use
+    # Return the rsync URL if set, otherwise assume you are syncing via Murfey
+    rsync_url = urlparse(
+        str(machine_config["rsync_url"])
+        if machine_config.get("rsync_url", "")
+        else _get_murfey_url()
+    )
+    rsync_module = machine_config.get("rsync_module", "data")
+    rsync_path = f"{rsync_url.hostname}::{rsync_module}/{visit_path}/{destination_dir}"
+
+    # Construct the expected destination path on the server side
+    rsync_basepath: str | None = machine_config.get("rsync_basepath")
+    if not rsync_basepath:
+        logger.error(f"No rsync base path was configured for {instrument_name}")
+        return {"success": False}
+    destination_path = (
+        f"{rsync_basepath}/{visit_path}/{destination_dir}/"
+        f"{secure_filename(otf_dir_info.dir_path.name)}"
+    )
+
+    # Run rsync subprocess to transfer OTF directory and contents
+    cmd = [
+        "rsync",
+        "-a",  # Sync folder contents
+    ]
+    if rsync_chmod := machine_config.get("rsync_chmod"):
+        cmd.append(f"--chmod={rsync_chmod}")
+    cmd.extend([posix_path(Path(otf_dir_path)), rsync_path])
+    process = subprocess.run(
+        cmd,
+        capture_output=True,
+        text=True,
+    )
+    if process.returncode:
+        logger.warning(
+            f"Failed to transfer OTF directory {otf_dir_path!r} to {f'{destination_path}'!r} \n"
+            f"Executed the following command: {' '.join(cmd)!r} \n"
+            f"Returned the following error: \n"
+            f"{process.stderr}"
+        )
+        return {"success": False}
+
+    # Return the full path as part of the message
+    return {
+        "success": True,
+        "destination_path": str(destination_path),
+    }
+
+
 @router.get(
     "/instruments/{instrument_name}/sessions/{session_id}/possible_gain_references"
 )

src/murfey/server/api/instrument.py

@@ -2,7 +2,9 @@
 
 import asyncio
 import datetime
+import json
 import logging
+import os
 from pathlib import Path
 from typing import Annotated, Any, List, Optional
 from urllib.parse import quote
@@ -408,6 +410,89 @@ async def get_possible_otf_dirs(
     return data
 
 
+class OTFDirectoryUploadRequest(BaseModel):
+    dir_path: Path
+
+
+@router.post("/sessions/{session_id}/upload_otf_dir")
+async def request_otf_dir_upload(
+    session_id: MurfeySessionID,
+    otf_dir_request: OTFDirectoryUploadRequest,
+    db=murfey_db,
+):
+    # Load session information from database
+    murfey_session = db.exec(select(Session).where(Session.id == session_id)).one()
+    visit_name = murfey_session.visit
+    instrument_name = murfey_session.instrument_name
+
+    # Load machine config
+    machine_config = get_machine_config(instrument_name=instrument_name)[
+        instrument_name
+    ]
+
+    # Default data to return
+    data: dict[str, Any] = {"success": False}
+
+    # Load the rsync basepath
+    rsync_basepath = machine_config.rsync_basepath
+    if rsync_basepath is None:
+        log.error(f"No rsync basepath was configured for instrument {instrument_name}")
+        return data
+
+    # Construct the partial and full paths to the server-side visit directory
+    visit_path = f"{datetime.datetime.now().year}/{visit_name}"
+    visit_dir = rsync_basepath / visit_path
+    if not visit_dir.exists():  # Check previous year in case of rollover
+        visit_dir_prev = visit_dir
+        visit_path = f"{datetime.datetime.now().year - 1}/{visit_name}"
+        visit_dir = rsync_basepath / visit_path
+        if not visit_dir.exists():
+            log.error(
+                "Unable to find visit directory under "
+                f"{str(visit_dir_prev)} or {str(visit_dir)}"
+            )
+            return data
+
+    # Ensure that the OTF destination directory exists
+    otf_dir = visit_dir / machine_config.gain_directory_name
+    otf_dir.mkdir(exist_ok=True)
+    os.chmod(otf_dir, mode=machine_config.mkdir_chmod)  # Set permissions
+
+    if machine_config.instrument_server_url:
+        async with aiohttp.ClientSession() as clientsession:
+            url_path = url_path_for(
+                "api.router",
+                "upload_otf_dir",
+                instrument_name=instrument_name,
+                session_id=session_id,
+            )
+            payload = {
+                "dir_path": str(otf_dir_request.dir_path),
+                "visit_path": visit_path,
+                "destination_dir": machine_config.gain_directory_name,
+            }
+
+            # REMOVE AFTER TESTING
+            log.info(
+                "Submitting the following payload to instrument server: \n"
+                f"{json.dumps(payload, indent=2, default=str)}"
+            )
+
+            async with clientsession.post(
+                f"{machine_config.instrument_server_url}{url_path}",
+                json=payload,
+                headers={
+                    "Authorization": f"Bearer {instrument_server_tokens[session_id]['access_token']}"
+                },
+            ) as resp:
+                data = await resp.json()
+    else:
+        log.error(
+            f"No instrument server URL was configured for instrument {instrument_name}"
+        )
+    return data
+
+
 @router.get(
     "/instruments/{instrument_name}/sessions/{session_id}/possible_gain_references"
 )

src/murfey/util/route_manifest.yaml

@@ -139,6 +139,15 @@ murfey.instrument_server.api.router:
         type: int
     methods:
       - GET
+  - path: /instruments/{instrument_name}/sessions/{session_id}/upload_otf_dir
+    function: upload_otf_dir
+    path_params:
+      - name: instrument_name
+        type: str
+      - name: session_id
+        type: int
+    methods:
+      - POST
   - path: /instruments/{instrument_name}/sessions/{session_id}/possible_gain_references
     function: get_possible_gain_references
     path_params:
@@ -561,6 +570,13 @@ murfey.server.api.instrument.router:
         type: int
     methods:
       - GET
+  - path: /instrument_server/sessions/{session_id}/upload_otf_dir
+    function: request_otf_dir_upload
+    path_params:
+      - name: session_id
+        type: int
+    methods:
+      - POST
   - path: /instrument_server/instruments/{instrument_name}/sessions/{session_id}/possible_gain_references
     function: get_possible_gain_references
     path_params:

tests/instrument_server/test_api.py

@@ -170,6 +170,118 @@ def test_get_possible_otf_dirs(
     )
 
 
+@pytest.mark.parametrize(
+    "test_params",
+    (
+        # Rsync URL settings
+        ("http://1.1.1.1",),  # When rsync_url is provided
+        ("",),  # When rsync_url is blank
+        (None,),  # When rsync_url not provided
+    ),
+)
+def test_upload_otf_dir(
+    mocker: MockerFixture,
+    test_params: tuple[Optional[str]],
+    tmp_path: Path,
+):
+    # Unpack test parameters and define other ones
+    (rsync_url_setting,) = test_params
+    server_url = "https://murfey.server.test"
+    instrument_name = "murfey"
+    session_id = 1
+
+    # Mock out objects
+    mock_request = mocker.patch("murfey.instrument_server.api.requests")
+    mock_get_server_url = mocker.patch("murfey.instrument_server.api._get_murfey_url")
+    mock_subprocess = mocker.patch("murfey.instrument_server.api.subprocess")
+    mocker.patch("murfey.instrument_server.api.tokens", {session_id: ANY})
+
+    # Create a mock machine config base on the test params
+    rsync_module = "data"
+    rsync_basepath = tmp_path / "data"
+    otf_dir = "C:/ProgramData/SIM/OTFs"
+    mock_machine_config = {
+        "rsync_module": rsync_module,
+        "rsync_basepath": str(rsync_basepath),
+        "gain_reference_directory": otf_dir,
+    }
+    if rsync_url_setting is not None:
+        mock_machine_config["rsync_url"] = rsync_url_setting
+
+    # Assign expected values to the mock objects
+    mock_response = MagicMock()
+    mock_response.status_code = 200
+    mock_response.json.return_value = mock_machine_config
+    mock_request.get.return_value = mock_response
+    mock_get_server_url.return_value = server_url
+    mock_subprocess.run.return_value = MagicMock(returncode=0)
+
+    # Construct payload and pass request to function
+    otf_folder_name = "OTFs-123456"
+    otf_dir_path = f"{otf_dir}/{otf_folder_name}"
+    visit_path = "2025/aa00000-0"
+    destination_dir = "setup"
+    payload = {
+        "dir_path": otf_dir_path,
+        "visit_path": visit_path,
+        "destination_dir": destination_dir,
+    }
+
+    # Set up instrument server test client
+    client_server = set_up_test_client(session_id=session_id)
+
+    # Poke the endpoint with the expected data
+    url_path = url_path_for(
+        "api.router",
+        "upload_otf_dir",
+        instrument_name=instrument_name,
+        session_id=session_id,
+    )
+    response = client_server.post(url_path, json=payload)
+
+    # Check that the machine config request was called
+    machine_config_url = url_path_for(
+        "session_control.router",
+        "machine_info_by_instrument",
+        instrument_name=instrument_name,
+    )
+    mock_request.get.assert_called_once_with(
+        f"{server_url}{machine_config_url}",
+        headers={"Authorization": ANY},
+    )
+
+    # Check that the subprocess was run with the expected arguments
+    # If no rsync_url key is provided, or rsync_url key is empty,
+    # It should default to the server URL
+    expected_rsync_url = (
+        urlparse(server_url) if not rsync_url_setting else urlparse(rsync_url_setting)
+    )
+    expected_rsync_path = (
+        f"{expected_rsync_url.hostname}::{rsync_module}/{visit_path}/{destination_dir}"
+    )
+    expected_rsync_cmd = [
+        "rsync",
+        "-a",
+        posix_path(Path(otf_dir_path)),
+        expected_rsync_path,
+    ]
+    expected_destination_path = (
+        rsync_basepath / visit_path / destination_dir / otf_folder_name
+    )
+
+    mock_subprocess.run.assert_called_once_with(
+        expected_rsync_cmd,
+        capture_output=True,
+        text=True,
+    )
+
+    # Check that the function ran through to completion successfully
+    assert response.json() == {
+        "success": True,
+        "destination_path": str(expected_destination_path),
+    }
+
+
 @pytest.mark.parametrize(
     "test_params",
     (