Fix type errors caught by pyright but not mypy

README.rst

@@ -79,8 +79,11 @@ it increments.
     class CounterDevice(Device):
         """A simple device which increments a value."""
 
-        Inputs: TypedDict = TypedDict("Inputs", {})
-        Outputs: TypedDict = TypedDict("Outputs", {"value":int})
+        class Inputs(TypedDict):
+            ...
+
+        class Outputs(TypedDict):
+            value: int
 
         def __init__(self, initial_value: int = 0, callback_period: int = int(1e9)) -> None:
             self._value = initial_value

docs/user/explanations/devices.rst

@@ -18,9 +18,11 @@ values and requests to be called back for an update sometime later.
     """A trivial toy device which produced a random output and requests a callback."""
 
         #: An empty typed mapping of device inputs
-        Inputs: TypedDict = TypedDict("Inputs", {})
+        class Inputs(TypedDict):
+            ...
         #: A typed mapping containing the 'output' output value
-        Outputs: TypedDict = TypedDict("Outputs", {"output": int})
+        class Outputs(TypedDict):
+            output: int
 
         def __init__(self, callback_period: int = int(1e9)) -> None:
             self.callback_period = SimTime(callback_period)
@@ -45,4 +47,4 @@ values and requests to be called back for an update sometime later.
 
 Logic can be implemented into the device via device methods. For an example of
 this look at the ``ShutterDevice``. It acts to attenuate the flux of any incoming
-value.
+value.

docs/user/how-to/use-epics-adapter.rst

@@ -55,8 +55,10 @@ outputs a current.
     class FemtoDevice(Device):
         """Electronic signal amplifier."""
 
