Fix linter warnings

.github/pages/make_switcher.py

@@ -3,28 +3,27 @@
 from argparse import ArgumentParser
 from pathlib import Path
 from subprocess import CalledProcessError, check_output
-from typing import List, Optional
 
 
-def report_output(stdout: bytes, label: str) -> List[str]:
+def report_output(stdout: bytes, label: str) -> list[str]:
     ret = stdout.decode().strip().split("\n")
     print(f"{label}: {ret}")
     return ret
 
 
-def get_branch_contents(ref: str) -> List[str]:
+def get_branch_contents(ref: str) -> list[str]:
     """Get the list of directories in a branch."""
     stdout = check_output(["git", "ls-tree", "-d", "--name-only", ref])
     return report_output(stdout, "Branch contents")
 
 
-def get_sorted_tags_list() -> List[str]:
+def get_sorted_tags_list() -> list[str]:
     """Get a list of sorted tags in descending order from the repository."""
     stdout = check_output(["git", "tag", "-l", "--sort=-v:refname"])
     return report_output(stdout, "Tags list")
 
 
-def get_versions(ref: str, add: Optional[str]) -> List[str]:
+def get_versions(ref: str, add: str | None) -> list[str]:
     """Generate the file containing the list of all GitHub Pages builds."""
     # Get the directories (i.e. builds) from the GitHub Pages branch
     try:
@@ -41,7 +40,7 @@ def get_versions(ref: str, add: Optional[str]) -> List[str]:
     tags = get_sorted_tags_list()
 
     # Make the sorted versions list from main branches and tags
-    versions: List[str] = []
+    versions: list[str] = []
     for version in ["master", "main"] + tags:
         if version in builds:
             versions.append(version)

docs/conf.py

@@ -77,10 +77,10 @@
 
 # This means you can link things like `str` and `asyncio` to the relevant
 # docs in the python documentation.
-intersphinx_mapping = dict(python=("https://docs.python.org/3/", None))
+intersphinx_mapping = {"python": ("https://docs.python.org/3/", None)}
 
 # A dictionary of graphviz graph attributes for inheritance diagrams.
-inheritance_graph_attrs = dict(rankdir="TB")
+inheritance_graph_attrs = {"rankdir": "TB"}
 
 # Common links that should be available on every page
 rst_epilog = """
@@ -96,7 +96,7 @@
 html_theme = "sphinx_rtd_theme_github_versions"
 
 # Options for the sphinx rtd theme, use DLS blue
-html_theme_options = dict(style_nav_header_background="rgb(7, 43, 93)")
+html_theme_options = {"style_nav_header_background": "rgb(7, 43, 93)"}
 
 # Add any paths that contain custom static files (such as style sheets) here,
 # relative to this directory. They are copied after the builtin static files,

src/arc_hvbias/comms.py

@@ -3,7 +3,6 @@
 # import logging
 import socket
 import warnings
-from typing import Optional
 
 # Constants
 CR = "\r"
@@ -42,7 +41,7 @@ def clear_socket(self):
         while True:
             try:
                 self._socket.recv(RECV_BUFFER)
-            except socket.timeout:
+            except TimeoutError:
                 break
 
     @staticmethod
@@ -67,11 +66,11 @@ def _send(self, request: bytes):
             # print(f"Sending request:\n{self._format_message(request)}")
             self._socket.send(self._format_message(request))
         except BrokenPipeError:
-            warnings.warn("Pipe broken, make sure device is connected.")
+            warnings.warn("Pipe broken, make sure device is connected.", stacklevel=1)
         except BlockingIOError:
             print("IO blocked, make sure device is connected.")
 
-    async def _send_receive(self, request: bytes) -> Optional[bytes]:
+    async def _send_receive(self, request: bytes) -> bytes | None:
         """Sends a request and attempts to decode the response. Does not determine if
         the response indicates acknowledgement from the device.
 
@@ -93,13 +92,12 @@ async def _send_receive(self, request: bytes) -> Optional[bytes]:
                     return response
             # except UnicodeDecodeError as e:
             #     warnings.warn(f"{e}:\n{self._format_message(response).decode()}")
-            except socket.timeout:
-                warnings.warn("Didn't receive a response in time.")
+            except TimeoutError:
+                warnings.warn("Didn't receive a response in time.", stacklevel=1)
 
