From 16c38f5a5047c229e74ade76718818ddab8aa89a Mon Sep 17 00:00:00 2001
From: Akio Taniguchi <13254278+astropenguin@users.noreply.github.com>
Date: Mon, 2 Mar 2026 10:00:10 +0000
Subject: [PATCH] #68 Do not use Union and Optional

---
 drs4/cal/dsbs.py   | 16 ++++++++--------
 drs4/ctrl/cw.py    | 25 ++++++++++++-------------
 drs4/ctrl/scpi.py  | 12 ++++++------
 drs4/ctrl/self.py  | 46 +++++++++++++++++++++++-----------------------
 drs4/daq/tcp.py    | 24 ++++++++++++------------
 drs4/daq/udp.py    | 40 ++++++++++++++++++++--------------------
 drs4/qlook/gui.py  | 11 +++++------
 drs4/specs/gain.py |  4 ++--
 drs4/specs/ms.py   | 12 ++++++------
 drs4/specs/vdif.py |  4 ++--
 drs4/utils.py      |  8 ++++----
 11 files changed, 100 insertions(+), 102 deletions(-)

diff --git a/drs4/cal/dsbs.py b/drs4/cal/dsbs.py
index f44b18d..1b1b267 100644
--- a/drs4/cal/dsbs.py
+++ b/drs4/cal/dsbs.py
@@ -3,7 +3,7 @@
 
 # standard library
 from logging import getLogger
-from typing import Optional, Union, get_args
+from typing import get_args
 
 # dependencies
 import xarray as xr
@@ -18,19 +18,19 @@
 
 
 def set_gain(
-    ms: Optional[Union[xr.Dataset, StrPath]] = None,
+    ms: xr.Dataset | StrPath | None = None,
     /,
     *,
     # for measurement (required)
-    chassis: Optional[Chassis] = None,
-    interface: Optional[Interface] = None,
+    chassis: Chassis | None = None,
+    interface: Interface | None = None,
     ones: bool = False,
     zeros: bool = False,
     # for connection (optional)
-    ctrl_addr: Optional[str] = None,
-    ctrl_user: Optional[str] = None,
-    timeout: Optional[float] = None,
-    workdir: Optional[StrPath] = None,
+    ctrl_addr: str | None = None,
+    ctrl_user: str | None = None,
+    timeout: float | None = None,
+    workdir: StrPath | None = None,
 ) -> None:
     """Set a gain file (DRS4 MS file) to DRS4.
 
diff --git a/drs4/ctrl/cw.py b/drs4/ctrl/cw.py
index 2c30155..ae2cca1 100644
--- a/drs4/ctrl/cw.py
+++ b/drs4/ctrl/cw.py
@@ -4,7 +4,6 @@
 # standard library
 from logging import getLogger
 from os import getenv
-from typing import Optional
 
 # constants
 LOGGER = getLogger(__name__)
@@ -33,13 +32,13 @@ def on(
     signal_sb: SideBand,
     signal_chan: Channel,
     # for measurement (optional)
-    lo_freq: Optional[float] = None,
-    lo_mult: Optional[int] = None,
-    sg_ampl: Optional[float] = None,
+    lo_freq: float | None = None,
+    lo_mult: int | None = None,
+    sg_ampl: float | None = None,
     # for connection (optional)
-    sg_host: Optional[str] = None,
-    sg_port: Optional[int] = None,
-    timeout: Optional[float] = None,
+    sg_host: str | None = None,
+    sg_port: int | None = None,
+    timeout: float | None = None,
 ) -> None:
     """Start outputting the CW signal after setting the SG amplitude and frequency.
 
