Fix panda hdf writing isort and import warnings

linting
bluesky · Dec 8, 2023 · f3bd571 · f3bd571
1 parent b4b5456
commit f3bd571
Show file tree

Hide file tree

Showing 3 changed files with 21 additions and 32 deletions.
diff --git a/src/ophyd_async/panda/writers/hdf_writer.py b/src/ophyd_async/panda/writers/hdf_writer.py
@@ -1,18 +1,18 @@
 import asyncio
+import atexit
+from enum import Enum
 from typing import (
     AsyncIterator,
+    Callable,
     Dict,
+    FrozenSet,
     List,
     Optional,
-    FrozenSet,
-    Callable,
+    Tuple,
     get_type_hints,
 )
-import atexit
-from enum import Enum
 
 from bluesky.protocols import Asset, Descriptor, Hints
-
 from p4p.client.thread import Context
 
 from ophyd_async.core import (
@@ -25,13 +25,12 @@
     Signal,
     wait_for_value,
 )
-
 from ophyd_async.epics.signal import (
-    pvi_get,
     epics_signal_r,
     epics_signal_rw,
     epics_signal_w,
     epics_signal_x,
+    pvi_get,
 )
 
 from .panda_hdf import DataBlock, _HDFDataset, _HDFFile
@@ -76,13 +75,14 @@ async def connect(self, sim=False) -> None:
         pvi_info = await pvi_get(self._prefix + ":PVI", self.ctxt) if not sim else {}
 
         # signals to connect, giving block name, signal name and datatype
-        desired_signals = {}
+        desired_signals: Dict[str, List[Tuple[str, type]]] = {}
         for block_name, block in self._to_capture.items():
             if block_name not in desired_signals:
                 desired_signals[block_name] = []
             for signal_name in block:
                 desired_signals[block_name].append(
-                    [f"{signal_name}_capture", SimpleCapture]
+
+                    (f"{signal_name}_capture", SimpleCapture)
                 )
         # add signals from DataBlock using type hints
         if "hdf5" not in desired_signals:
@@ -160,7 +160,7 @@ async def open(self, multiplier: int = 1) -> Dict[str, Descriptor]:
         # Wait for it to start, stashing the status that tells us when it finishes
         await self.hdf5.capture.set(True)
         self._capture_status = await wait_for_value(
-            self.hdf5.capturing, 1, DEFAULT_TIMEOUT
+            self.hdf5.capturing, True, DEFAULT_TIMEOUT
         )
         name = self._name_provider()
         if multiplier > 1:
@@ -199,7 +199,6 @@ def matcher(value: int) -> bool:
         await wait_for_value(self.hdf5.numwritten_rbv, matcher, timeout=timeout)
 
     async def get_indices_written(self) -> int:
-        value = await self.hdf5.numwritten_rbv.get_value()
         num_written = await self.hdf5.numwritten_rbv.get_value()
         return num_written // self._multiplier
 

diff --git a/src/ophyd_async/panda/writers/panda_hdf.py b/src/ophyd_async/panda/writers/panda_hdf.py
@@ -1,13 +1,10 @@
 from dataclasses import dataclass
-from typing import Iterator, List, Tuple
+from typing import Iterator, List
 
 from event_model import StreamDatum, StreamResource, compose_stream_resource
 
+from ophyd_async.core import SignalR, SignalRW
 from ophyd_async.core.device import Device
-from ophyd_async.core import (
-    SignalR,
-    SignalRW,
-)
 
 
 class DataBlock(Device):

diff --git a/tests/panda/test_writer.py b/tests/panda/test_writer.py
@@ -9,7 +9,7 @@
     StaticDirectoryProvider,
     set_and_wait_for_value,
 )
-from ophyd_async.epics.signal.signal import SignalR, SignalRW, epics_signal_rw
+from ophyd_async.epics.signal.signal import SignalRW
 from ophyd_async.panda.writers import PandaHDFWriter
 
 
@@ -18,27 +18,20 @@ async def sim_writer(tmp_path) -> PandaHDFWriter:
     dir_prov = StaticDirectoryProvider(str(tmp_path), "test")
     async with DeviceCollector(sim=True):
         writer = PandaHDFWriter("TEST-PANDA", dir_prov, lambda: "test-panda")
-        writer.hdf5.filepath = SignalRW(
-            SimSignalBackend(str, "TEST-PANDA:HDF5:FilePath")
+        writer.hdf5.filepath = SignalRW(backend=SimSignalBackend(str, source="TEST-PANDA:HDF5:FilePath")
         )
-        writer.hdf5.filename = SignalRW(
-            SimSignalBackend(str, "TEST-PANDA:HDF5:FileName")
+        writer.hdf5.filename = SignalRW(backend=SimSignalBackend(str, source="TEST-PANDA:HDF5:FileName")
         )
-        writer.hdf5.fullfilename = SignalRW(
-            SimSignalBackend(str, "TEST-PANDA:HDF5:FullFileName")
+        writer.hdf5.fullfilename = SignalRW(backend=SimSignalBackend(str, source="TEST-PANDA:HDF5:FullFileName")
         )
-        writer.hdf5.numcapture = SignalRW(
-            SimSignalBackend(int, "TEST-PANDA:HDF5:NumCapture")
+        writer.hdf5.numcapture = SignalRW(backend=SimSignalBackend(int, source="TEST-PANDA:HDF5:NumCapture")
         )
-        writer.hdf5.capture = SignalRW(SimSignalBackend(int, "TEST-PANDA:HDF5:Capture"))
-        writer.hdf5.capturing = SignalRW(
-            SimSignalBackend(int, "TEST-PANDA:HDF5:Capturing")
+        writer.hdf5.capture = SignalRW(backend=SimSignalBackend(int, source="TEST-PANDA:HDF5:Capture"))
+        writer.hdf5.capturing = SignalRW(backend=SimSignalBackend(int, source="TEST-PANDA:HDF5:Capturing")
         )
-        writer.hdf5.flushnow = SignalRW(
-            SimSignalBackend(int, "TEST-PANDA:HDF5:FlushNow")
+        writer.hdf5.flushnow = SignalRW(backend=SimSignalBackend(int, source="TEST-PANDA:HDF5:FlushNow")
         )
-        writer.hdf5.numwritten_rbv = SignalRW(
-            SimSignalBackend(int, "TEST-PANDA:HDF5:NumWritten_RBV")
+        writer.hdf5.numwritten_rbv = SignalRW(backend=SimSignalBackend(int, source="TEST-PANDA:HDF5:NumWritten_RBV")
         )
         await writer.connect(sim=True)
     return writer