Merge branch 'main' into 897-temperature-sims

CHANGES.rst

@@ -16,8 +16,8 @@
 Change History
 ##############
 
-The project `milestones <https://github.com/BCDA-APS/apstools/milestones>`_
-describe the future plans.
+Project `milestones <https://github.com/BCDA-APS/apstools/milestones>`_
+describe future plans.
 
 ..
    1.6.19
@@ -43,6 +43,11 @@ New Features
 * Add template support for writing NeXus/HDF5 files.
 * New lineup2() plan can be used in console, notebooks, and queueserver.
 
+Fixes
+-----------
+
+* Fix ``AD_full_file_name_local()`` for case when the read & write paths are identical.
+
 Maintenance
 -----------
 
@@ -52,7 +57,7 @@ Maintenance
 Known Problems
 --------------
 
-* Remove ScalerMotorFlyer, pending issue #763.
+* Remove ``ScalerMotorFlyer``, pending issue #763.
 
 1.6.17
 ******

apstools/devices/area_detector_support.py

@@ -49,6 +49,8 @@
 from ophyd.areadetector.plugins import TIFFPlugin_V34 as TIFFPlugin
 from packaging import version
 
+from ..utils import count_common_subdirs
+
 logger = logging.getLogger(__name__)
 
 # fmt: off
@@ -332,15 +334,9 @@ def AD_full_file_name_local(plugin):
     if plugin.read_path_template == plugin.write_path_template:
         return ffname
 
-    # identify the common last parts of the file directories
     read_parts = pathlib.Path(plugin.read_path_template).parts
     write_parts = pathlib.Path(plugin.write_path_template).parts
-    icommon = 0
-    for i in range(min(len(read_parts), len(write_parts))):
-        i1 = -i - 1
-        if read_parts[i1:] != write_parts[i1:]:
-            icommon = i
-            break
+    icommon = count_common_subdirs(plugin.read_path_template, plugin.write_path_template)
 
     # fmt: off
     if icommon == 0:

apstools/utils/__init__.py

@@ -29,6 +29,8 @@
 from .memory import rss_mem
 from .misc import cleanupText
 from .misc import connect_pvlist
+from .misc import count_child_devices_and_signals
+from .misc import count_common_subdirs
 from .misc import dictionary_table
 from .misc import full_dotted_name
 from .misc import itemizer

apstools/utils/misc.py

@@ -7,6 +7,7 @@
    ~cleanupText
    ~connect_pvlist
    ~count_child_devices_and_signals
+   ~count_common_subdirs
    ~dictionary_table
    ~full_dotted_name
    ~itemizer
@@ -25,6 +26,7 @@
 """
 
 import logging
+import pathlib
 import re
 import subprocess
 import sys
@@ -82,6 +84,22 @@ def count_child_devices_and_signals(device):
     return count
 
 
+def count_common_subdirs(p1, p2):
+    """Count how many subdirectories are common to both file paths."""
+    parts1 = pathlib.Path(p1).parts
+    parts2 = pathlib.Path(p2).parts
+    count = 0
+    for x, y in zip(reversed(parts1), reversed(parts2)):
+        if x != y:
+            break
+        count += 1
+    if count == 0 and min(len(parts1), len(parts2)) > 0:
+        # special case when first part of path is common
+        if parts1[0] == parts2[0]:
+            count = 1
+    return count
+
+
 def dictionary_table(dictionary, **kwargs):
     """
     Return a text table from ``dictionary``.

apstools/utils/tests/test_utils.py

@@ -515,3 +515,18 @@ def test_utils_getStreamValues_Exception(
         )
     assert str(exc.value).startswith(first_words)
 # fmt: on
+
+
+@pytest.mark.parametrize(
+    "p1, p2, expected",
+    [
+        ["/home/bl13user/images", "/home/bl13user/images", 4],
+        ["/home/bl13user/images", "/tmp/home/bl13user/images", 3],
+        ["/tmp/home/bl13user/images", "/home/bl13user/images", 3],
+        ["/a", "/home/bl13user/images", 1],
+        [r"C:\\", "/home/bl13user/images", 0],
+    ],
+)
+def test_count_common_subdirs(p1, p2, expected):
+    icommon = utils.count_common_subdirs(p1, p2)
+    assert icommon == expected, f"{p1=} {p2=}"