@@ -109,9 +108,9 @@ def on(
 def off(
     *,
     # for connection (optional)
-    sg_host: Optional[str] = None,
-    sg_port: Optional[int] = None,
-    timeout: Optional[float] = None,
+    sg_host: str | None = None,
+    sg_port: int | None = None,
+    timeout: float | None = None,
 ) -> None:
     """Stop outputting the CW signal.
 
@@ -147,9 +146,9 @@ def off(
 def status(
     *,
     # for connection (optional)
-    sg_host: Optional[str] = None,
-    sg_port: Optional[int] = None,
-    timeout: Optional[float] = None,
+    sg_host: str | None = None,
+    sg_port: int | None = None,
+    timeout: float | None = None,
 ) -> None:
     """Show the status of CW signal in the logger.
 
diff --git a/drs4/ctrl/scpi.py b/drs4/ctrl/scpi.py
index 572e672..028b8a4 100644
--- a/drs4/ctrl/scpi.py
+++ b/drs4/ctrl/scpi.py
@@ -4,7 +4,7 @@
 # standard library
 from logging import getLogger
 from socket import socket, AF_INET, SOCK_STREAM
-from typing import IO, Optional, Sequence, Union
+from typing import IO, Sequence
 
 # dependencies
 from ..utils import StrPath
@@ -15,7 +15,7 @@
 DEFAULT_ENCODING: str = "ascii"
 DEFAULT_END: str = "\n"
 DEFAULT_FLAGS: int = 0
-DEFAULT_TIMEOUT: Optional[float] = None
+DEFAULT_TIMEOUT: float | None = None
 LOGGER = getLogger(__name__)
 
 
@@ -54,7 +54,7 @@ def recv(
 
 
 def connect(
-    host: str, port: int, timeout: Optional[float] = DEFAULT_TIMEOUT
+    host: str, port: int, timeout: float | None = DEFAULT_TIMEOUT
 ) -> CustomSocket:
     """Connect to an SCPI server and returns a custom socket object.
 
@@ -94,10 +94,10 @@ def connect(
 
 
 def send_commands(
-    commands: Union[IO[str], Sequence[str], str],
+    commands: IO[str] | Sequence[str] | str,
     host: str,
     port: int,
-    timeout: Optional[float] = DEFAULT_TIMEOUT,
+    timeout: float | None = DEFAULT_TIMEOUT,
     encoding: str = DEFAULT_ENCODING,
     autorecv: bool = DEFAULT_AUTORECV,
     bufsize: int = DEFAULT_BUFSIZE,
@@ -152,7 +152,7 @@ def send_commands_in(
     path: StrPath,
     host: str,
     port: int,
-    timeout: Optional[float] = DEFAULT_TIMEOUT,
+    timeout: float | None = DEFAULT_TIMEOUT,
     encoding: str = DEFAULT_ENCODING,
     autorecv: bool = DEFAULT_AUTORECV,
     bufsize: int = DEFAULT_BUFSIZE,
diff --git a/drs4/ctrl/self.py b/drs4/ctrl/self.py
index 2e749fe..9afa028 100644
--- a/drs4/ctrl/self.py
+++ b/drs4/ctrl/self.py
@@ -5,7 +5,7 @@
 from logging import getLogger
 from os import getenv
 from subprocess import PIPE, CompletedProcess, run as sprun
-from typing import Optional, Union, overload
+from typing import overload
 
 # dependencies
 from ..specs.common import ENV_CTRL_ADDR, ENV_CTRL_USER, Chassis
@@ -24,9 +24,9 @@
 def run(
     *commands: str,
     chassis: Chassis,
-    ctrl_addr: Optional[str] = None,
-    ctrl_user: Optional[str] = None,
-    timeout: Optional[float] = None,
+    ctrl_addr: str | None = None,
+    ctrl_user: str | None = None,
+    timeout: float | None = None,
     workdir: StrPath = PATH_CMD_DIR,
 ) -> StrCP: ...
 
@@ -35,9 +35,9 @@ def run(
 def run(
     *commands: str,
     chassis: None = None,
-    ctrl_addr: Optional[str] = None,
-    ctrl_user: Optional[str] = None,
-    timeout: Optional[float] = None,
+    ctrl_addr: str | None = None,
+    ctrl_user: str | None = None,
+    timeout: float | None = None,
     workdir: StrPath = PATH_CMD_DIR,
 ) -> tuple[StrCP, StrCP]: ...
 
@@ -46,12 +46,12 @@ def run(
     # for connection (required)
     *commands: str,
     # for connection (optional)
-    chassis: Optional[Chassis] = None,
-    ctrl_addr: Optional[str] = None,
-    ctrl_user: Optional[str] = None,
-    timeout: Optional[float] = None,
+    chassis: Chassis | None = None,
+    ctrl_addr: str | None = None,
+    ctrl_user: str | None = None,
+    timeout: float | None = None,
     workdir: StrPath = PATH_CMD_DIR,
-) -> Union[StrCP, tuple[StrCP, StrCP]]:
+) -> StrCP | tuple[StrCP, StrCP]:
     """Run commands in DRS4.
 
     Args:
@@ -140,9 +140,9 @@ def send(
     /,
     *,
     chassis: Chassis,
-    ctrl_addr: Optional[str] = None,
-    ctrl_user: Optional[str] = None,
-    timeout: Optional[float] = None,
+    ctrl_addr: str | None = None,
+    ctrl_user: str | None = None,
+    timeout: float | None = None,
 ) -> StrCP: ...
 
 
@@ -153,9 +153,9 @@ def send(
     /,
     *,
     chassis: None = None,
-    ctrl_addr: Optional[str] = None,
-    ctrl_user: Optional[str] = None,
-    timeout: Optional[float] = None,
+    ctrl_addr: str | None = None,
+    ctrl_user: str | None = None,
+    timeout: float | None = None,
 ) -> tuple[StrCP, StrCP]: ...
 
 
@@ -166,11 +166,11 @@ def send(
     /,
     *,
     # for connection (optional)
-    chassis: Optional[Chassis] = None,
-    ctrl_addr: Optional[str] = None,
-    ctrl_user: Optional[str] = None,
-    timeout: Optional[float] = None,
-) -> Union[StrCP, tuple[StrCP, StrCP]]:
+    chassis: Chassis | None = None,
+    ctrl_addr: str | None = None,
+    ctrl_user: str | None = None,
+    timeout: float | None = None,
+) -> StrCP | tuple[StrCP, StrCP]:
     """Send a file to DRS4.
 
     Args:
diff --git a/drs4/daq/tcp.py b/drs4/daq/tcp.py
index d8cd3e8..04b2301 100644
--- a/drs4/daq/tcp.py
+++ b/drs4/daq/tcp.py
@@ -6,7 +6,7 @@
 from logging import getLogger
 from os import getenv
 from pathlib import Path
-from typing import Optional, Union, get_args
+from typing import get_args
 
 # dependencies
 import xarray as xr
@@ -50,27 +50,27 @@ def cross(
     freq_range_if1: FreqRange = "inner",
     freq_range_if2: FreqRange = "outer",
     integ_time: IntegTime = 100,
-    signal_if: Optional[Interface] = None,
-    signal_sb: Optional[SideBand] = None,
-    signal_chan: Optional[Channel] = None,
+    signal_if: Interface | None = None,
+    signal_sb: SideBand | None = None,
+    signal_chan: Channel | None = None,
     # for file saving (optional)
     append: bool = False,
     integrate: bool = False,
     join: XarrayJoin = "inner",
     overwrite: bool = False,
     progress: bool = False,
-    workdir: Optional[StrPath] = None,
-    zarr_if1: Optional[StrPath] = None,
-    zarr_if2: Optional[StrPath] = None,
+    workdir: StrPath | None = None,
+    zarr_if1: StrPath | None = None,
+    zarr_if2: StrPath | None = None,
     # for DRS4 settings (optional)
     dsp_mode: DSPMode = "IQ",
-    gain_if1: Optional[Union[xr.Dataset, StrPath]] = None,
-    gain_if2: Optional[Union[xr.Dataset, StrPath]] = None,
+    gain_if1: xr.Dataset | StrPath | None = None,
+    gain_if2: xr.Dataset | StrPath | None = None,
     settings: bool = True,
     # for connection (optional)
-    ctrl_addr: Optional[str] = None,
-    ctrl_user: Optional[str] = None,
-    timeout: Optional[float] = None,
+    ctrl_addr: str | None = None,
+    ctrl_user: str | None = None,
+    timeout: float | None = None,
 ) -> tuple[Path, Path]:
     """"""
     obsid = datetime.now(timezone.utc).strftime(OBSID_FORMAT)
diff --git a/drs4/daq/udp.py b/drs4/daq/udp.py
index 6e0c5fd..ae523dd 100644
--- a/drs4/daq/udp.py
+++ b/drs4/daq/udp.py
@@ -19,7 +19,7 @@
 )
 from threading import Event
 from time import sleep
-from typing import Optional, Union, get_args
+from typing import get_args
 
 # dependencies
 import xarray as xr
@@ -64,32 +64,32 @@ def auto(
     freq_range_if1: FreqRange = "inner",
     freq_range_if2: FreqRange = "outer",
     integ_time: IntegTime = 100,
-    signal_if: Optional[Interface] = None,
-    signal_sb: Optional[SideBand] = None,
-    signal_chan: Optional[Channel] = None,
+    signal_if: Interface | None = None,
+    signal_sb: SideBand | None = None,
+    signal_chan: Channel | None = None,
     # for file saving (optional)
     append: bool = False,
     integrate: bool = False,
     join: XarrayJoin = "inner",
     overwrite: bool = False,
     progress: bool = False,
-    workdir: Optional[StrPath] = None,
-    zarr_if1: Optional[StrPath] = None,
-    zarr_if2: Optional[StrPath] = None,
+    workdir: StrPath | None = None,
+    zarr_if1: StrPath | None = None,
+    zarr_if2: StrPath | None = None,
     # for DRS4 settings (optional)
     dsp_mode: DSPMode = "SB",
-    gain_if1: Optional[Union[xr.Dataset, StrPath]] = None,
-    gain_if2: Optional[Union[xr.Dataset, StrPath]] = None,
+    gain_if1: xr.Dataset | StrPath | None = None,
+    gain_if2: xr.Dataset | StrPath | None = None,
     settings: bool = True,
     # for connection (optional)
-    ctrl_addr: Optional[str] = None,
-    ctrl_user: Optional[str] = None,
-    dest_addr: Optional[str] = None,
-    dest_port1: Optional[int] = None,
-    dest_port2: Optional[int] = None,
-    dest_port3: Optional[int] = None,
-    dest_port4: Optional[int] = None,
-    timeout: Optional[float] = None,
+    ctrl_addr: str | None = None,
+    ctrl_user: str | None = None,
+    dest_addr: str | None = None,
+    dest_port1: int | None = None,
+    dest_port2: int | None = None,
+    dest_port3: int | None = None,
+    dest_port4: int | None = None,
+    timeout: float | None = None,
 ) -> tuple[Path, Path]:
     """"""
     obsid = datetime.now(timezone.utc).strftime(OBSID_FORMAT)
@@ -301,7 +301,7 @@ def auto(
 
 
 def dump(
-    vdif: Union[Path, str],
+    vdif: StrPath,
     /,
     *,
     # for connection (required)
@@ -310,8 +310,8 @@ def dump(
     # for connection (optional)
     group: str = GROUP,
     # for file saving (optional)
-    cancel: Optional[Event] = None,
-    timeout: Optional[float] = None,
+    cancel: Event | None = None,
+    timeout: float | None = None,
     progress: bool = False,
     overwrite: bool = False,
 ) -> None:
diff --git a/drs4/qlook/gui.py b/drs4/qlook/gui.py
index 3a1589a..62ffd85 100644
--- a/drs4/qlook/gui.py
+++ b/drs4/qlook/gui.py
@@ -8,7 +8,6 @@
 from os.path import getmtime
 from pathlib import Path
 from subprocess import PIPE, run as sprun
-from typing import Optional
 
 # dependencies
 import matplotlib.pyplot as plt
@@ -33,13 +32,13 @@ def run(
     # for connection (required)
     chassis: Chassis,
     # for connection (optional)
-    ctrl_addr: Optional[str] = None,
-    ctrl_user: Optional[str] = None,
-    timeout: Optional[float] = None,
+    ctrl_addr: str | None = None,
+    ctrl_user: str | None = None,
+    timeout: float | None = None,
     # for plotting (optional)
-    figsize: Optional[tuple[int, int]] = (12, 6),
+    figsize: tuple[int, int] | None = (12, 6),
     interval: int = 10,
-    workdir: Optional[StrPath] = None,
+    workdir: StrPath | None = None,
 ) -> None:
     """Display current auto-spectra and bit distributions of DRS4.
 
diff --git a/drs4/specs/gain.py b/drs4/specs/gain.py
index 48fc29e..d118325 100644
--- a/drs4/specs/gain.py
+++ b/drs4/specs/gain.py
@@ -3,7 +3,7 @@
 
 # standard library
 from dataclasses import dataclass, field
-from typing import Literal as L, Union, overload
+from typing import Literal as L, overload
 
 # dependencies
 import numpy as np
@@ -78,7 +78,7 @@ def open_gain(
     /,
     *,
     format: L["DataFrame", "Dataset"] = "Dataset",
-) -> Union[pd.DataFrame, xr.Dataset]:
+) -> pd.DataFrame | xr.Dataset:
     """Open gain file (DRS4 MS file) as a DataFrame or a Dataset.
 
     Args:
diff --git a/drs4/specs/ms.py b/drs4/specs/ms.py
index 52215e1..23d7d92 100644
--- a/drs4/specs/ms.py
+++ b/drs4/specs/ms.py
@@ -3,7 +3,7 @@
 
 # standard library
 from dataclasses import dataclass, field
-from typing import Literal as L, Optional, get_args
+from typing import Literal as L, get_args
 
 # dependencies
 import numpy as np
@@ -105,8 +105,8 @@ def open_csvs(
     freq_range: FreqRange,
     integ_time: IntegTime,
     # for measurement (optional)
-    signal_sb: Optional[SideBand] = None,
-    signal_chan: Optional[Channel] = None,
+    signal_sb: SideBand | None = None,
+    signal_chan: Channel | None = None,
     # for file loading (optional)
     join: XarrayJoin = "inner",
 ) -> xr.Dataset:
@@ -186,9 +186,9 @@ def open_vdifs(
     interface: Interface,
     freq_range: FreqRange,
     # for measurement (optional)
-    integ_time: Optional[IntegTime] = None,
-    signal_sb: Optional[SideBand] = None,
-    signal_chan: Optional[Channel] = None,
+    integ_time: IntegTime | None = None,
+    signal_sb: SideBand | None = None,
+    signal_chan: Channel | None = None,
     # for file loading (optional)
     join: XarrayJoin = "inner",
 ) -> xr.Dataset:
diff --git a/drs4/specs/vdif.py b/drs4/specs/vdif.py
index 8387ffe..b58d94d 100644
--- a/drs4/specs/vdif.py
+++ b/drs4/specs/vdif.py
@@ -3,7 +3,7 @@
 
 # standard library
 from dataclasses import dataclass, field
-from typing import Literal as L, Optional, get_args
+from typing import Literal as L, get_args
 
 # dependencies
 import numpy as np
@@ -72,7 +72,7 @@ def open_vdif(
     /,
     *,
     # for measurement (optional)
-    integ_time: Optional[IntegTime] = None,
+    integ_time: IntegTime | None = None,
     # for file loading (optional)
     join: XarrayJoin = "inner",
 ) -> xr.DataArray:
diff --git a/drs4/utils.py b/drs4/utils.py
index 623f46b..55b1db0 100644
--- a/drs4/utils.py
+++ b/drs4/utils.py
@@ -9,7 +9,7 @@
 from pathlib import Path
 from tempfile import TemporaryDirectory
 from time import gmtime
-from typing import Any, Literal as L, Optional, Union
+from typing import Any, Literal as L
 
 # dependencies
 import numpy as np
@@ -21,8 +21,8 @@
 
 
 # type hints
-Axis = Optional[Union[Sequence[int], int]]
-StrPath = Union[PathLike[str], str]
+Axis = Sequence[int] | int | None
+StrPath = PathLike[str] | str
 XarrayJoin = L["outer", "inner", "left", "right", "exact", "override"]
 
 
@@ -32,7 +32,7 @@ def is_strpath(obj: Any, /) -> TypeGuard[StrPath]:
 
 
 @contextmanager
-def set_workdir(workdir: Optional[StrPath] = None, /) -> Iterator[Path]:
+def set_workdir(workdir: StrPath | None = None, /) -> Iterator[Path]:
     """Set the working directory for output VDIF files."""
     if workdir is not None:
         yield Path(workdir).expanduser()