Fix panda hdf writing isort and import warnings

linting

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,20 +75,20 @@ 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:
             desired_signals["hdf5"] = []
         for signal_name, hint in get_type_hints(self.hdf5).items():
             dtype = hint.__args__[0]
-            desired_signals["hdf5"].append([signal_name, dtype])
+            desired_signals["hdf5"].append((signal_name, dtype))
         # loop over desired signals and set
         for block_name, block_signals in desired_signals.items():
             if block_name not in pvi_info:
@@ -116,7 +115,7 @@ async def connect(self, sim=False) -> None:
                 )
                 setattr(block, signal_name, signal)
         for block_name in desired_signals.keys():
-            block = getattr(self, block_name)
+            block: Device = getattr(self, block_name)
             if block:
                 await block.connect(sim=sim)
 
@@ -160,7 +159,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 +198,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
 

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):

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
 
 
@@ -19,26 +19,28 @@ async def sim_writer(tmp_path) -> PandaHDFWriter:
     async with DeviceCollector(sim=True):
         writer = PandaHDFWriter("TEST-PANDA", dir_prov, lambda: "test-panda")
         writer.hdf5.filepath = SignalRW(
-            SimSignalBackend(str, "TEST-PANDA:HDF5:FilePath")
+            backend=SimSignalBackend(str, source="TEST-PANDA:HDF5:FilePath")
         )
         writer.hdf5.filename = SignalRW(
-            SimSignalBackend(str, "TEST-PANDA:HDF5:FileName")
+            backend=SimSignalBackend(str, source="TEST-PANDA:HDF5:FileName")
         )
         writer.hdf5.fullfilename = SignalRW(
-            SimSignalBackend(str, "TEST-PANDA:HDF5:FullFileName")
+            backend=SimSignalBackend(str, source="TEST-PANDA:HDF5:FullFileName")
         )
         writer.hdf5.numcapture = SignalRW(
-            SimSignalBackend(int, "TEST-PANDA:HDF5:NumCapture")
+            backend=SimSignalBackend(int, source="TEST-PANDA:HDF5:NumCapture")
+        )
+        writer.hdf5.capture = SignalRW(
+            backend=SimSignalBackend(bool, source="TEST-PANDA:HDF5:Capture")
         )
-        writer.hdf5.capture = SignalRW(SimSignalBackend(int, "TEST-PANDA:HDF5:Capture"))
         writer.hdf5.capturing = SignalRW(
-            SimSignalBackend(int, "TEST-PANDA:HDF5:Capturing")
+            backend=SimSignalBackend(bool, source="TEST-PANDA:HDF5:Capturing")
         )
         writer.hdf5.flushnow = SignalRW(
-            SimSignalBackend(int, "TEST-PANDA:HDF5:FlushNow")
+            backend=SimSignalBackend(bool, source="TEST-PANDA:HDF5:FlushNow")
         )
         writer.hdf5.numwritten_rbv = SignalRW(
-            SimSignalBackend(int, "TEST-PANDA:HDF5:NumWritten_RBV")
+            backend=SimSignalBackend(int, source="TEST-PANDA:HDF5:NumWritten_RBV")
         )
         await writer.connect(sim=True)
     return writer