Allow for overriding read/uri location for stream resource docs. (#940)

* Start working on adding directory_uri to PathInfo

* AD writers mostly modified to use directory uri

* Ruff and pre commit, modify panda writer to use directory_uri

* Fix failing test on linux and windows, apply suggestions from review

* Add test for new directory uri override option, ensure path separator always added to filepath in AD writers

* Don't use os.sep, since separator must be defined by IOC host

* Fix test failures on windows

* Fix failing test on windows

* Append separator if it is missing

* Use normalized path for error message

* Make sure directory path is always absolute for AD writers

* Fix failing windows test

* Use PurePath, and ensure path must be absolute in PathInfo

* Ruff

* Minor fixes

* Add tests for directory path override

* Add test for different write/read path semantics to detector tests

* Linter fixes

* Fix typing errors

* Use same type of path as base path in ymd path provider. Use isinstance for checking path type in ad core writer

Author: jwlodek

src/ophyd_async/core/_hdf_dataset.py

@@ -1,6 +1,4 @@
 from collections.abc import Iterator
-from pathlib import Path
-from urllib.parse import urlunparse
 
 from bluesky.protocols import StreamAsset
 from event_model import (  # type: ignore
@@ -48,22 +46,11 @@ class HDFDocumentComposer:
 
     def __init__(
         self,
-        full_file_name: Path,
+        file_uri: str,
         datasets: list[HDFDatasetDescription],
-        hostname: str = "localhost",
     ) -> None:
         self._last_emitted = 0
-        self._hostname = hostname
-        uri = urlunparse(
-            (
-                "file",
-                self._hostname,
-                str(full_file_name.absolute()),
-                "",
-                "",
-                None,
-            )
-        )
+        uri = file_uri
         bundler_composer = ComposeStreamResource()
         self._bundles: list[ComposeStreamResourceBundle] = [
             bundler_composer(

src/ophyd_async/core/_providers.py

@@ -1,11 +1,11 @@
-import os
 import uuid
 from abc import abstractmethod
 from collections.abc import Callable
 from dataclasses import dataclass
 from datetime import date
-from pathlib import Path
+from pathlib import PurePath, PureWindowsPath
 from typing import Protocol
+from urllib.parse import urlunparse
 
 
 @dataclass
@@ -16,11 +16,36 @@ class PathInfo:
     :param filename: Base filename to use generated by FilenameProvider, w/o extension
     :param create_dir_depth: Optional depth of directories to create if they do not
                              exist
+    :param directory_uri: Optional URI to use for reading back resources. If not set,
+                          it will be generated from the directory path.
     """
 
-    directory_path: Path
+    directory_path: PurePath
     filename: str
     create_dir_depth: int = 0
+    directory_uri: str | None = None
+
+    def __post_init__(self):
+        if not self.directory_path.is_absolute():
+            raise ValueError(
+                f"directory_path must be an absolute path, got {self.directory_path}"
+            )
+
+        # If directory uri is not set, set it using the directory path.
+        if self.directory_uri is None:
+            self.directory_uri = urlunparse(
+                (
+                    "file",
+                    "localhost",
+                    f"{self.directory_path.as_posix()}/",
+                    "",
+                    "",
+                    None,
+                )
+            )
+        elif not self.directory_uri.endswith("/"):
+            # Ensure the directory URI ends with a slash.
+            self.directory_uri += "/"
 
 
 class FilenameProvider(Protocol):
@@ -112,18 +137,21 @@ class StaticPathProvider(PathProvider):
     def __init__(
         self,
         filename_provider: FilenameProvider,
-        directory_path: Path | str,
+        directory_path: PurePath,
+        directory_uri: str | None = None,
         create_dir_depth: int = 0,
     ) -> None:
         self._filename_provider = filename_provider
-        self._directory_path = Path(directory_path)
+        self._directory_path = directory_path
+        self._directory_uri = directory_uri
         self._create_dir_depth = create_dir_depth
 
     def __call__(self, device_name: str | None = None) -> PathInfo:
         filename = self._filename_provider(device_name)
 
         return PathInfo(
             directory_path=self._directory_path,
+            directory_uri=self._directory_uri,
             filename=filename,
             create_dir_depth=self._create_dir_depth,
         )
@@ -135,7 +163,8 @@ class AutoIncrementingPathProvider(PathProvider):
     def __init__(
         self,
         filename_provider: FilenameProvider,
-        base_directory_path: Path,
+        base_directory_path: PurePath,
+        base_directory_uri: str | None = None,
         create_dir_depth: int = 0,
         max_digits: int = 5,
         starting_value: int = 0,
@@ -146,6 +175,12 @@ def __init__(
     ) -> None:
         self._filename_provider = filename_provider
         self._base_directory_path = base_directory_path
+        self._base_directory_uri = base_directory_uri
+        if (
+            self._base_directory_uri is not None
+            and not self._base_directory_uri.endswith("/")
+        ):
+            self._base_directory_uri += "/"
         self._create_dir_depth = create_dir_depth
         self._base_name = base_name
         self._starting_value = starting_value
@@ -174,8 +209,13 @@ def __call__(self, device_name: str | None = None) -> PathInfo:
             self._inc_counter = 0
             self._current_value += self._increment
 
+        directory_uri = None
+        if self._base_directory_uri is not None:
+            directory_uri = f"{self._base_directory_uri}{auto_inc_dir_name}"
+
         return PathInfo(
             directory_path=self._base_directory_path / auto_inc_dir_name,
+            directory_uri=directory_uri,
             filename=filename,
             create_dir_depth=self._create_dir_depth,
         )
@@ -187,34 +227,52 @@ class YMDPathProvider(PathProvider):
     def __init__(
         self,
         filename_provider: FilenameProvider,
-        base_directory_path: Path,
+        base_directory_path: PurePath,
+        base_directory_uri: str | None = None,
         create_dir_depth: int = -3,  # Default to -3 to create YMD dirs
         device_name_as_base_dir: bool = False,
     ) -> None:
         self._filename_provider = filename_provider
-        self._base_directory_path = Path(base_directory_path)
+        self._base_directory_path = base_directory_path
+        self._base_directory_uri = base_directory_uri
+        if (
+            self._base_directory_uri is not None
+            and not self._base_directory_uri.endswith("/")
+        ):
+            self._base_directory_uri += "/"
         self._create_dir_depth = create_dir_depth
         self._device_name_as_base_dir = device_name_as_base_dir
 
     def __call__(self, device_name: str | None = None) -> PathInfo:
-        sep = os.path.sep
+        path_type = type(self._base_directory_path)
+        if path_type == PureWindowsPath:
+            sep = "\\"
+        else:
+            sep = "/"
+
         current_date = date.today().strftime(f"%Y{sep}%m{sep}%d")
         if device_name is None:
             ymd_dir_path = current_date
         elif self._device_name_as_base_dir:
-            ymd_dir_path = os.path.join(
+            ymd_dir_path = path_type(
                 current_date,
                 device_name,
             )
         else:
-            ymd_dir_path = os.path.join(
+            ymd_dir_path = path_type(
                 device_name,
                 current_date,
             )
 
         filename = self._filename_provider(device_name)
+
+        directory_uri = None
+        if self._base_directory_uri is not None:
+            directory_uri = f"{self._base_directory_uri}{ymd_dir_path}"
+
         return PathInfo(
             directory_path=self._base_directory_path / ymd_dir_path,
+            directory_uri=directory_uri,
             filename=filename,
             create_dir_depth=self._create_dir_depth,
         )

src/ophyd_async/epics/adcore/_core_writer.py

@@ -1,8 +1,7 @@
 import asyncio
 from collections.abc import AsyncGenerator, AsyncIterator
-from pathlib import Path
+from pathlib import PureWindowsPath
 from typing import Generic, TypeVar, get_args
-from urllib.parse import urlunparse
 
 from bluesky.protocols import Hints, StreamAsset
 from event_model import (  # type: ignore
@@ -13,14 +12,14 @@
 from pydantic import PositiveInt
 
 from ophyd_async.core._detector import DetectorWriter
-from ophyd_async.core._providers import DatasetDescriber, PathProvider
+from ophyd_async.core._providers import DatasetDescriber, PathInfo, PathProvider
 from ophyd_async.core._signal import (
     observe_value,
     set_and_wait_for_value,
     wait_for_value,
 )
 from ophyd_async.core._status import AsyncStatus
-from ophyd_async.core._utils import DEFAULT_TIMEOUT
+from ophyd_async.core._utils import DEFAULT_TIMEOUT, error_if_none
 
 # from ophyd_async.epics.adcore._core_logic import ADBaseDatasetDescriber
 from ._core_io import (
@@ -53,7 +52,8 @@ def __init__(
     ) -> None:
         self._plugins = plugins or {}
         self.fileio = fileio
-        self._path_provider = path_provider
+        self._path_provider: PathProvider = path_provider
+        self._path_info: PathInfo | None = None
         self._dataset_describer = dataset_describer
         self._file_extension = file_extension
         self._mimetype = mimetype
@@ -83,20 +83,32 @@ def with_io(
         writer = cls(fileio, path_provider, dataset_describer, plugins=plugins)
         return writer
 
-    async def begin_capture(self, name: str) -> None:
-        info = self._path_provider(device_name=name)
+    async def _begin_capture(self, name: str) -> None:
+        path_info = error_if_none(
+            self._path_info, "Writer must be opened before beginning capture!"
+        )
 
         if isinstance(self.fileio, NDFilePluginIO):
             await self.fileio.enable_callbacks.set(ADCallbacks.ENABLE)
 
         # Set the directory creation depth first, since dir creation callback happens
         # when directory path PV is processed.
-        await self.fileio.create_directory.set(info.create_dir_depth)
+        await self.fileio.create_directory.set(path_info.create_dir_depth)
+
+        # Need to ensure that trailing separator is added to the directory path.
+        # When setting the path for windows based AD IOCs, a '/' is added rather than
+        # a '\\', which will cause the readback to never register the same value.
+        dir_path_as_str = str(path_info.directory_path)
+        separator = "/"
+        if isinstance(path_info.directory_path, PureWindowsPath):
+            separator = "\\"
+
+        dir_path_as_str += separator
 
         await asyncio.gather(
             # See https://github.com/bluesky/ophyd-async/issues/122
-            self.fileio.file_path.set(str(info.directory_path)),
-            self.fileio.file_name.set(info.filename),
+            self.fileio.file_path.set(dir_path_as_str),
+            self.fileio.file_name.set(path_info.filename),
             self.fileio.file_write_mode.set(ADFileWriteMode.STREAM),
             # For non-HDF file writers, use AD file templating mechanism
             # for generating multi-image datasets
@@ -108,7 +120,7 @@ async def begin_capture(self, name: str) -> None:
         )
 
         if not await self.fileio.file_path_exists.get_value():
-            msg = f"File path {info.directory_path} for file plugin does not exist"
+            msg = f"Path {dir_path_as_str} doesn't exist or not writable!"
             raise FileNotFoundError(msg)
 
         # Overwrite num_capture to go forever
@@ -127,7 +139,9 @@ async def open(
         frame_shape = await self._dataset_describer.shape()
         dtype_numpy = await self._dataset_describer.np_datatype()
 
-        await self.begin_capture(name)
+        self._path_info = self._path_provider(device_name=name)
+
+        await self._begin_capture(name)
 
         describe = {
             name: DataKey(
@@ -154,30 +168,22 @@ async def get_indices_written(self) -> int:
     async def collect_stream_docs(
         self, name: str, indices_written: int
     ) -> AsyncIterator[StreamAsset]:
+        path_info = error_if_none(
+            self._path_info, "Writer must be opened before collecting stream docs!"
+        )
+
         if indices_written:
             if not self._emitted_resource:
-                file_path = Path(await self.fileio.file_path.get_value())
                 file_name = await self.fileio.file_name.get_value()
                 file_template = file_name + "_{:06d}" + self._file_extension
 
                 frame_shape = await self._dataset_describer.shape()
 
-                uri = urlunparse(
-                    (
-                        "file",
-                        "localhost",
-                        str(file_path.absolute()) + "/",
-                        "",
-                        "",
-                        None,
-                    )
-                )
-
                 bundler_composer = ComposeStreamResource()
 
                 self._emitted_resource = bundler_composer(
                     mimetype=self._mimetype,
-                    uri=uri,
+                    uri=str(path_info.directory_uri),
                     # TODO no reference to detector's name
                     data_key=name,
                     parameters={

src/ophyd_async/epics/adcore/_hdf_writer.py

@@ -1,6 +1,5 @@
 import asyncio
 from collections.abc import AsyncIterator
-from pathlib import Path
 from typing import TypeGuard
 from xml.etree import ElementTree as ET
 
@@ -65,8 +64,10 @@ async def open(
             self.fileio.xml_file_name.set(""),
         )
 
+        self._path_info = self._path_provider(device_name=name)
+
         # Set common AD file plugin params, begin capturing
-        await self.begin_capture(name)
+        await self._begin_capture(name)
 
         detector_shape = await self._dataset_describer.shape()
         np_dtype = await self._dataset_describer.np_datatype()
@@ -100,7 +101,7 @@ async def open(
 
         self._composer = HDFDocumentComposer(
             # See https://github.com/bluesky/ophyd-async/issues/122
-            Path(await self.fileio.full_file_name.get_value()),
+            f"{self._path_info.directory_uri}{self._path_info.filename}{self._file_extension}",
             self._datasets,
         )
 

src/ophyd_async/fastcs/panda/_writer.py

@@ -1,6 +1,5 @@
 import asyncio
 from collections.abc import AsyncGenerator, AsyncIterator
-from pathlib import Path
 
 from bluesky.protocols import StreamAsset
 from event_model import DataKey
@@ -67,8 +66,7 @@ async def open(self, name: str, exposures_per_event: int = 1) -> dict[str, DataK
         describe = await self._describe(name)
 
         self._composer = HDFDocumentComposer(
-            Path(await self.panda_data_block.hdf_directory.get_value())
-            / Path(await self.panda_data_block.hdf_file_name.get_value()),
+            f"{info.directory_uri}{info.filename}.h5",
             self._datasets,
         )
 

src/ophyd_async/sim/__main__.py

@@ -1,6 +1,7 @@
 """Used for tutorial `Using Devices`."""
 
 # Import bluesky and ophyd
+from pathlib import PurePath
 from tempfile import mkdtemp
 
 import bluesky.plan_stubs as bps  # noqa: F401
@@ -27,7 +28,7 @@
 
 # Make a path provider that makes UUID filenames within a static
 # temporary directory
-path_provider = StaticPathProvider(UUIDFilenameProvider(), mkdtemp())
+path_provider = StaticPathProvider(UUIDFilenameProvider(), PurePath(mkdtemp()))
 
 # All Devices created within this block will be
 # connected and named at the end of the with block