-        # self._log.debug(f"Received response:\n{self._format_message(decoded_response)}")
         return None
 
-    async def send_receive(self, request: bytes) -> Optional[bytes]:
+    async def send_receive(self, request: bytes) -> bytes | None:
         """Sends a request and attempts to decode the response.
 
         Args:

src/arc_hvbias/decorators.py

@@ -1,6 +1,7 @@
 import asyncio
 import warnings
-from typing import Any, Callable, Coroutine, cast
+from collections.abc import Callable, Coroutine
+from typing import Any, cast
 
 # from .ioc import Ioc
 
@@ -78,16 +79,16 @@ async def catch_exceptions(*args, **kwargs) -> None:
             await func(*args, **kwargs)
         except ValueError as e:
             # catch conversion errors when device returns and error string
-            warnings.warn(f"{e}, {self.k.last_recv}")
+            warnings.warn(f"{e}, {self.k.last_recv}", stacklevel=1)
             # cothread.Yield()
-        except AbortException as e:
+        except AbortException:
             # pass
             print("AbortException")
         # except asyncio.CancelledError as e:
         #     curr_t = asyncio.current_task()
         #     print(f"Cancel called on {curr_t}, restarting thread...")
         except Exception as e:
-            warnings.warn(f"{e}")
+            warnings.warn(f"{e}", stacklevel=1)
 
     return cast(_AsyncFuncType, catch_exceptions)
 

src/arc_hvbias/ioc.py

@@ -1,8 +1,9 @@
 import asyncio
 import math
 import threading
+from collections.abc import Coroutine
 from datetime import datetime
-from typing import Any, Coroutine, List, Optional
+from typing import Any
 
 # Import the basic framework components.
 from softioc import builder
@@ -114,8 +115,8 @@ def __init__(
         self.max_time = builder.longOut("MAX-TIME", initial_value=900)
 
         # other state variables
-        self.at_voltage_time: Optional[datetime] = None
-        self.cycle_stop_time: Optional[datetime] = None
+        self.at_voltage_time: datetime | None = None
+        self.cycle_stop_time: datetime | None = None
         self.last_transition = datetime.now()
         # self.pause_flag = False
         self.cycle_flag: bool = False
@@ -133,12 +134,12 @@ def __init__(
         self.run_cycle_control.clear()
         self.run_check_cycle.clear()
 
-        self._cycle_thread: Optional[threading.Thread] = None
-        self._cycle_thread_task: Optional[asyncio.Task] = None
+        self._cycle_thread: threading.Thread | None = None
+        self._cycle_thread_task: asyncio.Task | None = None
 
-        self._task_list: List[asyncio.Task] = []
-        self.tg1: Optional[List[Coroutine[Any, Any, Any]]] = None
-        self.tg2: Optional[List[Coroutine[Any, Any, Any]]] = None
+        self._task_list: list[asyncio.Task] = []
+        self.tg1: list[Coroutine[Any, Any, Any]] | None = None
+        self.tg2: list[Coroutine[Any, Any, Any]] | None = None
         self.task_group_1 = asyncio.TaskGroup()
         self.task_group_2 = asyncio.TaskGroup()
         # self.pause_cycle = cothread.Event()
@@ -148,11 +149,11 @@ def __init__(
     # -----------------------------------------------------------------------
 
     async def _run_loops(
-        self, task_group: asyncio.TaskGroup, task_list: List[Coroutine[Any, Any, Any]]
+        self, task_group: asyncio.TaskGroup, task_list: list[Coroutine[Any, Any, Any]]
     ) -> None:
         async def create_task_group(
             task_group: asyncio.TaskGroup,
-            tasks: List[Coroutine[Any, Any, Any]],
+            tasks: list[Coroutine[Any, Any, Any]],
         ) -> None:
             # handle exceptions
             try:
@@ -347,7 +348,8 @@ async def cycle_control(self) -> None:
         self.run_update_time_params.set()
 
         tprint("Start Cycle")
-        # If already off, instantly set Time Since to 0 and also Ramp to ON, then wait MAX TIME
+        # If already off, instantly set Time Since to 0 and also Ramp to ON,
+        # then wait MAX TIME
         if math.isclose(self.voltage_rbv.get(), 0.0, abs_tol=volt_tol):
             await self.depolarise()