-        Inputs: TypedDict = TypedDict("Inputs", {"input": float})
-        Outputs: TypedDict = TypedDict("Outputs", {"current": float})
+        class Inputs(TypedDict):
+            input: float
+        class Outputs(TypedDict):
+            current: float
 
         def __init__(
             self,

docs/user/tutorials/create-a-device.rst

@@ -32,8 +32,10 @@ maps as members. As such we shall put in the following boilerplate.
 
     class AmplifierDevice(Device):
 
-    Inputs: TypedDict = TypedDict("Inputs", {})
-    Outputs: TypedDict = TypedDict("Outputs", {})
+    class Inputs(TypedDict):
+        ...
+    class Outputs(TypedDict):
+        ...
 
     def __init__(self) -> None:
 
@@ -59,8 +61,10 @@ here.
 
     class AmplifierDevice(Device):
 
-    Inputs: TypedDict = TypedDict("Inputs", {})
-    Outputs: TypedDict = TypedDict("Outputs", {})
+    class Inputs(TypedDict):
+        ...
+    class Outputs(TypedDict):
+        ...
 
     def __init__(self, initial_amplification: float = 2) -> None:
             self.amplification = initial_amplification
@@ -95,8 +99,10 @@ we define our inputs and outputs in the maps, and the line of logic in the ``upd
 
     class AmplifierDevice(Device):
 
-        Inputs: TypedDict = TypedDict("Inputs", {"initial_signal":float})
-        Outputs: TypedDict = TypedDict("Outputs", {"amplified_signal":float})
+        class Inputs(TypedDict):
+            initial_signal: float
+        class Outputs(TypedDict):
+            amplified_signal: float
 
         def __init__(self, initial_amplification: float = 2.0) -> None:
             self.amplification = initial_amplification

examples/devices/amplifier.py

@@ -15,8 +15,11 @@
 class AmplifierDevice(Device):
     """Amplifier device which multiplies an input signal by an amplification value."""
 
-    Inputs: TypedDict = TypedDict("Inputs", {"initial_signal": float})
-    Outputs: TypedDict = TypedDict("Outputs", {"amplified_signal": float})
+    class Inputs(TypedDict):
+        initial_signal: float
+
+    class Outputs(TypedDict):
+        amplified_signal: float
 
     def __init__(self, initial_amplification: float = 2) -> None:
         """Amplifier constructor which configures the initial amplification.

examples/devices/counter.py

@@ -23,9 +23,12 @@ class CounterDevice(Device):
     """A simple device which increments a value."""
 
     #: An empty typed mapping of input values
-    Inputs: TypedDict = TypedDict("Inputs", {})
+    class Inputs(TypedDict):
+        ...
+
     #: A typed mapping containing the 'value' output value
-    Outputs: TypedDict = TypedDict("Outputs", {"value": int})
+    class Outputs(TypedDict):
+        value: int
 
     def __init__(self, initial_value: int = 0, callback_period: int = int(1e9)) -> None:
         """A constructor of the counter, which increments the input value.

examples/devices/isolated_device.py

@@ -21,8 +21,11 @@ class IsolatedBoxDevice(Device):
     The device has no inputs or outputs and interacts solely through adapters.
     """
 
-    Inputs: TypedDict = TypedDict("Inputs", {})
-    Outputs: TypedDict = TypedDict("Outputs", {})
+    class Inputs(TypedDict):
+        ...
+
+    class Outputs(TypedDict):
+        ...
 
     def __init__(self, initial_value: float = 2) -> None:
         """Constructor which configures the initial value

examples/devices/remote_controlled.py

@@ -22,9 +22,12 @@ class RemoteControlledDevice(Device):
     """A trivial toy device which is controlled by an adapter."""
 
     #: An empty typed mapping of device inputs
-    Inputs: TypedDict = TypedDict("Inputs", {})
+    class Inputs(TypedDict):
+        ...
+
     #: A typed mapping containing the 'observed' output value
-    Outputs: TypedDict = TypedDict("Outputs", {"observed": float})
+    class Outputs(TypedDict):
+        observed: float
 
     def __init__(
         self,

examples/devices/shutter.py

@@ -24,9 +24,12 @@ class ShutterDevice(Device):
     """
 
     #: A typed mapping containing the 'flux' input value
-    Inputs: TypedDict = TypedDict("Inputs", {"flux": float})
+    class Inputs(TypedDict):
+        flux: float
+
     #: A typed mapping containing the 'flux' output value
-    Outputs: TypedDict = TypedDict("Outputs", {"flux": float})
+    class Outputs(TypedDict):
+        flux: float
 
     def __init__(
         self, default_position: float, initial_position: Optional[float] = None

examples/devices/trampoline.py

@@ -16,9 +16,12 @@ class TrampolineDevice(Device):
     """A trivial toy device which requests a callback every update."""
 
     #: An empty typed mapping of device inputs
-    Inputs: TypedDict = TypedDict("Inputs", {})
+    class Inputs(TypedDict):
+        ...
+
     #: An empty typed mapping of device outputs
-    Outputs: TypedDict = TypedDict("Outputs", {})
+    class Outputs(TypedDict):
+        ...
 
     def __init__(self, callback_period: int = int(1e9)) -> None:
         """A constructor of the sink which configures the device callback period.
@@ -55,9 +58,12 @@ class RandomTrampolineDevice(Device):
     """A trivial toy device which produced a random output and requests a callback."""
 
     #: An empty typed mapping of device inputs
-    Inputs: TypedDict = TypedDict("Inputs", {})
+    class Inputs(TypedDict):
+        ...
+
     #: A typed mapping containing the 'output' output value
-    Outputs: TypedDict = TypedDict("Outputs", {"output": int})
+    class Outputs(TypedDict):
+        output: int
 
     def __init__(self, callback_period: int = int(1e9)) -> None:
         """A constructor of the sink which configures the device callback period.

examples/devices/zeromq_push_device.py

@@ -23,7 +23,7 @@ def __init__(
         self,
         host: str = "127.0.0.1",
         port: int = 5555,
-        socket_factory: Optional[SocketFactory] = create_zmq_push_socket,
+        socket_factory: SocketFactory = create_zmq_push_socket,
         addresses_to_publish: Optional[Set[str]] = None,
     ) -> None:
         super().__init__(host, port, socket_factory)
@@ -32,7 +32,7 @@ def __init__(
     def after_update(self):
         for address in self._addresses_to_publish:
             value = self.device.read(address)
-            self.send_message([{address: value}])
+            _ = self.send_message([{address: value}])
 
 
 @pydantic.v1.dataclasses.dataclass

src/tickit/adapters/epicsadapter/adapter.py

@@ -73,14 +73,15 @@ def on_db_load(self) -> None:
 
     def load_records_without_DTYP_fields(self):
         """Load records from database file without DTYP fields."""
-        with open(self.db_file, "rb") as inp:
-            with NamedTemporaryFile(suffix=".db", delete=False) as out:
-                for line in inp.readlines():
-                    if not re.match(rb"\s*field\s*\(\s*DTYP", line):
-                        out.write(line)
-
-        softioc.dbLoadDatabase(out.name, substitutions=f"device={self.ioc_name}")
-        os.unlink(out.name)
+        if self.db_file:
+            with open(self.db_file, "rb") as inp:
+                with NamedTemporaryFile(suffix=".db", delete=False) as out:
+                    for line in inp.readlines():
+                        if not re.match(rb"\s*field\s*\(\s*DTYP", line):
+                            out.write(line)
+
+            softioc.dbLoadDatabase(out.name, substitutions=f"device={self.ioc_name}")
+            os.unlink(out.name)
 
     async def run_forever(
         self, device: Device, raise_interrupt: RaiseInterrupt

src/tickit/adapters/epicsadapter/ioc_manager.py

@@ -63,5 +63,5 @@ def _build_and_run_ioc() -> None:
     # dbl directly prints out all record names, so we have to check
     # the log level in order to only do it in DEBUG.
     if LOGGER.level <= logging.DEBUG:
-        softioc.dbl()
+        softioc.dbl()  # type: ignore
     LOGGER.debug("IOC started")

src/tickit/adapters/interpreters/command/command_interpreter.py

@@ -2,7 +2,7 @@
 from inspect import getmembers
 from typing import (
     AnyStr,
-    AsyncIterable,
+    AsyncIterator,
     Optional,
     Protocol,
     Sequence,
@@ -11,7 +11,7 @@
     runtime_checkable,
 )
 
-from tickit.adapters.interpreters.utils import wrap_as_async_iterable
+from tickit.adapters.interpreters.utils import wrap_as_async_iterator
 from tickit.core.adapter import Adapter, Interpreter
 
 
@@ -48,35 +48,24 @@ class CommandInterpreter(Interpreter[AnyStr]):
     called with the parsed arguments.
     """
 
-    @staticmethod
-    async def unknown_command() -> AsyncIterable[bytes]:
-        """An asynchronous iterable of containing a single unknown command reply.
-
-        Returns:
-            AsyncIterable[bytes]:
-                An asynchronous iterable of containing a single unknown command reply:
-                "Request does not match any known command".
-        """
-        yield b"Request does not match any known command"
-
     async def handle(
         self, adapter: Adapter, message: AnyStr
-    ) -> Tuple[AsyncIterable[AnyStr], bool]:
+    ) -> Tuple[AsyncIterator[AnyStr], bool]:
         """Matches the message to an adapter command and calls the corresponding method.
 
         An asynchronous method which handles a message by attempting to match the
         message against each of the registered commands, if a match is found the
         corresponding command is called and its reply is returned with an asynchronous
-        iterable wrapper if required. If no match is found the unknown command message
+        iterator wrapper if required. If no match is found the unknown command message
         is returned with no request for interrupt.
 
         Args:
             adapter (Adapter): The adapter in which the function should be executed
             message (bytes): The message to be handled.
 
         Returns:
-            Tuple[AsyncIterable[Union[str, bytes]], bool]:
-                A tuple of the asynchronous iterable of reply messages and a flag
+            Tuple[AsyncIterator[Union[str, bytes]], bool]:
+                A tuple of the asynchronous iterator of reply messages and a flag
                 indicating whether an interrupt should be raised by the adapter.
         """
         for _, method in getmembers(adapter):
@@ -91,8 +80,16 @@ async def handle(
                 for arg, argtype in zip(args, get_type_hints(method).values())
             )
             resp = await method(*args)
-            if not isinstance(resp, AsyncIterable):
-                resp = wrap_as_async_iterable(resp)
+            if not isinstance(resp, AsyncIterator):
+                resp = wrap_as_async_iterator(resp)
             return resp, command.interrupt
-        resp = CommandInterpreter.unknown_command()
-        return resp, False
+
+        msg = "Request does not match any known command"
+        # This is a pain but the message needs to match the input message
+        # TODO: Fix command interpreters' handling of bytes vs str
+        if isinstance(message, bytes):
+            resp = wrap_as_async_iterator(msg.encode("utf-8"))
+            return resp, False
+        else:
+            resp = wrap_as_async_iterator(msg)
+            return resp, False

src/tickit/adapters/interpreters/command/regex_command.py

@@ -1,9 +1,9 @@
-import re
-from dataclasses import dataclass
+from dataclasses import InitVar, dataclass, field
+from re import Pattern, compile
 from typing import AnyStr, Callable, Generic, Optional, Sequence
 
 
-@dataclass(frozen=True)
+@dataclass
 class RegexCommand(Generic[AnyStr]):
     """A decorator to register an adapter method as a regex parsed command.
 
@@ -20,9 +20,25 @@ class RegexCommand(Generic[AnyStr]):
             A decorator which registers the adapter method as a message handler.
     """
 
-    regex: AnyStr
+    regex: InitVar[AnyStr]
     interrupt: bool = False
-    format: Optional[str] = None
+    format: InitVar[Optional[str]] = None
+    pattern: Pattern[AnyStr] = field(init=False)
+    convert: Callable[[bytes], AnyStr] = field(init=False)
+
+    def __post_init__(self, regex: AnyStr, format: Optional[str]):
+        # The type checking fails here as it can't determine that the return type matches the
+        # regex type. The isinstance should narrow the AnyStr type but doesn't
+        if isinstance(regex, str):
+            if format is None:
+                raise ValueError(
+                    "If regex is a string, format is required to decode input"
+                )
+            self.convert = lambda b: b.decode(format)  # type: ignore
+        else:
+            self.convert = lambda b: b  # type: ignore
+
+        self.pattern = compile(regex)
 
     def __call__(self, func: Callable) -> Callable:
         """A decorator which registers the adapter method as a message handler.
@@ -51,9 +67,8 @@ def parse(self, data: bytes) -> Optional[Sequence[AnyStr]]:
                 If a full match is found a sequence of function arguments is returned,
                 otherwise the method returns None.
         """
-        message = data.decode(self.format, "ignore").strip() if self.format else data
-        if isinstance(message, type(self.regex)):
-            match = re.fullmatch(self.regex, message)
-            if match:
-                return match.groups()
+        message = self.convert(data)
+        match = self.pattern.fullmatch(message)
+        if match:
+            return match.groups()
         return None