From 4245c77ffdb458b9352bf1a9467fb6ce6d092201 Mon Sep 17 00:00:00 2001
From: Oscar Gustafsson <oscar.gustafsson@liu.se>
Date: Thu, 27 Feb 2025 16:33:31 +0100
Subject: [PATCH] Update code to Python 3.10
---
b_asic/GUI/arrow.py | 4 +-
b_asic/GUI/drag_button.py | 6 +-
b_asic/GUI/main_window.py | 31 +-
b_asic/GUI/simulate_sfg_window.py | 4 +-
b_asic/architecture.py | 100 +++----
b_asic/core_operations.py | 366 ++++++++++++------------
b_asic/graph_component.py | 5 +-
b_asic/gui_utils/plot_window.py | 6 +-
b_asic/logger.py | 5 +-
b_asic/operation.py | 113 ++++----
b_asic/port.py | 17 +-
b_asic/process.py | 48 ++--
b_asic/research/interleaver.py | 9 +-
b_asic/resources.py | 115 ++++----
b_asic/save_load_structure.py | 6 +-
b_asic/schedule.py | 57 ++--
b_asic/scheduler.py | 16 +-
b_asic/scheduler_gui/axes_item.py | 13 +-
b_asic/scheduler_gui/main_window.py | 24 +-
b_asic/scheduler_gui/operation_item.py | 12 +-
b_asic/scheduler_gui/scheduler_event.py | 10 +-
b_asic/scheduler_gui/scheduler_item.py | 18 +-
b_asic/scheduler_gui/timeline_item.py | 10 +-
b_asic/sfg_generators.py | 35 +--
b_asic/signal.py | 13 +-
b_asic/signal_flow_graph.py | 92 +++---
b_asic/signal_generator.py | 22 +-
b_asic/simulation.py | 24 +-
b_asic/special_operations.py | 28 +-
b_asic/utils.py | 10 +-
30 files changed, 597 insertions(+), 622 deletions(-)
diff --git a/b_asic/GUI/arrow.py b/b_asic/GUI/arrow.py
index 84d52441..02a72bde 100644
--- a/b_asic/GUI/arrow.py
+++ b/b_asic/GUI/arrow.py
@@ -1,4 +1,4 @@
-from typing import TYPE_CHECKING, Optional, cast
+from typing import TYPE_CHECKING, cast
from qtpy.QtCore import QPointF
from qtpy.QtGui import QPainterPath, QPen
@@ -38,7 +38,7 @@ class Arrow(QGraphicsPathItem):
source_port_button: "PortButton",
destination_port_button: "PortButton",
window: "SFGMainWindow",
- signal: Optional[Signal] = None,
+ signal: Signal | None = None,
parent=None,
):
super().__init__(parent)
diff --git a/b_asic/GUI/drag_button.py b/b_asic/GUI/drag_button.py
index dc13b756..18308ef7 100644
--- a/b_asic/GUI/drag_button.py
+++ b/b_asic/GUI/drag_button.py
@@ -5,7 +5,7 @@ Contains a GUI class for drag buttons.
"""
import os.path
-from typing import TYPE_CHECKING, List
+from typing import TYPE_CHECKING
from qtpy.QtCore import QSize, Qt, Signal
from qtpy.QtGui import QIcon
@@ -56,7 +56,7 @@ class DragButton(QPushButton):
parent=None,
):
self.name = operation.name or operation.graph_id
- self._ports: List[PortButton] = []
+ self._ports: list[PortButton] = []
self.show_name = show_name
self._window = window
self.operation = operation
@@ -115,7 +115,7 @@ class DragButton(QPushButton):
super().mousePressEvent(event)
@property
- def port_list(self) -> List[PortButton]:
+ def port_list(self) -> list[PortButton]:
"""Return a list of PortButtons."""
return self._ports
diff --git a/b_asic/GUI/main_window.py b/b_asic/GUI/main_window.py
index 90146b87..15b5f98c 100644
--- a/b_asic/GUI/main_window.py
+++ b/b_asic/GUI/main_window.py
@@ -10,8 +10,9 @@ import os
import sys
import webbrowser
from collections import deque
+from collections.abc import Sequence
from types import ModuleType
-from typing import TYPE_CHECKING, Deque, Dict, List, Optional, Sequence, Tuple, cast
+from typing import TYPE_CHECKING, Deque, cast
from qtpy.QtCore import QCoreApplication, QFileInfo, QSettings, QSize, Qt, QThread, Slot
from qtpy.QtGui import QCursor, QIcon, QKeySequence, QPainter
@@ -78,20 +79,20 @@ class SFGMainWindow(QMainWindow):
self._ui.setupUi(self)
self.setWindowIcon(QIcon("small_logo.png"))
self._scene = QGraphicsScene(self._ui.splitter)
- self._operations_from_name: Dict[str, Operation] = {}
+ self._operations_from_name: dict[str, Operation] = {}
self._zoom = 1
- self._drag_operation_scenes: Dict[DragButton, "QGraphicsProxyWidget"] = {}
- self._drag_buttons: Dict[Operation, DragButton] = {}
+ self._drag_operation_scenes: dict[DragButton, "QGraphicsProxyWidget"] = {}
+ self._drag_buttons: dict[Operation, DragButton] = {}
self._mouse_pressed = False
self._mouse_dragging = False
self._starting_port = None
- self._pressed_operations: List[DragButton] = []
- self._arrow_ports: Dict[Arrow, List[Tuple[PortButton, PortButton]]] = {}
- self._operation_to_sfg: Dict[DragButton, SFG] = {}
- self._pressed_ports: List[PortButton] = []
- self._sfg_dict: Dict[str, SFG] = {}
- self._plot: Dict[Simulation, PlotWindow] = {}
- self._ports: Dict[DragButton, List[PortButton]] = {}
+ self._pressed_operations: list[DragButton] = []
+ self._arrow_ports: dict[Arrow, list[tuple[PortButton, PortButton]]] = {}
+ self._operation_to_sfg: dict[DragButton, SFG] = {}
+ self._pressed_ports: list[PortButton] = []
+ self._sfg_dict: dict[str, SFG] = {}
+ self._plot: dict[Simulation, PlotWindow] = {}
+ self._ports: dict[DragButton, list[PortButton]] = {}
# Create Graphics View
self._graphics_view = QGraphicsView(self._scene, self._ui.splitter)
@@ -119,7 +120,7 @@ class SFGMainWindow(QMainWindow):
# Add operations
self._max_recent_files = 4
- self._recent_files_actions: List[QAction] = []
+ self._recent_files_actions: list[QAction] = []
self._recent_files_paths: Deque[str] = deque(maxlen=self._max_recent_files)
self.add_operations_from_namespace(
@@ -600,7 +601,7 @@ class SFGMainWindow(QMainWindow):
"""Callback for toggling the status bar."""
self._statusbar.setVisible(self._statusbar_visible.isChecked())
- def get_operations_from_namespace(self, namespace: ModuleType) -> List[str]:
+ def get_operations_from_namespace(self, namespace: ModuleType) -> list[str]:
"""
Return a list of all operations defined in a namespace (module).
@@ -671,7 +672,7 @@ class SFGMainWindow(QMainWindow):
def add_operation(
self,
op: Operation,
- position: Optional[Tuple[float, float]] = None,
+ position: tuple[float, float] | None = None,
is_flipped: bool = False,
) -> None:
"""
@@ -963,7 +964,7 @@ class SFGMainWindow(QMainWindow):
self._keybindings_page.show()
-def start_editor(sfg: Optional[SFG] = None) -> Dict[str, SFG]:
+def start_editor(sfg: SFG | None = None) -> dict[str, SFG]:
"""
Start the SFG editor.
diff --git a/b_asic/GUI/simulate_sfg_window.py b/b_asic/GUI/simulate_sfg_window.py
index 29dc5c8e..6e655242 100644
--- a/b_asic/GUI/simulate_sfg_window.py
+++ b/b_asic/GUI/simulate_sfg_window.py
@@ -2,7 +2,7 @@
B-ASIC window to simulate an SFG.
"""
-from typing import TYPE_CHECKING, Dict
+from typing import TYPE_CHECKING
from qtpy.QtCore import Qt, Signal
from qtpy.QtWidgets import (
@@ -181,6 +181,6 @@ class SimulateSFGWindow(QDialog):
self.simulate.emit()
@property
- def properties(self) -> Dict:
+ def properties(self) -> dict:
"""Return the simulation properties."""
return self._properties
diff --git a/b_asic/architecture.py b/b_asic/architecture.py
index 6e10d0f6..03c98379 100644
--- a/b_asic/architecture.py
+++ b/b_asic/architecture.py
@@ -3,20 +3,12 @@ B-ASIC architecture classes.
"""
from collections import defaultdict
+from collections.abc import Iterable, Iterator
from io import TextIOWrapper
from itertools import chain
from typing import (
DefaultDict,
- Dict,
- Iterable,
- Iterator,
- List,
Literal,
- Optional,
- Set,
- Tuple,
- Type,
- Union,
cast,
)
@@ -55,13 +47,13 @@ class HardwareBlock:
"""
__slots__ = "_entity_name"
- _entity_name: Optional[str]
+ _entity_name: str | None
__slots__ = "_entity_name"
- _entity_name: Optional[str]
+ _entity_name: str | None
- def __init__(self, entity_name: Optional[str] = None):
- self._entity_name: Optional[str] = None
+ def __init__(self, entity_name: str | None = None):
+ self._entity_name: str | None = None
if entity_name is not None:
self.set_entity_name(entity_name)
@@ -161,7 +153,7 @@ class Resource(HardwareBlock):
"""
def __init__(
- self, process_collection: ProcessCollection, entity_name: Optional[str] = None
+ self, process_collection: ProcessCollection, entity_name: str | None = None
):
if not len(process_collection):
raise ValueError("Do not create Resource with empty ProcessCollection")
@@ -169,7 +161,7 @@ class Resource(HardwareBlock):
self._collection = process_collection
self._input_count = -1
self._output_count = -1
- self._assignment: Optional[List[ProcessCollection]] = None
+ self._assignment: list[ProcessCollection] | None = None
def __repr__(self):
return self.entity_name
@@ -311,7 +303,7 @@ class Resource(HardwareBlock):
return self._collection
@property
- def operation_type(self) -> Union[Type[MemoryProcess], Type[Operation]]:
+ def operation_type(self) -> type[MemoryProcess] | type[Operation]:
raise NotImplementedError("ABC Resource does not implement operation_type")
def add_process(self, proc: Process, assign=False):
@@ -387,12 +379,12 @@ class ProcessingElement(Resource):
_color = f"#{''.join(f'{v:0>2X}' for v in PE_COLOR)}"
__slots__ = ("_process_collection", "_entity_name")
_process_collection: ProcessCollection
- _entity_name: Optional[str]
+ _entity_name: str | None
def __init__(
self,
process_collection: ProcessCollection,
- entity_name: Optional[str] = None,
+ entity_name: str | None = None,
assign: bool = True,
):
super().__init__(process_collection=process_collection, entity_name=entity_name)
@@ -426,7 +418,7 @@ class ProcessingElement(Resource):
self.assign()
@property
- def processes(self) -> List[OperatorProcess]:
+ def processes(self) -> list[OperatorProcess]:
return [cast(OperatorProcess, p) for p in self._collection]
def assign(
@@ -451,7 +443,7 @@ class ProcessingElement(Resource):
raise ValueError("Cannot map ProcessCollection to single ProcessingElement")
@property
- def operation_type(self) -> Type[Operation]:
+ def operation_type(self) -> type[Operation]:
return self._operation_type
@@ -489,20 +481,20 @@ class Memory(Resource):
)
_process_collection: ProcessCollection
_memory_type: Literal["RAM", "register"]
- _entity_name: Optional[str]
- _read_ports: Optional[int]
- _write_ports: Optional[int]
- _total_ports: Optional[int]
+ _entity_name: str | None
+ _read_ports: int | None
+ _write_ports: int | None
+ _total_ports: int | None
_assign: bool
def __init__(
self,
process_collection: ProcessCollection,
memory_type: Literal["RAM", "register"] = "RAM",
- entity_name: Optional[str] = None,
- read_ports: Optional[int] = None,
- write_ports: Optional[int] = None,
- total_ports: Optional[int] = None,
+ entity_name: str | None = None,
+ read_ports: int | None = None,
+ write_ports: int | None = None,
+ total_ports: int | None = None,
assign: bool = False,
):
super().__init__(process_collection=process_collection, entity_name=entity_name)
@@ -587,7 +579,7 @@ class Memory(Resource):
raise NotImplementedError()
@property
- def operation_type(self) -> Type[MemoryProcess]:
+ def operation_type(self) -> type[MemoryProcess]:
return self._operation_type
@@ -611,10 +603,10 @@ of :class:`~b_asic.architecture.ProcessingElement`
def __init__(
self,
- processing_elements: Union[ProcessingElement, Iterable[ProcessingElement]],
- memories: Union[Memory, Iterable[Memory]],
+ processing_elements: ProcessingElement | Iterable[ProcessingElement],
+ memories: Memory | Iterable[Memory],
entity_name: str = "arch",
- direct_interconnects: Optional[ProcessCollection] = None,
+ direct_interconnects: ProcessCollection | None = None,
):
super().__init__(entity_name)
self._processing_elements = (
@@ -625,13 +617,13 @@ of :class:`~b_asic.architecture.ProcessingElement`
self._memories = [memories] if isinstance(memories, Memory) else list(memories)
self._direct_interconnects = direct_interconnects
self._variable_input_port_to_resource: DefaultDict[
- InputPort, Set[Tuple[Resource, int]]
+ InputPort, set[tuple[Resource, int]]
] = defaultdict(set)
self._variable_outport_to_resource: DefaultDict[
- OutputPort, Set[Tuple[Resource, int]]
+ OutputPort, set[tuple[Resource, int]]
] = defaultdict(set)
- self._operation_input_port_to_resource: Dict[InputPort, Resource] = {}
- self._operation_outport_to_resource: Dict[OutputPort, Resource] = {}
+ self._operation_input_port_to_resource: dict[InputPort, Resource] = {}
+ self._operation_outport_to_resource: dict[OutputPort, Resource] = {}
self._schedule_time = self._check_and_get_schedule_time()
@@ -705,8 +697,8 @@ of :class:`~b_asic.architecture.ProcessingElement`
memory_write_ports.add(mv.write_port)
memory_read_ports.update(mv.read_ports)
- pe_input_ports: Set[InputPort] = set()
- pe_output_ports: Set[OutputPort] = set()
+ pe_input_ports: set[InputPort] = set()
+ pe_output_ports: set[OutputPort] = set()
for pe in self.processing_elements:
for operator in pe.processes:
pe_input_ports.update(operator.operation.inputs)
@@ -727,8 +719,8 @@ of :class:`~b_asic.architecture.ProcessingElement`
)
def get_interconnects_for_memory(
- self, mem: Union[Memory, str]
- ) -> Tuple[Dict[Resource, int], Dict[Resource, int]]:
+ self, mem: Memory | str
+ ) -> tuple[dict[Resource, int], dict[Resource, int]]:
"""
Return a dictionary with interconnect information for a Memory.
@@ -756,9 +748,9 @@ of :class:`~b_asic.architecture.ProcessingElement`
return dict(d_in), dict(d_out)
def get_interconnects_for_pe(
- self, pe: Union[str, ProcessingElement]
- ) -> Tuple[
- List[Dict[Tuple[Resource, int], int]], List[Dict[Tuple[Resource, int], int]]
+ self, pe: str | ProcessingElement
+ ) -> tuple[
+ list[dict[tuple[Resource, int], int]], list[dict[tuple[Resource, int], int]]
]:
"""
Return with interconnect information for a ProcessingElement.
@@ -782,10 +774,10 @@ of :class:`~b_asic.architecture.ProcessingElement`
if isinstance(pe, str):
pe = cast(ProcessingElement, self.resource_from_name(pe))
- d_in: List[DefaultDict[Tuple[Resource, int], int]] = [
+ d_in: list[DefaultDict[tuple[Resource, int], int]] = [
defaultdict(_interconnect_dict) for _ in range(pe.input_count)
]
- d_out: List[DefaultDict[Tuple[Resource, int], int]] = [
+ d_out: list[DefaultDict[tuple[Resource, int], int]] = [
defaultdict(_interconnect_dict) for _ in range(pe.output_count)
]
for var in pe.collection:
@@ -817,7 +809,7 @@ of :class:`~b_asic.architecture.ProcessingElement`
def remove_resource(
self,
- resource: Union[str, Resource],
+ resource: str | Resource,
) -> None:
"""
Remove an empty :class:`Resource` from the architecture.
@@ -860,9 +852,9 @@ of :class:`~b_asic.architecture.ProcessingElement`
def move_process(
self,
- proc: Union[str, Process],
- source: Union[str, Resource],
- destination: Union[str, Resource],
+ proc: str | Process,
+ source: str | Resource,
+ destination: str | Resource,
assign: bool = False,
) -> None:
"""
@@ -1032,8 +1024,8 @@ of :class:`~b_asic.architecture.ProcessingElement`
)
# Create list of interconnects
- edges: DefaultDict[str, Set[Tuple[str, str]]] = defaultdict(set)
- destination_edges: DefaultDict[str, Set[str]] = defaultdict(set)
+ edges: DefaultDict[str, set[tuple[str, str]]] = defaultdict(set)
+ destination_edges: DefaultDict[str, set[str]] = defaultdict(set)
for pe in self._processing_elements:
inputs, outputs = self.get_interconnects_for_pe(pe)
for i, inp in enumerate(inputs):
@@ -1106,15 +1098,15 @@ of :class:`~b_asic.architecture.ProcessingElement`
return dg
@property
- def memories(self) -> List[Memory]:
+ def memories(self) -> list[Memory]:
return self._memories
@property
- def processing_elements(self) -> List[ProcessingElement]:
+ def processing_elements(self) -> list[ProcessingElement]:
return self._processing_elements
@property
- def direct_interconnects(self) -> Optional[ProcessCollection]:
+ def direct_interconnects(self) -> ProcessCollection | None:
return self._direct_interconnects
@property
diff --git a/b_asic/core_operations.py b/b_asic/core_operations.py
index d09e0c62..e16a5ce0 100644
--- a/b_asic/core_operations.py
+++ b/b_asic/core_operations.py
@@ -4,8 +4,6 @@ B-ASIC Core Operations Module.
Contains some of the most commonly used mathematical operations.
"""
-from typing import Dict, Optional
-
from numpy import abs as np_abs
from numpy import conjugate, sqrt
@@ -149,24 +147,24 @@ class Addition(AbstractOperation):
"_latency_offsets",
"_execution_time",
)
- _src0: Optional[SignalSourceProvider]
- _src1: Optional[SignalSourceProvider]
+ _src0: SignalSourceProvider | None
+ _src1: SignalSourceProvider | None
_name: Name
- _latency: Optional[int]
- _latency_offsets: Optional[Dict[str, int]]
- _execution_time: Optional[int]
+ _latency: int | None
+ _latency_offsets: dict[str, int] | None
+ _execution_time: int | None
is_linear = True
is_swappable = True
def __init__(
self,
- src0: Optional[SignalSourceProvider] = None,
- src1: Optional[SignalSourceProvider] = None,
+ src0: SignalSourceProvider | None = None,
+ src1: SignalSourceProvider | None = None,
name: Name = Name(""),
- latency: Optional[int] = None,
- latency_offsets: Optional[Dict[str, int]] = None,
- execution_time: Optional[int] = None,
+ latency: int | None = None,
+ latency_offsets: dict[str, int] | None = None,
+ execution_time: int | None = None,
):
"""
Construct an Addition operation.
@@ -231,21 +229,21 @@ class Subtraction(AbstractOperation):
"_latency_offsets",
"_execution_time",
)
- _src0: Optional[SignalSourceProvider]
- _src1: Optional[SignalSourceProvider]
+ _src0: SignalSourceProvider | None
+ _src1: SignalSourceProvider | None
_name: Name
- _latency: Optional[int]
- _latency_offsets: Optional[Dict[str, int]]
- _execution_time: Optional[int]
+ _latency: int | None
+ _latency_offsets: dict[str, int] | None
+ _execution_time: int | None
def __init__(
self,
- src0: Optional[SignalSourceProvider] = None,
- src1: Optional[SignalSourceProvider] = None,
+ src0: SignalSourceProvider | None = None,
+ src1: SignalSourceProvider | None = None,
name: Name = Name(""),
- latency: Optional[int] = None,
- latency_offsets: Optional[Dict[str, int]] = None,
- execution_time: Optional[int] = None,
+ latency: int | None = None,
+ latency_offsets: dict[str, int] | None = None,
+ execution_time: int | None = None,
):
"""Construct a Subtraction operation."""
super().__init__(
@@ -317,24 +315,24 @@ class AddSub(AbstractOperation):
"_execution_time",
)
_is_add: bool
- _src0: Optional[SignalSourceProvider]
- _src1: Optional[SignalSourceProvider]
+ _src0: SignalSourceProvider | None
+ _src1: SignalSourceProvider | None
_name: Name
- _latency: Optional[int]
- _latency_offsets: Optional[Dict[str, int]]
- _execution_time: Optional[int]
+ _latency: int | None
+ _latency_offsets: dict[str, int] | None
+ _execution_time: int | None
is_linear = True
def __init__(
self,
is_add: bool = True,
- src0: Optional[SignalSourceProvider] = None,
- src1: Optional[SignalSourceProvider] = None,
+ src0: SignalSourceProvider | None = None,
+ src1: SignalSourceProvider | None = None,
name: Name = Name(""),
- latency: Optional[int] = None,
- latency_offsets: Optional[Dict[str, int]] = None,
- execution_time: Optional[int] = None,
+ latency: int | None = None,
+ latency_offsets: dict[str, int] | None = None,
+ execution_time: int | None = None,
):
"""Construct an Addition/Subtraction operation."""
super().__init__(
@@ -410,23 +408,23 @@ class Multiplication(AbstractOperation):
"_latency_offsets",
"_execution_time",
)
- _src0: Optional[SignalSourceProvider]
- _src1: Optional[SignalSourceProvider]
+ _src0: SignalSourceProvider | None
+ _src1: SignalSourceProvider | None
_name: Name
- _latency: Optional[int]
- _latency_offsets: Optional[Dict[str, int]]
- _execution_time: Optional[int]
+ _latency: int | None
+ _latency_offsets: dict[str, int] | None
+ _execution_time: int | None
is_swappable = True
def __init__(
self,
- src0: Optional[SignalSourceProvider] = None,
- src1: Optional[SignalSourceProvider] = None,
+ src0: SignalSourceProvider | None = None,
+ src1: SignalSourceProvider | None = None,
name: Name = Name(""),
- latency: Optional[int] = None,
- latency_offsets: Optional[Dict[str, int]] = None,
- execution_time: Optional[int] = None,
+ latency: int | None = None,
+ latency_offsets: dict[str, int] | None = None,
+ execution_time: int | None = None,
):
"""Construct a Multiplication operation."""
super().__init__(
@@ -493,21 +491,21 @@ class Division(AbstractOperation):
"_latency_offsets",
"_execution_time",
)
- _src0: Optional[SignalSourceProvider]
- _src1: Optional[SignalSourceProvider]
+ _src0: SignalSourceProvider | None
+ _src1: SignalSourceProvider | None
_name: Name
- _latency: Optional[int]
- _latency_offsets: Optional[Dict[str, int]]
- _execution_time: Optional[int]
+ _latency: int | None
+ _latency_offsets: dict[str, int] | None
+ _execution_time: int | None
def __init__(
self,
- src0: Optional[SignalSourceProvider] = None,
- src1: Optional[SignalSourceProvider] = None,
+ src0: SignalSourceProvider | None = None,
+ src1: SignalSourceProvider | None = None,
name: Name = Name(""),
- latency: Optional[int] = None,
- latency_offsets: Optional[Dict[str, int]] = None,
- execution_time: Optional[int] = None,
+ latency: int | None = None,
+ latency_offsets: dict[str, int] | None = None,
+ execution_time: int | None = None,
):
"""Construct a Division operation."""
super().__init__(
@@ -574,23 +572,23 @@ class Min(AbstractOperation):
"_latency_offsets",
"_execution_time",
)
- _src0: Optional[SignalSourceProvider]
- _src1: Optional[SignalSourceProvider]
+ _src0: SignalSourceProvider | None
+ _src1: SignalSourceProvider | None
_name: Name
- _latency: Optional[int]
- _latency_offsets: Optional[Dict[str, int]]
- _execution_time: Optional[int]
+ _latency: int | None
+ _latency_offsets: dict[str, int] | None
+ _execution_time: int | None
is_swappable = True
def __init__(
self,
- src0: Optional[SignalSourceProvider] = None,
- src1: Optional[SignalSourceProvider] = None,
+ src0: SignalSourceProvider | None = None,
+ src1: SignalSourceProvider | None = None,
name: Name = Name(""),
- latency: Optional[int] = None,
- latency_offsets: Optional[Dict[str, int]] = None,
- execution_time: Optional[int] = None,
+ latency: int | None = None,
+ latency_offsets: dict[str, int] | None = None,
+ execution_time: int | None = None,
):
"""Construct a Min operation."""
super().__init__(
@@ -655,23 +653,23 @@ class Max(AbstractOperation):
"_latency_offsets",
"_execution_time",
)
- _src0: Optional[SignalSourceProvider]
- _src1: Optional[SignalSourceProvider]
+ _src0: SignalSourceProvider | None
+ _src1: SignalSourceProvider | None
_name: Name
- _latency: Optional[int]
- _latency_offsets: Optional[Dict[str, int]]
- _execution_time: Optional[int]
+ _latency: int | None
+ _latency_offsets: dict[str, int] | None
+ _execution_time: int | None
is_swappable = True
def __init__(
self,
- src0: Optional[SignalSourceProvider] = None,
- src1: Optional[SignalSourceProvider] = None,
+ src0: SignalSourceProvider | None = None,
+ src1: SignalSourceProvider | None = None,
name: Name = Name(""),
- latency: Optional[int] = None,
- latency_offsets: Optional[Dict[str, int]] = None,
- execution_time: Optional[int] = None,
+ latency: int | None = None,
+ latency_offsets: dict[str, int] | None = None,
+ execution_time: int | None = None,
):
"""Construct a Max operation."""
super().__init__(
@@ -719,19 +717,19 @@ class SquareRoot(AbstractOperation):
"""
__slots__ = ("_src0", "_name", "_latency", "_latency_offsets", "_execution_time")
- _src0: Optional[SignalSourceProvider]
+ _src0: SignalSourceProvider | None
_name: Name
- _latency: Optional[int]
- _latency_offsets: Optional[Dict[str, int]]
- _execution_time: Optional[int]
+ _latency: int | None
+ _latency_offsets: dict[str, int] | None
+ _execution_time: int | None
def __init__(
self,
- src0: Optional[SignalSourceProvider] = None,
+ src0: SignalSourceProvider | None = None,
name: Name = Name(""),
- latency: Optional[int] = None,
- latency_offsets: Optional[Dict[str, int]] = None,
- execution_time: Optional[int] = None,
+ latency: int | None = None,
+ latency_offsets: dict[str, int] | None = None,
+ execution_time: int | None = None,
):
"""Construct a SquareRoot operation."""
super().__init__(
@@ -777,19 +775,19 @@ class ComplexConjugate(AbstractOperation):
"""
__slots__ = ("_src0", "_name", "_latency", "_latency_offsets", "_execution_time")
- _src0: Optional[SignalSourceProvider]
+ _src0: SignalSourceProvider | None
_name: Name
- _latency: Optional[int]
- _latency_offsets: Optional[Dict[str, int]]
- _execution_time: Optional[int]
+ _latency: int | None
+ _latency_offsets: dict[str, int] | None
+ _execution_time: int | None
def __init__(
self,
- src0: Optional[SignalSourceProvider] = None,
+ src0: SignalSourceProvider | None = None,
name: Name = Name(""),
- latency: Optional[int] = None,
- latency_offsets: Optional[Dict[str, int]] = None,
- execution_time: Optional[int] = None,
+ latency: int | None = None,
+ latency_offsets: dict[str, int] | None = None,
+ execution_time: int | None = None,
):
"""Construct a ComplexConjugate operation."""
super().__init__(
@@ -835,19 +833,19 @@ class Absolute(AbstractOperation):
"""
__slots__ = ("_src0", "_name", "_latency", "_latency_offsets", "_execution_time")
- _src0: Optional[SignalSourceProvider]
+ _src0: SignalSourceProvider | None
_name: Name
- _latency: Optional[int]
- _latency_offsets: Optional[Dict[str, int]]
- _execution_time: Optional[int]
+ _latency: int | None
+ _latency_offsets: dict[str, int] | None
+ _execution_time: int | None
def __init__(
self,
- src0: Optional[SignalSourceProvider] = None,
+ src0: SignalSourceProvider | None = None,
name: Name = Name(""),
- latency: Optional[int] = None,
- latency_offsets: Optional[Dict[str, int]] = None,
- execution_time: Optional[int] = None,
+ latency: int | None = None,
+ latency_offsets: dict[str, int] | None = None,
+ execution_time: int | None = None,
):
"""Construct an Absolute operation."""
super().__init__(
@@ -907,22 +905,22 @@ class ConstantMultiplication(AbstractOperation):
"_execution_time",
)
_value: Num
- _src0: Optional[SignalSourceProvider]
+ _src0: SignalSourceProvider | None
_name: Name
- _latency: Optional[int]
- _latency_offsets: Optional[Dict[str, int]]
- _execution_time: Optional[int]
+ _latency: int | None
+ _latency_offsets: dict[str, int] | None
+ _execution_time: int | None
is_linear = True
def __init__(
self,
value: Num = 0,
- src0: Optional[SignalSourceProvider] = None,
+ src0: SignalSourceProvider | None = None,
name: Name = Name(""),
- latency: Optional[int] = None,
- latency_offsets: Optional[Dict[str, int]] = None,
- execution_time: Optional[int] = None,
+ latency: int | None = None,
+ latency_offsets: dict[str, int] | None = None,
+ execution_time: int | None = None,
):
"""Construct a ConstantMultiplication operation with the given value."""
super().__init__(
@@ -995,23 +993,23 @@ class Butterfly(AbstractOperation):
"_latency_offsets",
"_execution_time",
)
- _src0: Optional[SignalSourceProvider]
- _src1: Optional[SignalSourceProvider]
+ _src0: SignalSourceProvider | None
+ _src1: SignalSourceProvider | None
_name: Name
- _latency: Optional[int]
- _latency_offsets: Optional[Dict[str, int]]
- _execution_time: Optional[int]
+ _latency: int | None
+ _latency_offsets: dict[str, int] | None
+ _execution_time: int | None
is_linear = True
def __init__(
self,
- src0: Optional[SignalSourceProvider] = None,
- src1: Optional[SignalSourceProvider] = None,
+ src0: SignalSourceProvider | None = None,
+ src1: SignalSourceProvider | None = None,
name: Name = Name(""),
- latency: Optional[int] = None,
- latency_offsets: Optional[Dict[str, int]] = None,
- execution_time: Optional[int] = None,
+ latency: int | None = None,
+ latency_offsets: dict[str, int] | None = None,
+ execution_time: int | None = None,
):
"""Construct a Butterfly operation."""
super().__init__(
@@ -1075,25 +1073,25 @@ class MAD(AbstractOperation):
"_latency_offsets",
"_execution_time",
)
- _src0: Optional[SignalSourceProvider]
- _src1: Optional[SignalSourceProvider]
- _src2: Optional[SignalSourceProvider]
+ _src0: SignalSourceProvider | None
+ _src1: SignalSourceProvider | None
+ _src2: SignalSourceProvider | None
_name: Name
- _latency: Optional[int]
- _latency_offsets: Optional[Dict[str, int]]
- _execution_time: Optional[int]
+ _latency: int | None
+ _latency_offsets: dict[str, int] | None
+ _execution_time: int | None
is_swappable = True
def __init__(
self,
- src0: Optional[SignalSourceProvider] = None,
- src1: Optional[SignalSourceProvider] = None,
- src2: Optional[SignalSourceProvider] = None,
+ src0: SignalSourceProvider | None = None,
+ src1: SignalSourceProvider | None = None,
+ src2: SignalSourceProvider | None = None,
name: Name = Name(""),
- latency: Optional[int] = None,
- latency_offsets: Optional[Dict[str, int]] = None,
- execution_time: Optional[int] = None,
+ latency: int | None = None,
+ latency_offsets: dict[str, int] | None = None,
+ execution_time: int | None = None,
):
"""Construct a MAD operation."""
super().__init__(
@@ -1142,29 +1140,29 @@ class MADS(AbstractOperation):
"_latency_offsets",
"_execution_time",
)
- _is_add: Optional[bool]
- _override_zero_on_src0: Optional[bool]
- _src0: Optional[SignalSourceProvider]
- _src1: Optional[SignalSourceProvider]
- _src2: Optional[SignalSourceProvider]
+ _is_add: bool | None
+ _override_zero_on_src0: bool | None
+ _src0: SignalSourceProvider | None
+ _src1: SignalSourceProvider | None
+ _src2: SignalSourceProvider | None
_name: Name
- _latency: Optional[int]
- _latency_offsets: Optional[Dict[str, int]]
- _execution_time: Optional[int]
+ _latency: int | None
+ _latency_offsets: dict[str, int] | None
+ _execution_time: int | None
is_swappable = True
def __init__(
self,
- is_add: Optional[bool] = True,
- override_zero_on_src0: Optional[bool] = False,
- src0: Optional[SignalSourceProvider] = None,
- src1: Optional[SignalSourceProvider] = None,
- src2: Optional[SignalSourceProvider] = None,
+ is_add: bool | None = True,
+ override_zero_on_src0: bool | None = False,
+ src0: SignalSourceProvider | None = None,
+ src1: SignalSourceProvider | None = None,
+ src2: SignalSourceProvider | None = None,
name: Name = Name(""),
- latency: Optional[int] = None,
- latency_offsets: Optional[Dict[str, int]] = None,
- execution_time: Optional[int] = None,
+ latency: int | None = None,
+ latency_offsets: dict[str, int] | None = None,
+ execution_time: int | None = None,
):
"""Construct a MADS operation."""
super().__init__(
@@ -1251,12 +1249,12 @@ class SymmetricTwoportAdaptor(AbstractOperation):
"_latency_offsets",
"_execution_time",
)
- _src0: Optional[SignalSourceProvider]
- _src1: Optional[SignalSourceProvider]
+ _src0: SignalSourceProvider | None
+ _src1: SignalSourceProvider | None
_name: Name
- _latency: Optional[int]
- _latency_offsets: Optional[Dict[str, int]]
- _execution_time: Optional[int]
+ _latency: int | None
+ _latency_offsets: dict[str, int] | None
+ _execution_time: int | None
is_linear = True
is_swappable = True
@@ -1264,12 +1262,12 @@ class SymmetricTwoportAdaptor(AbstractOperation):
def __init__(
self,
value: Num = 0,
- src0: Optional[SignalSourceProvider] = None,
- src1: Optional[SignalSourceProvider] = None,
+ src0: SignalSourceProvider | None = None,
+ src1: SignalSourceProvider | None = None,
name: Name = Name(""),
- latency: Optional[int] = None,
- latency_offsets: Optional[Dict[str, int]] = None,
- execution_time: Optional[int] = None,
+ latency: int | None = None,
+ latency_offsets: dict[str, int] | None = None,
+ execution_time: int | None = None,
):
"""Construct a SymmetricTwoportAdaptor operation."""
super().__init__(
@@ -1344,19 +1342,19 @@ class Reciprocal(AbstractOperation):
"""
__slots__ = ("_src0", "_name", "_latency", "_latency_offsets", "_execution_time")
- _src0: Optional[SignalSourceProvider]
+ _src0: SignalSourceProvider | None
_name: Name
- _latency: Optional[int]
- _latency_offsets: Optional[Dict[str, int]]
- _execution_time: Optional[int]
+ _latency: int | None
+ _latency_offsets: dict[str, int] | None
+ _execution_time: int | None
def __init__(
self,
- src0: Optional[SignalSourceProvider] = None,
+ src0: SignalSourceProvider | None = None,
name: Name = Name(""),
- latency: Optional[int] = None,
- latency_offsets: Optional[Dict[str, int]] = None,
- execution_time: Optional[int] = None,
+ latency: int | None = None,
+ latency_offsets: dict[str, int] | None = None,
+ execution_time: int | None = None,
):
"""Construct a Reciprocal operation."""
super().__init__(
@@ -1418,22 +1416,22 @@ class RightShift(AbstractOperation):
"_execution_time",
)
_value: Num
- _src0: Optional[SignalSourceProvider]
+ _src0: SignalSourceProvider | None
_name: Name
- _latency: Optional[int]
- _latency_offsets: Optional[Dict[str, int]]
- _execution_time: Optional[int]
+ _latency: int | None
+ _latency_offsets: dict[str, int] | None
+ _execution_time: int | None
is_linear = True
def __init__(
self,
value: int = 0,
- src0: Optional[SignalSourceProvider] = None,
+ src0: SignalSourceProvider | None = None,
name: Name = Name(""),
- latency: Optional[int] = None,
- latency_offsets: Optional[Dict[str, int]] = None,
- execution_time: Optional[int] = None,
+ latency: int | None = None,
+ latency_offsets: dict[str, int] | None = None,
+ execution_time: int | None = None,
):
"""Construct a RightShift operation with the given value."""
super().__init__(
@@ -1510,22 +1508,22 @@ class LeftShift(AbstractOperation):
"_execution_time",
)
_value: Num
- _src0: Optional[SignalSourceProvider]
+ _src0: SignalSourceProvider | None
_name: Name
- _latency: Optional[int]
- _latency_offsets: Optional[Dict[str, int]]
- _execution_time: Optional[int]
+ _latency: int | None
+ _latency_offsets: dict[str, int] | None
+ _execution_time: int | None
is_linear = True
def __init__(
self,
value: int = 0,
- src0: Optional[SignalSourceProvider] = None,
+ src0: SignalSourceProvider | None = None,
name: Name = Name(""),
- latency: Optional[int] = None,
- latency_offsets: Optional[Dict[str, int]] = None,
- execution_time: Optional[int] = None,
+ latency: int | None = None,
+ latency_offsets: dict[str, int] | None = None,
+ execution_time: int | None = None,
):
"""Construct a RightShift operation with the given value."""
super().__init__(
@@ -1603,22 +1601,22 @@ class Shift(AbstractOperation):
"_execution_time",
)
_value: Num
- _src0: Optional[SignalSourceProvider]
+ _src0: SignalSourceProvider | None
_name: Name
- _latency: Optional[int]
- _latency_offsets: Optional[Dict[str, int]]
- _execution_time: Optional[int]
+ _latency: int | None
+ _latency_offsets: dict[str, int] | None
+ _execution_time: int | None
is_linear = True
def __init__(
self,
value: int = 0,
- src0: Optional[SignalSourceProvider] = None,
+ src0: SignalSourceProvider | None = None,
name: Name = Name(""),
- latency: Optional[int] = None,
- latency_offsets: Optional[Dict[str, int]] = None,
- execution_time: Optional[int] = None,
+ latency: int | None = None,
+ latency_offsets: dict[str, int] | None = None,
+ execution_time: int | None = None,
):
"""Construct a Shift operation with the given value."""
super().__init__(
diff --git a/b_asic/graph_component.py b/b_asic/graph_component.py
index 238c6ecd..2864904b 100644
--- a/b_asic/graph_component.py
+++ b/b_asic/graph_component.py
@@ -6,8 +6,9 @@ Contains the base for all components with an ID in a signal flow graph.
from abc import ABC, abstractmethod
from collections import deque
+from collections.abc import Generator, Iterable, Mapping
from copy import copy, deepcopy
-from typing import Any, Dict, Generator, Iterable, Mapping, cast
+from typing import Any, cast
from b_asic.types import GraphID, Name, TypeName
@@ -121,7 +122,7 @@ class AbstractGraphComponent(GraphComponent):
_name: Name
_graph_id: GraphID
- _parameters: Dict[str, Any]
+ _parameters: dict[str, Any]
def __init__(self, name: Name = Name("")):
"""Construct a graph component."""
diff --git a/b_asic/gui_utils/plot_window.py b/b_asic/gui_utils/plot_window.py
index 96c3bec5..4a9b0d58 100644
--- a/b_asic/gui_utils/plot_window.py
+++ b/b_asic/gui_utils/plot_window.py
@@ -2,7 +2,7 @@
import re
import sys
-from typing import Dict, List, Mapping, Optional, Sequence # , Union
+from collections.abc import Mapping, Sequence
# from numpy import (array, real, imag, real_if_close, absolute, angle)
import numpy as np
@@ -46,7 +46,7 @@ class PlotWindow(QWidget):
def __init__(
self,
sim_result: Mapping[ResultKey, Sequence[Num]],
- sfg_name: Optional[str] = None,
+ sfg_name: str | None = None,
):
super().__init__()
self.setWindowFlags(
@@ -237,7 +237,7 @@ class PlotWindow(QWidget):
def start_simulation_dialog(
- sim_results: Dict[str, List[complex]], sfg_name: Optional[str] = None
+ sim_results: dict[str, list[complex]], sfg_name: str | None = None
):
"""
Display the simulation results window.
diff --git a/b_asic/logger.py b/b_asic/logger.py
index 5e18464d..53b510e3 100644
--- a/b_asic/logger.py
+++ b/b_asic/logger.py
@@ -52,7 +52,6 @@ import os
import sys
from logging import Logger
from types import TracebackType
-from typing import Type, Union
def getLogger(source: str, filename: str, loglevel: str = "INFO") -> Logger:
@@ -124,9 +123,9 @@ def getLogger(source: str, filename: str, loglevel: str = "INFO") -> Logger:
# log uncaught exceptions
def handle_exceptions(
- exc_type: Type[BaseException],
+ exc_type: type[BaseException],
exc_value: BaseException,
- exc_traceback: Union[TracebackType, None],
+ exc_traceback: TracebackType | None,
) -> None:
# def log_exceptions(type, value, tb):
"""This function is a helper function to log uncaught exceptions. Install with:
diff --git a/b_asic/operation.py b/b_asic/operation.py
index 02995d3f..bc6bb748 100644
--- a/b_asic/operation.py
+++ b/b_asic/operation.py
@@ -8,19 +8,12 @@ import collections
import collections.abc
import itertools as it
from abc import abstractmethod
+from collections.abc import Iterable, Mapping, MutableMapping, Sequence
from numbers import Number
from typing import (
TYPE_CHECKING,
- Dict,
- Iterable,
- List,
- Mapping,
- MutableMapping,
NewType,
Optional,
- Sequence,
- Tuple,
- Union,
cast,
overload,
)
@@ -127,8 +120,8 @@ class Operation(GraphComponent, SignalSourceProvider):
@abstractmethod
def current_output(
- self, index: int, delays: Optional[DelayMap] = None, prefix: str = ""
- ) -> Optional[Num]:
+ self, index: int, delays: DelayMap | None = None, prefix: str = ""
+ ) -> Num | None:
"""
Get the current output at the given index of this operation, if available.
@@ -147,10 +140,10 @@ class Operation(GraphComponent, SignalSourceProvider):
self,
index: int,
input_values: Sequence[Num],
- results: Optional[MutableResultMap] = None,
- delays: Optional[MutableDelayMap] = None,
+ results: MutableResultMap | None = None,
+ delays: MutableDelayMap | None = None,
prefix: str = "",
- bits_override: Optional[int] = None,
+ bits_override: int | None = None,
quantize: bool = True,
) -> Num:
"""
@@ -186,8 +179,8 @@ class Operation(GraphComponent, SignalSourceProvider):
@abstractmethod
def current_outputs(
- self, delays: Optional[DelayMap] = None, prefix: str = ""
- ) -> Sequence[Optional[Num]]:
+ self, delays: DelayMap | None = None, prefix: str = ""
+ ) -> Sequence[Num | None]:
"""
Get all current outputs of this operation, if available.
@@ -201,10 +194,10 @@ class Operation(GraphComponent, SignalSourceProvider):
def evaluate_outputs(
self,
input_values: Sequence[Num],
- results: Optional[MutableResultMap] = None,
- delays: Optional[MutableDelayMap] = None,
+ results: MutableResultMap | None = None,
+ delays: MutableDelayMap | None = None,
prefix: str = "",
- bits_override: Optional[int] = None,
+ bits_override: int | None = None,
quantize: bool = True,
) -> Sequence[Num]:
"""
@@ -265,7 +258,7 @@ class Operation(GraphComponent, SignalSourceProvider):
@property
@abstractmethod
- def latency_offsets(self) -> Dict[str, Optional[int]]:
+ def latency_offsets(self) -> dict[str, int | None]:
"""
Get a dictionary with all the operations ports latency-offsets.
"""
@@ -292,7 +285,7 @@ class Operation(GraphComponent, SignalSourceProvider):
raise NotImplementedError
@abstractmethod
- def set_latency_offsets(self, latency_offsets: Dict[str, int]) -> None:
+ def set_latency_offsets(self, latency_offsets: dict[str, int]) -> None:
"""
Set the latency-offsets for the operations port.
@@ -304,7 +297,7 @@ class Operation(GraphComponent, SignalSourceProvider):
@property
@abstractmethod
- def execution_time(self) -> Optional[int]:
+ def execution_time(self) -> int | None:
"""
Get the execution time of the operation.
@@ -315,7 +308,7 @@ class Operation(GraphComponent, SignalSourceProvider):
@execution_time.setter
@abstractmethod
- def execution_time(self, execution_time: Optional[int]) -> None:
+ def execution_time(self, execution_time: int | None) -> None:
"""
Set the execution time of the operation.
@@ -335,7 +328,7 @@ class Operation(GraphComponent, SignalSourceProvider):
@abstractmethod
def get_plot_coordinates(
self,
- ) -> Tuple[Tuple[Tuple[float, float], ...], Tuple[Tuple[float, float], ...]]:
+ ) -> tuple[tuple[tuple[float, float], ...], tuple[tuple[float, float], ...]]:
"""
Return coordinates for the latency and execution time polygons.
@@ -348,7 +341,7 @@ class Operation(GraphComponent, SignalSourceProvider):
@abstractmethod
def get_input_coordinates(
self,
- ) -> Tuple[Tuple[float, float], ...]:
+ ) -> tuple[tuple[float, float], ...]:
"""
Return coordinates for inputs.
@@ -364,7 +357,7 @@ class Operation(GraphComponent, SignalSourceProvider):
@abstractmethod
def get_output_coordinates(
self,
- ) -> Tuple[Tuple[float, float], ...]:
+ ) -> tuple[tuple[float, float], ...]:
"""
Return coordinates for outputs.
@@ -455,19 +448,19 @@ class AbstractOperation(Operation, AbstractGraphComponent):
"""
__slots__ = ("_input_ports", "_output_ports", "_execution_time")
- _input_ports: List[InputPort]
- _output_ports: List[OutputPort]
- _execution_time: Optional[int]
+ _input_ports: list[InputPort]
+ _output_ports: list[OutputPort]
+ _execution_time: int | None
def __init__(
self,
input_count: int,
output_count: int,
name: Name = Name(""),
- input_sources: Optional[Sequence[Optional[SignalSourceProvider]]] = None,
- latency: Optional[int] = None,
- latency_offsets: Optional[Dict[str, int]] = None,
- execution_time: Optional[int] = None,
+ input_sources: Sequence[SignalSourceProvider | None] | None = None,
+ latency: int | None = None,
+ latency_offsets: dict[str, int] | None = None,
+ execution_time: int | None = None,
):
"""
Construct an operation with the given input/output count.
@@ -520,12 +513,12 @@ class AbstractOperation(Operation, AbstractGraphComponent):
@abstractmethod
def evaluate(
self, *inputs: Operation
- ) -> List[Operation]: # pylint: disable=arguments-differ
+ ) -> list[Operation]: # pylint: disable=arguments-differ
...
@overload
@abstractmethod
- def evaluate(self, *inputs: Num) -> List[Num]: # pylint: disable=arguments-differ
+ def evaluate(self, *inputs: Num) -> list[Num]: # pylint: disable=arguments-differ
...
@abstractmethod
@@ -552,7 +545,7 @@ class AbstractOperation(Operation, AbstractGraphComponent):
def __str__(self) -> str:
"""Get a string representation of this operation."""
- inputs_dict: Dict[int, Union[List[GraphID], str]] = {}
+ inputs_dict: dict[int, list[GraphID] | str] = {}
for i, current_input in enumerate(self.inputs):
if current_input.signal_count == 0:
inputs_dict[i] = "-"
@@ -571,7 +564,7 @@ class AbstractOperation(Operation, AbstractGraphComponent):
dict_ele.append(GraphID("no_id"))
inputs_dict[i] = dict_ele
- outputs_dict: Dict[int, Union[List[GraphID], str]] = {}
+ outputs_dict: dict[int, list[GraphID] | str] = {}
for i, outport in enumerate(self.outputs):
if outport.signal_count == 0:
outputs_dict[i] = "-"
@@ -644,18 +637,18 @@ class AbstractOperation(Operation, AbstractGraphComponent):
return ResultKey(key)
def current_output(
- self, index: int, delays: Optional[DelayMap] = None, prefix: str = ""
- ) -> Optional[Num]:
+ self, index: int, delays: DelayMap | None = None, prefix: str = ""
+ ) -> Num | None:
return None
def evaluate_output(
self,
index: int,
input_values: Sequence[Num],
- results: Optional[MutableResultMap] = None,
- delays: Optional[MutableDelayMap] = None,
+ results: MutableResultMap | None = None,
+ delays: MutableDelayMap | None = None,
prefix: str = "",
- bits_override: Optional[int] = None,
+ bits_override: int | None = None,
quantize: bool = True,
) -> Num:
if index < 0 or index >= self.output_count:
@@ -701,8 +694,8 @@ class AbstractOperation(Operation, AbstractGraphComponent):
return values[index]
def current_outputs(
- self, delays: Optional[DelayMap] = None, prefix: str = ""
- ) -> Sequence[Optional[Num]]:
+ self, delays: DelayMap | None = None, prefix: str = ""
+ ) -> Sequence[Num | None]:
return [
self.current_output(i, delays, prefix) for i in range(self.output_count)
]
@@ -710,10 +703,10 @@ class AbstractOperation(Operation, AbstractGraphComponent):
def evaluate_outputs(
self,
input_values: Sequence[Num],
- results: Optional[MutableResultMap] = None,
- delays: Optional[MutableDelayMap] = None,
+ results: MutableResultMap | None = None,
+ delays: MutableDelayMap | None = None,
prefix: str = "",
- bits_override: Optional[int] = None,
+ bits_override: int | None = None,
quantize: bool = True,
) -> Sequence[Num]:
return [
@@ -737,7 +730,7 @@ class AbstractOperation(Operation, AbstractGraphComponent):
if isinstance(result, collections.abc.Sequence) and all(
isinstance(e, Operation) for e in result
):
- return cast(List[Operation], result)
+ return cast(list[Operation], result)
return [self]
def to_sfg(self) -> "SFG":
@@ -838,7 +831,7 @@ class AbstractOperation(Operation, AbstractGraphComponent):
def quantize_inputs(
self,
input_values: Sequence[Num],
- bits_override: Optional[int] = None,
+ bits_override: int | None = None,
) -> Sequence[Num]:
"""
Quantize the values to be used as inputs.
@@ -879,7 +872,7 @@ class AbstractOperation(Operation, AbstractGraphComponent):
)
@property
- def latency_offsets(self) -> Dict[str, Optional[int]]:
+ def latency_offsets(self) -> dict[str, int | None]:
latency_offsets = {}
for i, input_ in enumerate(self.inputs):
@@ -897,7 +890,7 @@ class AbstractOperation(Operation, AbstractGraphComponent):
self.input_latency_offsets()
self.output_latency_offsets()
- def input_latency_offsets(self) -> List[int]:
+ def input_latency_offsets(self) -> list[int]:
latency_offsets = [i.latency_offset for i in self.inputs]
if any(val is None for val in latency_offsets):
@@ -906,9 +899,9 @@ class AbstractOperation(Operation, AbstractGraphComponent):
]
raise ValueError(f"Missing latencies for input(s) {missing}")
- return cast(List[int], latency_offsets)
+ return cast(list[int], latency_offsets)
- def output_latency_offsets(self) -> List[int]:
+ def output_latency_offsets(self) -> list[int]:
latency_offsets = [i.latency_offset for i in self.outputs]
if any(val is None for val in latency_offsets):
@@ -917,7 +910,7 @@ class AbstractOperation(Operation, AbstractGraphComponent):
]
raise ValueError(f"Missing latencies for output(s) {missing}")
- return cast(List[int], latency_offsets)
+ return cast(list[int], latency_offsets)
def set_latency(self, latency: int) -> None:
if latency < 0:
@@ -927,7 +920,7 @@ class AbstractOperation(Operation, AbstractGraphComponent):
for outport in self.outputs:
outport.latency_offset = latency
- def set_latency_offsets(self, latency_offsets: Dict[str, int]) -> None:
+ def set_latency_offsets(self, latency_offsets: dict[str, int]) -> None:
for port_str, latency_offset in latency_offsets.items():
port_str = port_str.lower()
if port_str.startswith("in"):
@@ -953,7 +946,7 @@ class AbstractOperation(Operation, AbstractGraphComponent):
)
@property
- def execution_time(self) -> Optional[int]:
+ def execution_time(self) -> int | None:
"""Execution time of operation."""
return self._execution_time
@@ -979,7 +972,7 @@ class AbstractOperation(Operation, AbstractGraphComponent):
def get_plot_coordinates(
self,
- ) -> Tuple[Tuple[Tuple[float, float], ...], Tuple[Tuple[float, float], ...]]:
+ ) -> tuple[tuple[tuple[float, float], ...], tuple[tuple[float, float], ...]]:
# Doc-string inherited
return (
self._get_plot_coordinates_for_latency(),
@@ -988,7 +981,7 @@ class AbstractOperation(Operation, AbstractGraphComponent):
def _get_plot_coordinates_for_execution_time(
self,
- ) -> Tuple[Tuple[float, float], ...]:
+ ) -> tuple[tuple[float, float], ...]:
# Always a rectangle, but easier if coordinates are returned
execution_time = self._execution_time # Copy for type checking
if execution_time is None:
@@ -1003,7 +996,7 @@ class AbstractOperation(Operation, AbstractGraphComponent):
def _get_plot_coordinates_for_latency(
self,
- ) -> Tuple[Tuple[float, float], ...]:
+ ) -> tuple[tuple[float, float], ...]:
# Points for latency polygon
latency = []
input_latencies = self.input_latency_offsets()
@@ -1028,7 +1021,7 @@ class AbstractOperation(Operation, AbstractGraphComponent):
return tuple(latency)
- def get_input_coordinates(self) -> Tuple[Tuple[float, float], ...]:
+ def get_input_coordinates(self) -> tuple[tuple[float, float], ...]:
# doc-string inherited
num_in = self.input_count
return tuple(
@@ -1039,7 +1032,7 @@ class AbstractOperation(Operation, AbstractGraphComponent):
for k in range(num_in)
)
- def get_output_coordinates(self) -> Tuple[Tuple[float, float], ...]:
+ def get_output_coordinates(self) -> tuple[tuple[float, float], ...]:
# doc-string inherited
num_out = self.output_count
return tuple(
diff --git a/b_asic/port.py b/b_asic/port.py
index b3b3ccfb..65df10d3 100644
--- a/b_asic/port.py
+++ b/b_asic/port.py
@@ -5,9 +5,10 @@ Contains classes for managing the ports of operations.
"""
from abc import ABC, abstractmethod
+from collections.abc import Sequence
from copy import copy
from numbers import Number
-from typing import TYPE_CHECKING, List, Optional, Sequence, Union
+from typing import TYPE_CHECKING, Optional, Union
from b_asic.graph_component import Name
from b_asic.signal import Signal
@@ -55,7 +56,7 @@ class Port(ABC):
@property
@abstractmethod
- def latency_offset(self) -> Optional[int]:
+ def latency_offset(self) -> int | None:
"""Get the latency_offset of the port."""
raise NotImplementedError
@@ -133,13 +134,13 @@ class AbstractPort(Port):
__slots__ = ("_operation", "_index", "_latency_offset")
_operation: "Operation"
_index: int
- _latency_offset: Optional[int]
+ _latency_offset: int | None
def __init__(
self,
operation: "Operation",
index: int,
- latency_offset: Optional[int] = None,
+ latency_offset: int | None = None,
):
"""Construct a port of the given operation at the given port index."""
self._operation = operation
@@ -155,11 +156,11 @@ class AbstractPort(Port):
return self._index
@property
- def latency_offset(self) -> Optional[int]:
+ def latency_offset(self) -> int | None:
return self._latency_offset
@latency_offset.setter
- def latency_offset(self, latency_offset: Optional[int]):
+ def latency_offset(self, latency_offset: int | None):
self._latency_offset = latency_offset
@property
@@ -278,7 +279,7 @@ class InputPort(AbstractPort):
"""
__slots__ = ("_source_signal",)
- _source_signal: Optional[Signal]
+ _source_signal: Signal | None
def __init__(self, operation: "Operation", index: int):
"""Construct an InputPort."""
@@ -375,7 +376,7 @@ class OutputPort(AbstractPort, SignalSourceProvider):
"""
__slots__ = ("_destination_signals",)
- _destination_signals: List[Signal]
+ _destination_signals: list[Signal]
def __init__(self, operation: "Operation", index: int):
"""Construct an OutputPort."""
diff --git a/b_asic/process.py b/b_asic/process.py
index e51fc31e..0a5d7fa7 100644
--- a/b_asic/process.py
+++ b/b_asic/process.py
@@ -1,6 +1,6 @@
"""B-ASIC classes representing resource usage."""
-from typing import Any, Dict, List, Optional, Tuple, cast
+from typing import Any, Optional, cast
from b_asic.operation import Operation
from b_asic.port import InputPort, OutputPort
@@ -69,7 +69,7 @@ class Process:
return f"Process({self.start_time}, {self.execution_time}, {self.name!r})"
@property
- def read_times(self) -> Tuple[int, ...]:
+ def read_times(self) -> tuple[int, ...]:
return (self.start_time + self.execution_time,)
@@ -94,7 +94,7 @@ class OperatorProcess(Process):
self,
start_time: int,
operation: Operation,
- name: Optional[str] = None,
+ name: str | None = None,
):
execution_time = operation.execution_time
if execution_time is None:
@@ -140,12 +140,12 @@ class MemoryProcess(Process):
"""
__slots__ = ("_life_times",)
- _life_times: List[int]
+ _life_times: list[int]
def __init__(
self,
write_time: int,
- life_times: List[int],
+ life_times: list[int],
name: str = "",
):
pass
@@ -157,19 +157,19 @@ class MemoryProcess(Process):
)
@property
- def read_times(self) -> Tuple[int, ...]:
+ def read_times(self) -> tuple[int, ...]:
return tuple(self.start_time + read for read in self._life_times)
@property
- def life_times(self) -> List[int]:
+ def life_times(self) -> list[int]:
return self._life_times
@property
- def reads(self) -> Dict[Any, int]:
+ def reads(self) -> dict[Any, int]:
raise NotImplementedError("MultiReadProcess should be derived from")
@property
- def read_ports(self) -> List[Any]:
+ def read_ports(self) -> list[Any]:
raise NotImplementedError("MultiReadProcess should be derived from")
@property
@@ -179,7 +179,7 @@ class MemoryProcess(Process):
def split_on_length(
self,
length: int = 0,
- ) -> Tuple[Optional["MemoryProcess"], Optional["MemoryProcess"]]:
+ ) -> tuple[Optional["MemoryProcess"], Optional["MemoryProcess"]]:
"""
Split this :class:`MemoryProcess` into two new :class:`MemoryProcess` objects.
@@ -286,16 +286,16 @@ class MemoryVariable(MemoryProcess):
"""
__slots__ = ("_reads", "_read_ports", "_write_port")
- _reads: Dict[InputPort, int]
- _read_ports: List[InputPort]
+ _reads: dict[InputPort, int]
+ _read_ports: list[InputPort]
_write_port: OutputPort
def __init__(
self,
write_time: int,
write_port: OutputPort,
- reads: Dict[InputPort, int],
- name: Optional[str] = None,
+ reads: dict[InputPort, int],
+ name: str | None = None,
):
self._read_ports = list(reads.keys())
self._reads = reads
@@ -307,11 +307,11 @@ class MemoryVariable(MemoryProcess):
)
@property
- def reads(self) -> Dict[InputPort, int]:
+ def reads(self) -> dict[InputPort, int]:
return self._reads
@property
- def read_ports(self) -> List[InputPort]:
+ def read_ports(self) -> list[InputPort]:
return self._read_ports
@property
@@ -328,7 +328,7 @@ class MemoryVariable(MemoryProcess):
def split_on_length(
self,
length: int = 0,
- ) -> Tuple[Optional["MemoryVariable"], Optional["MemoryVariable"]]:
+ ) -> tuple[Optional["MemoryVariable"], Optional["MemoryVariable"]]:
"""
Split this :class:`MemoryVariable` into two new :class:`MemoryVariable` objects,
based on lifetimes of read accesses.
@@ -376,16 +376,16 @@ class PlainMemoryVariable(MemoryProcess):
"""
__slots__ = ("_reads", "_read_ports", "_write_port")
- _reads: Dict[int, int]
- _read_ports: List[int]
+ _reads: dict[int, int]
+ _read_ports: list[int]
_write_port: OutputPort
def __init__(
self,
write_time: int,
write_port: int,
- reads: Dict[int, int],
- name: Optional[str] = None,
+ reads: dict[int, int],
+ name: str | None = None,
):
self._read_ports = list(reads.keys())
self._write_port = write_port
@@ -401,11 +401,11 @@ class PlainMemoryVariable(MemoryProcess):
)
@property
- def reads(self) -> Dict[int, int]:
+ def reads(self) -> dict[int, int]:
return self._reads
@property
- def read_ports(self) -> List[int]:
+ def read_ports(self) -> list[int]:
return self._read_ports
@property
@@ -422,7 +422,7 @@ class PlainMemoryVariable(MemoryProcess):
def split_on_length(
self,
length: int = 0,
- ) -> Tuple[Optional["PlainMemoryVariable"], Optional["PlainMemoryVariable"]]:
+ ) -> tuple[Optional["PlainMemoryVariable"], Optional["PlainMemoryVariable"]]:
"""
Split this :class:`PlainMemoryVariable` into two new
:class:`PlainMemoryVariable` objects, based on lifetimes of read accesses.
diff --git a/b_asic/research/interleaver.py b/b_asic/research/interleaver.py
index 3bb7a4ba..86b81491 100644
--- a/b_asic/research/interleaver.py
+++ b/b_asic/research/interleaver.py
@@ -4,19 +4,18 @@ Functions to generate memory-variable test data that are used for research.
import random
from itertools import product
-from typing import List, Optional, Tuple
from b_asic.process import PlainMemoryVariable
from b_asic.resources import ProcessCollection
def _insert_delays(
- inputorder: List[Tuple[int, int]],
- outputorder: List[Tuple[int, int]],
+ inputorder: list[tuple[int, int]],
+ outputorder: list[tuple[int, int]],
min_lifetime: int,
cyclic: bool,
time: int,
-) -> Tuple[List[Tuple[int, int]], List[Tuple[int, int]]]:
+) -> tuple[list[tuple[int, int]], list[tuple[int, int]]]:
size = len(inputorder)
maxdiff = min(outputorder[i][0] - inputorder[i][0] for i in range(size))
outputorder = [(o[0] - maxdiff + min_lifetime, o[1]) for o in outputorder]
@@ -71,7 +70,7 @@ def generate_random_interleaver(
def generate_matrix_transposer(
rows: int,
- cols: Optional[int] = None,
+ cols: int | None = None,
min_lifetime: int = 0,
cyclic: bool = True,
parallelism: int = 1,
diff --git a/b_asic/resources.py b/b_asic/resources.py
index 597e7f29..e54322bc 100644
--- a/b_asic/resources.py
+++ b/b_asic/resources.py
@@ -1,9 +1,10 @@
import io
import re
from collections import Counter, defaultdict
+from collections.abc import Iterable
from functools import reduce
from math import log2
-from typing import Dict, Iterable, List, Literal, Optional, Tuple, TypeVar, Union
+from typing import Literal, TypeVar
import matplotlib.pyplot as plt
import networkx as nx
@@ -34,7 +35,7 @@ _WARNING_COLOR = tuple(c / 255 for c in WARNING_COLOR)
_T = TypeVar('_T')
-def _sorted_nicely(to_be_sorted: Iterable[_T]) -> List[_T]:
+def _sorted_nicely(to_be_sorted: Iterable[_T]) -> list[_T]:
"""Sort the given iterable in the way that humans expect."""
def convert(text):
@@ -47,10 +48,10 @@ def _sorted_nicely(to_be_sorted: Iterable[_T]) -> List[_T]:
def _sanitize_port_option(
- read_ports: Optional[int] = None,
- write_ports: Optional[int] = None,
- total_ports: Optional[int] = None,
-) -> Tuple[int, int, int]:
+ read_ports: int | None = None,
+ write_ports: int | None = None,
+ total_ports: int | None = None,
+) -> tuple[int, int, int]:
"""
General port sanitization function used to test if a port specification makes sense.
Raises ValueError if the port specification is in-proper.
@@ -95,9 +96,9 @@ def _sanitize_port_option(
def draw_exclusion_graph_coloring(
exclusion_graph: nx.Graph,
- color_dict: Dict[Process, int],
- ax: Optional[Axes] = None,
- color_list: Optional[Union[List[str], List[Tuple[float, float, float]]]] = None,
+ color_dict: dict[Process, int],
+ ax: Axes | None = None,
+ color_list: list[str] | list[tuple[float, float, float]] | None = None,
**kwargs,
) -> None:
"""
@@ -176,12 +177,12 @@ def draw_exclusion_graph_coloring(
class _ForwardBackwardEntry:
def __init__(
self,
- inputs: Optional[List[Process]] = None,
- outputs: Optional[List[Process]] = None,
- regs: Optional[List[Optional[Process]]] = None,
- back_edge_to: Optional[Dict[int, int]] = None,
- back_edge_from: Optional[Dict[int, int]] = None,
- outputs_from: Optional[int] = None,
+ inputs: list[Process] | None = None,
+ outputs: list[Process] | None = None,
+ regs: list[Process | None] | None = None,
+ back_edge_to: dict[int, int] | None = None,
+ back_edge_from: dict[int, int] | None = None,
+ outputs_from: int | None = None,
):
"""
Single entry in a _ForwardBackwardTable.
@@ -204,11 +205,11 @@ class _ForwardBackwardEntry:
this entry.
outputs_from : int, optional
"""
- self.inputs: List[Process] = [] if inputs is None else inputs
- self.outputs: List[Process] = [] if outputs is None else outputs
- self.regs: List[Optional[Process]] = [] if regs is None else regs
- self.back_edge_to: Dict[int, int] = {} if back_edge_to is None else back_edge_to
- self.back_edge_from: Dict[int, int] = (
+ self.inputs: list[Process] = [] if inputs is None else inputs
+ self.outputs: list[Process] = [] if outputs is None else outputs
+ self.regs: list[Process | None] = [] if regs is None else regs
+ self.back_edge_to: dict[int, int] = {} if back_edge_to is None else back_edge_to
+ self.back_edge_from: dict[int, int] = (
{} if back_edge_from is None else back_edge_from
)
self.outputs_from = outputs_from
@@ -229,14 +230,14 @@ class _ForwardBackwardTable:
"""
# Generate an alive variable list
self._collection = set(collection.collection)
- self._live_variables: List[int] = [0] * collection.schedule_time
+ self._live_variables: list[int] = [0] * collection.schedule_time
for mv in self._collection:
stop_time = mv.start_time + mv.execution_time
for alive_time in range(mv.start_time, stop_time):
self._live_variables[alive_time % collection.schedule_time] += 1
# First, create an empty forward-backward table with the right dimensions
- self.table: List[_ForwardBackwardEntry] = []
+ self.table: list[_ForwardBackwardEntry] = []
for _ in range(collection.schedule_time):
entry = _ForwardBackwardEntry()
# https://github.com/microsoft/pyright/issues/1073
@@ -450,7 +451,7 @@ class ProcessCollection:
"""
__slots__ = ("_collection", "_schedule_time", "_cyclic")
- _collection: List[Process]
+ _collection: list[Process]
_schedule_time: int
_cyclic: bool
@@ -465,7 +466,7 @@ class ProcessCollection:
self._cyclic = cyclic
@property
- def collection(self) -> List[Process]:
+ def collection(self) -> list[Process]:
return self._collection
@property
@@ -520,15 +521,15 @@ class ProcessCollection:
def plot(
self,
- ax: Optional[Axes] = None,
+ ax: Axes | None = None,
*,
show_name: bool = True,
- bar_color: Union[str, Tuple[float, ...]] = _LATENCY_COLOR,
- marker_color: Union[str, Tuple[float, ...]] = "black",
+ bar_color: str | tuple[float, ...] = _LATENCY_COLOR,
+ marker_color: str | tuple[float, ...] = "black",
marker_read: str = "X",
marker_write: str = "o",
show_markers: bool = True,
- row: Optional[int] = None,
+ row: int | None = None,
allow_excessive_lifetimes: bool = False,
):
"""
@@ -689,13 +690,13 @@ class ProcessCollection:
self,
*,
show_name: bool = True,
- bar_color: Union[str, Tuple[float, ...]] = _LATENCY_COLOR,
- marker_color: Union[str, Tuple[float, ...]] = "black",
+ bar_color: str | tuple[float, ...] = _LATENCY_COLOR,
+ marker_color: str | tuple[float, ...] = "black",
marker_read: str = "X",
marker_write: str = "o",
show_markers: bool = True,
allow_excessive_lifetimes: bool = False,
- title: Optional[str] = None,
+ title: str | None = None,
) -> None:
"""
Display lifetime diagram using the current Matplotlib backend.
@@ -740,9 +741,9 @@ class ProcessCollection:
def create_exclusion_graph_from_ports(
self,
- read_ports: Optional[int] = None,
- write_ports: Optional[int] = None,
- total_ports: Optional[int] = None,
+ read_ports: int | None = None,
+ write_ports: int | None = None,
+ total_ports: int | None = None,
) -> nx.Graph:
"""
Create an exclusion graph based on concurrent read and write accesses.
@@ -871,7 +872,7 @@ class ProcessCollection:
self,
heuristic: Literal["graph_color", "left_edge"] = "left_edge",
coloring_strategy: str = "saturation_largest_first",
- ) -> List["ProcessCollection"]:
+ ) -> list["ProcessCollection"]:
"""
Split based on overlapping execution time.
@@ -908,10 +909,10 @@ class ProcessCollection:
def split_on_ports(
self,
heuristic: str = "left_edge",
- read_ports: Optional[int] = None,
- write_ports: Optional[int] = None,
- total_ports: Optional[int] = None,
- ) -> List["ProcessCollection"]:
+ read_ports: int | None = None,
+ write_ports: int | None = None,
+ total_ports: int | None = None,
+ ) -> list["ProcessCollection"]:
"""
Split based on concurrent read and write accesses.
@@ -962,8 +963,8 @@ class ProcessCollection:
read_ports: int,
write_ports: int,
total_ports: int,
- sequence: List[Process],
- ) -> List["ProcessCollection"]:
+ sequence: list[Process],
+ ) -> list["ProcessCollection"]:
"""
Split this collection into multiple new collections by sequentially assigning
processes in the order of `sequence`.
@@ -1027,7 +1028,7 @@ class ProcessCollection:
if set(self.collection) != set(sequence):
raise KeyError("processes in `sequence` must be equal to processes in self")
- collections: List[ProcessCollection] = []
+ collections: list[ProcessCollection] = []
for process in sequence:
process_added = False
for collection in collections:
@@ -1053,7 +1054,7 @@ class ProcessCollection:
write_ports: int,
total_ports: int,
coloring_strategy: str = "saturation_largest_first",
- ) -> List["ProcessCollection"]:
+ ) -> list["ProcessCollection"]:
"""
Parameters
----------
@@ -1089,8 +1090,8 @@ class ProcessCollection:
def _split_from_graph_coloring(
self,
- coloring: Dict[Process, int],
- ) -> List["ProcessCollection"]:
+ coloring: dict[Process, int],
+ ) -> list["ProcessCollection"]:
"""
Split :class:`Process` objects into a set of :class:`ProcessesCollection`
objects based on a provided graph coloring.
@@ -1143,8 +1144,8 @@ class ProcessCollection:
self,
coloring_strategy: str = "saturation_largest_first",
*,
- coloring: Optional[Dict[Process, int]] = None,
- ) -> List["ProcessCollection"]:
+ coloring: dict[Process, int] | None = None,
+ ) -> list["ProcessCollection"]:
"""
Perform assignment of the processes in this collection using graph coloring.
@@ -1172,7 +1173,7 @@ class ProcessCollection:
f"{process} has execution time greater than the schedule time"
)
- cell_assignment: Dict[int, ProcessCollection] = dict()
+ cell_assignment: dict[int, ProcessCollection] = dict()
exclusion_graph = self.create_exclusion_graph_from_execution_time()
if coloring is None:
coloring = nx.coloring.greedy_color(
@@ -1184,7 +1185,7 @@ class ProcessCollection:
cell_assignment[cell].add_process(process)
return list(cell_assignment.values())
- def _left_edge_assignment(self) -> List["ProcessCollection"]:
+ def _left_edge_assignment(self) -> list["ProcessCollection"]:
"""
Perform assignment of the processes in this collection using the left-edge
algorithm.
@@ -1199,7 +1200,7 @@ class ProcessCollection:
-------
List[ProcessCollection]
"""
- assignment: List[ProcessCollection] = []
+ assignment: list[ProcessCollection] = []
for next_process in sorted(self):
if next_process.execution_time > self.schedule_time:
# Can not assign process to any cell
@@ -1256,14 +1257,14 @@ class ProcessCollection:
filename: str,
entity_name: str,
word_length: int,
- assignment: List['ProcessCollection'],
+ assignment: list['ProcessCollection'],
read_ports: int = 1,
write_ports: int = 1,
total_ports: int = 2,
*,
input_sync: bool = True,
- adr_mux_size: Optional[int] = None,
- adr_pipe_depth: Optional[int] = None,
+ adr_mux_size: int | None = None,
+ adr_pipe_depth: int | None = None,
):
"""
Generate VHDL code for memory based storage of processes (MemoryVariables).
@@ -1411,7 +1412,7 @@ class ProcessCollection:
def split_on_length(
self, length: int = 0
- ) -> Tuple["ProcessCollection", "ProcessCollection"]:
+ ) -> tuple["ProcessCollection", "ProcessCollection"]:
"""
Split into two ProcessCollections based on execution time length.
@@ -1567,7 +1568,7 @@ class ProcessCollection:
"""
return max(self.read_port_accesses().values())
- def read_port_accesses(self) -> Dict[int, int]:
+ def read_port_accesses(self) -> dict[int, int]:
reads = sum(
(
[read_time % self.schedule_time for read_time in process.read_times]
@@ -1589,7 +1590,7 @@ class ProcessCollection:
"""
return max(self.write_port_accesses().values())
- def write_port_accesses(self) -> Dict[int, int]:
+ def write_port_accesses(self) -> dict[int, int]:
writes = [
process.start_time % self.schedule_time for process in self._collection
]
@@ -1607,7 +1608,7 @@ class ProcessCollection:
"""
return max(self.total_port_accesses().values())
- def total_port_accesses(self) -> Dict[int, int]:
+ def total_port_accesses(self) -> dict[int, int]:
accesses = sum(
(
list(read_time % self.schedule_time for read_time in process.read_times)
diff --git a/b_asic/save_load_structure.py b/b_asic/save_load_structure.py
index 8f9622d4..34d84498 100644
--- a/b_asic/save_load_structure.py
+++ b/b_asic/save_load_structure.py
@@ -7,7 +7,7 @@ stored as files.
from datetime import datetime
from inspect import signature
-from typing import Dict, Optional, Tuple, cast
+from typing import cast
from b_asic.graph_component import GraphComponent
from b_asic.port import InputPort
@@ -16,7 +16,7 @@ from b_asic.signal_flow_graph import SFG
def sfg_to_python(
- sfg: SFG, counter: int = 0, suffix: Optional[str] = None, schedule: bool = False
+ sfg: SFG, counter: int = 0, suffix: str | None = None, schedule: bool = False
) -> str:
"""
Given an SFG structure try to serialize it for saving to a file.
@@ -141,7 +141,7 @@ def sfg_to_python(
return result
-def python_to_sfg(path: str) -> Tuple[SFG, Dict[str, Tuple[int, int]]]:
+def python_to_sfg(path: str) -> tuple[SFG, dict[str, tuple[int, int]]]:
"""
Given a serialized file, try to deserialize it and load it to the library.
diff --git a/b_asic/schedule.py b/b_asic/schedule.py
index c815fdb2..13433612 100644
--- a/b_asic/schedule.py
+++ b/b_asic/schedule.py
@@ -7,7 +7,8 @@ Contains the schedule class for scheduling operations in an SFG.
import io
import sys
from collections import defaultdict
-from typing import Dict, List, Optional, Sequence, Tuple, cast
+from collections.abc import Sequence
+from typing import cast
import matplotlib.pyplot as plt
import numpy as np
@@ -39,11 +40,11 @@ from b_asic.special_operations import Delay, Input, Output
from b_asic.types import TypeName
# Need RGB from 0 to 1
-_EXECUTION_TIME_COLOR: Tuple[float, ...] = tuple(
+_EXECUTION_TIME_COLOR: tuple[float, ...] = tuple(
float(c / 255) for c in EXECUTION_TIME_COLOR
)
-_LATENCY_COLOR: Tuple[float, ...] = tuple(float(c / 255) for c in LATENCY_COLOR)
-_SIGNAL_COLOR: Tuple[float, ...] = tuple(float(c / 255) for c in SIGNAL_COLOR)
+_LATENCY_COLOR: tuple[float, ...] = tuple(float(c / 255) for c in LATENCY_COLOR)
+_SIGNAL_COLOR: tuple[float, ...] = tuple(float(c / 255) for c in SIGNAL_COLOR)
def _laps_default():
@@ -80,20 +81,20 @@ class Schedule:
"""
_sfg: SFG
- _start_times: Dict[GraphID, int]
- _laps: Dict[GraphID, int]
+ _start_times: dict[GraphID, int]
+ _laps: dict[GraphID, int]
_schedule_time: int
_cyclic: bool
- _y_locations: Dict[GraphID, Optional[int]]
+ _y_locations: dict[GraphID, int | None]
def __init__(
self,
sfg: SFG,
- scheduler: Optional[Scheduler] = None,
- schedule_time: Optional[int] = None,
+ scheduler: Scheduler | None = None,
+ schedule_time: int | None = None,
cyclic: bool = False,
- start_times: Optional[Dict[GraphID, int]] = None,
- laps: Optional[Dict[GraphID, int]] = None,
+ start_times: dict[GraphID, int] | None = None,
+ laps: dict[GraphID, int] | None = None,
):
"""Construct a Schedule from an SFG."""
if not isinstance(sfg, SFG):
@@ -126,7 +127,7 @@ class Schedule:
def __str__(self) -> str:
"""Return a string representation of this Schedule."""
- res: List[Tuple[GraphID, int, int, int]] = [
+ res: list[tuple[GraphID, int, int, int]] = [
(
op.graph_id,
self.start_time_of_operation(op.graph_id),
@@ -265,7 +266,7 @@ class Schedule:
def _forward_slacks(
self, graph_id: GraphID
- ) -> Dict["OutputPort", Dict["Signal", int]]:
+ ) -> dict["OutputPort", dict["Signal", int]]:
ret = {}
start_time = self._start_times[graph_id]
operation = cast(Operation, self._sfg.find_by_id(graph_id))
@@ -274,8 +275,8 @@ class Schedule:
return ret
def _output_slacks(
- self, output_port: "OutputPort", start_time: Optional[int] = None
- ) -> Dict[Signal, int]:
+ self, output_port: "OutputPort", start_time: int | None = None
+ ) -> dict[Signal, int]:
if start_time is None:
start_time = self._start_times[output_port.operation.graph_id]
output_slacks = {}
@@ -331,7 +332,7 @@ class Schedule:
),
)
- def _backward_slacks(self, graph_id: GraphID) -> Dict[InputPort, Dict[Signal, int]]:
+ def _backward_slacks(self, graph_id: GraphID) -> dict[InputPort, dict[Signal, int]]:
ret = {}
start_time = self._start_times[graph_id]
operation = cast(Operation, self._sfg.find_by_id(graph_id))
@@ -340,8 +341,8 @@ class Schedule:
return ret
def _input_slacks(
- self, input_port: InputPort, start_time: Optional[int] = None
- ) -> Dict[Signal, int]:
+ self, input_port: InputPort, start_time: int | None = None
+ ) -> dict[Signal, int]:
if start_time is None:
start_time = self._start_times[input_port.operation.graph_id]
input_slacks = {}
@@ -361,7 +362,7 @@ class Schedule:
input_slacks[signal] = usage_time - available_time
return input_slacks
- def slacks(self, graph_id: GraphID) -> Tuple[int, int]:
+ def slacks(self, graph_id: GraphID) -> tuple[int, int]:
"""
Return the backward and forward slacks of operation *graph_id*.
@@ -401,7 +402,7 @@ class Schedule:
* 1: backward slack
* 2: forward slack
"""
- res: List[Tuple[GraphID, int, int]] = [
+ res: list[tuple[GraphID, int, int]] = [
(
op.graph_id,
cast(int, self.backward_slack(op.graph_id)),
@@ -463,7 +464,7 @@ class Schedule:
return simplified_sfg
@property
- def start_times(self) -> Dict[GraphID, int]:
+ def start_times(self) -> dict[GraphID, int]:
"""The start times of the operations in the schedule."""
return self._start_times
@@ -477,7 +478,7 @@ class Schedule:
self._start_times = start_times
@property
- def laps(self) -> Dict[GraphID, int]:
+ def laps(self) -> dict[GraphID, int]:
"""
The number of laps for the start times of the operations in the schedule.
"""
@@ -529,7 +530,7 @@ class Schedule:
self._schedule_time *= factor
return self
- def _get_all_times(self) -> List[int]:
+ def _get_all_times(self) -> list[int]:
"""
Return a list of all times for the schedule.
@@ -548,7 +549,7 @@ class Schedule:
ret = [v for v in ret if v is not None]
return ret
- def get_possible_time_resolution_decrements(self) -> List[int]:
+ def get_possible_time_resolution_decrements(self) -> list[int]:
"""Return a list with possible factors to reduce time resolution."""
vals = self._get_all_times()
max_loop = min(val for val in vals if val)
@@ -912,8 +913,8 @@ class Schedule:
new_sfg = new_sfg.insert_operation_before(op, Delay(), port)
return new_sfg()
- def _get_memory_variables_list(self) -> List[MemoryVariable]:
- ret: List[MemoryVariable] = []
+ def _get_memory_variables_list(self) -> list[MemoryVariable]:
+ ret: list[MemoryVariable] = []
for graph_id, start_time in self._start_times.items():
slacks = self._forward_slacks(graph_id)
for outport, signals in slacks.items():
@@ -963,7 +964,7 @@ class Schedule:
self.cyclic,
)
- def get_used_type_names(self) -> List[TypeName]:
+ def get_used_type_names(self) -> list[TypeName]:
"""Get a list of all TypeNames used in the Schedule."""
return self._sfg.get_used_type_names()
@@ -1185,7 +1186,7 @@ class Schedule:
self._plot_schedule(ax, operation_gap=operation_gap)
def show(
- self, operation_gap: float = OPERATION_GAP, title: Optional[str] = None
+ self, operation_gap: float = OPERATION_GAP, title: str | None = None
) -> None:
"""
Show the schedule. Will display based on the current Matplotlib backend.
diff --git a/b_asic/scheduler.py b/b_asic/scheduler.py
index e6b8dfd7..9a40ec02 100644
--- a/b_asic/scheduler.py
+++ b/b_asic/scheduler.py
@@ -2,7 +2,7 @@ import copy
import sys
from abc import ABC, abstractmethod
from math import ceil
-from typing import TYPE_CHECKING, Optional, cast
+from typing import TYPE_CHECKING, cast
import b_asic.logger as logger
from b_asic.core_operations import DontCare
@@ -29,7 +29,7 @@ class Scheduler(ABC):
raise NotImplementedError
def _handle_outputs(
- self, schedule: "Schedule", non_schedulable_ops: Optional[list["GraphID"]] = []
+ self, schedule: "Schedule", non_schedulable_ops: list["GraphID"] | None = []
) -> None:
for output in schedule.sfg.find_by_type_name(Output.type_name()):
output = cast(Output, output)
@@ -165,12 +165,12 @@ class ListScheduler(Scheduler, ABC):
def __init__(
self,
- max_resources: Optional[dict[TypeName, int]] = None,
- max_concurrent_reads: Optional[int] = None,
- max_concurrent_writes: Optional[int] = None,
- input_times: Optional[dict["GraphID", int]] = None,
- output_delta_times: Optional[dict["GraphID", int]] = None,
- cyclic: Optional[bool] = False,
+ max_resources: dict[TypeName, int] | None = None,
+ max_concurrent_reads: int | None = None,
+ max_concurrent_writes: int | None = None,
+ input_times: dict["GraphID", int] | None = None,
+ output_delta_times: dict["GraphID", int] | None = None,
+ cyclic: bool | None = False,
) -> None:
super()
self._logger = logger.getLogger(__name__, "list_scheduler.log", "DEBUG")
diff --git a/b_asic/scheduler_gui/axes_item.py b/b_asic/scheduler_gui/axes_item.py
index 65520ea6..f65b11c0 100644
--- a/b_asic/scheduler_gui/axes_item.py
+++ b/b_asic/scheduler_gui/axes_item.py
@@ -6,7 +6,6 @@ Contains the scheduler-gui AxesItem class for drawing and maintaining the
axes in a graph.
"""
from math import pi, sin
-from typing import List, Optional, Union
# QGraphics and QPainter imports
from qtpy.QtCore import QPointF, Qt
@@ -51,12 +50,12 @@ class AxesItem(QGraphicsItemGroup):
_x_axis: QGraphicsLineItem
_x_label: QGraphicsSimpleTextItem
_x_arrow: QGraphicsPolygonItem
- _x_scale: List[QGraphicsLineItem]
- _x_scale_labels: List[QGraphicsSimpleTextItem]
- _x_ledger: List[Union[QGraphicsLineItem, TimelineItem]]
+ _x_scale: list[QGraphicsLineItem]
+ _x_scale_labels: list[QGraphicsSimpleTextItem]
+ _x_ledger: list[QGraphicsLineItem | TimelineItem]
_x_label_offset: float
_y_axis: QGraphicsLineItem
- _event_items: List[QGraphicsItem]
+ _event_items: list[QGraphicsItem]
_base_pen: QPen
_ledger_pen: QPen
_timeline_pen: QPen
@@ -69,7 +68,7 @@ class AxesItem(QGraphicsItemGroup):
height_indent: float = SCHEDULE_INDENT,
width_padding: float = 0.6,
height_padding: float = 0.5,
- parent: Optional[QGraphicsItem] = None,
+ parent: QGraphicsItem | None = None,
):
"""
Class for an AxesItem.
@@ -134,7 +133,7 @@ class AxesItem(QGraphicsItemGroup):
return self._height
@property
- def event_items(self) -> List[QGraphicsItem]:
+ def event_items(self) -> list[QGraphicsItem]:
"""Return a list of objects that receives events."""
return [self._x_ledger[-1]]
diff --git a/b_asic/scheduler_gui/main_window.py b/b_asic/scheduler_gui/main_window.py
index 9d035587..09756521 100644
--- a/b_asic/scheduler_gui/main_window.py
+++ b/b_asic/scheduler_gui/main_window.py
@@ -14,7 +14,7 @@ import webbrowser
from collections import defaultdict, deque
from copy import deepcopy
from importlib.machinery import SourceFileLoader
-from typing import TYPE_CHECKING, Deque, Dict, List, Optional, cast, overload
+from typing import TYPE_CHECKING, Deque, cast, overload
# Qt/qtpy
import qtpy
@@ -119,15 +119,15 @@ class ScheduleMainWindow(QMainWindow, Ui_MainWindow):
"""Schedule of an SFG with scheduled Operations."""
_scene: QGraphicsScene
- _schedule: Optional[Schedule]
- _graph: Optional[SchedulerItem]
+ _schedule: Schedule | None
+ _graph: SchedulerItem | None
_scale: float
_debug_rectangles: QGraphicsItemGroup
_splitter_pos: int
_splitter_min: int
_zoom: float
- _color_per_type: Dict[str, QColor] = dict()
- converted_colorPerType: Dict[str, str] = dict()
+ _color_per_type: dict[str, QColor] = dict()
+ converted_colorPerType: dict[str, str] = dict()
def __init__(self):
"""Initialize Scheduler-GUI."""
@@ -148,11 +148,11 @@ class ScheduleMainWindow(QMainWindow, Ui_MainWindow):
self._execution_time_plot_dialogs = defaultdict(lambda: None)
self._ports_accesses_for_storage = None
self._color_changed_perType = False
- self.changed_operation_colors: Dict[str, QColor] = dict()
+ self.changed_operation_colors: dict[str, QColor] = dict()
# Recent files
self._max_recent_files = 4
- self._recent_files_actions: List[QAction] = []
+ self._recent_files_actions: list[QAction] = []
self._recent_file_paths: Deque[str] = deque(maxlen=self._max_recent_files)
self._create_recent_file_actions_and_menus()
@@ -239,7 +239,7 @@ class ScheduleMainWindow(QMainWindow, Ui_MainWindow):
self._scene.sceneRectChanged.connect(self.shrink_scene_to_min_size)
@property
- def schedule(self) -> Optional[Schedule]:
+ def schedule(self) -> Schedule | None:
"""The current schedule."""
return self._schedule
@@ -1584,7 +1584,7 @@ class ScheduleMainWindow(QMainWindow, Ui_MainWindow):
@Slot()
@Slot(str)
- def _schedule_changed(self, type_name: Optional[str] = None):
+ def _schedule_changed(self, type_name: str | None = None):
self._update_execution_times_for_variables()
self._update_ports_accesses_for_storage()
for key, dialog in self._execution_time_plot_dialogs.items():
@@ -1666,14 +1666,14 @@ def start_scheduler(schedule: Schedule) -> Schedule: ...
@overload
-def start_scheduler(schedule: None) -> Optional[Schedule]: ...
+def start_scheduler(schedule: None) -> Schedule | None: ...
@overload
-def start_scheduler() -> Optional[Schedule]: ...
+def start_scheduler() -> Schedule | None: ...
-def start_scheduler(schedule: Optional[Schedule] = None) -> Optional[Schedule]:
+def start_scheduler(schedule: Schedule | None = None) -> Schedule | None:
"""
Start scheduler GUI.
diff --git a/b_asic/scheduler_gui/operation_item.py b/b_asic/scheduler_gui/operation_item.py
index 6b9fc774..eab2c317 100644
--- a/b_asic/scheduler_gui/operation_item.py
+++ b/b_asic/scheduler_gui/operation_item.py
@@ -5,7 +5,7 @@ B-ASIC Scheduler-GUI Operation Item Module.
Contains the scheduler_gui OperationItem class for drawing and maintain an operation
in the schedule.
"""
-from typing import TYPE_CHECKING, Dict, List, Union, cast
+from typing import TYPE_CHECKING, cast
# QGraphics and QPainter imports
from qtpy.QtCore import QPointF, Qt
@@ -56,13 +56,13 @@ class OperationItem(QGraphicsItemGroup):
"""Static, changed from MainWindow."""
_operation: Operation
_height: float
- _ports: Dict[str, Dict[str, Union[float, QPointF]]] # ['port-id']['latency/pos']
+ _ports: dict[str, dict[str, float | QPointF]] # ['port-id']['latency/pos']
_end_time: int
_latency_item: QGraphicsPathItem
_execution_time_item: QGraphicsPathItem
_label_item: QGraphicsSimpleTextItem
- _port_items: List[QGraphicsEllipseItem]
- _port_number_items: List[QGraphicsSimpleTextItem]
+ _port_items: list[QGraphicsEllipseItem]
+ _port_number_items: list[QGraphicsSimpleTextItem]
_inactive_color: QColor = Latency_Color.DEFAULT
def __init__(
@@ -80,7 +80,7 @@ class OperationItem(QGraphicsItemGroup):
self._operation = operation
self._height = height
operation._check_all_latencies_set()
- latency_offsets = cast(Dict[str, int], operation.latency_offsets)
+ latency_offsets = cast(dict[str, int], operation.latency_offsets)
self._ports = {k: {"latency": float(v)} for k, v in latency_offsets.items()}
self._end_time = max(latency_offsets.values())
self._port_items = []
@@ -177,7 +177,7 @@ class OperationItem(QGraphicsItemGroup):
self._make_component()
@property
- def event_items(self) -> List[QGraphicsItem]:
+ def event_items(self) -> list[QGraphicsItem]:
"""List of objects that receives events."""
return [self]
diff --git a/b_asic/scheduler_gui/scheduler_event.py b/b_asic/scheduler_gui/scheduler_event.py
index 617e14cb..872b4511 100644
--- a/b_asic/scheduler_gui/scheduler_event.py
+++ b/b_asic/scheduler_gui/scheduler_event.py
@@ -6,7 +6,7 @@ Contains the scheduler_ui SchedulerEvent class containing event filters and
handlers for SchedulerItem objects.
"""
import math
-from typing import List, Optional, overload
+from typing import overload
# QGraphics and QPainter imports
from qtpy.QtCore import QEvent, QObject, QPointF, Qt, Signal
@@ -40,14 +40,14 @@ class SchedulerEvent:
execution_time_plot = Signal(str)
TextSignal = Signal(str)
- _axes: Optional[AxesItem]
+ _axes: AxesItem | None
_current_pos: QPointF
_delta_time: int
_signals: Signals
_schedule: Schedule
_old_op_position: int = -1
- def __init__(self, parent: Optional[QGraphicsItem] = None):
+ def __init__(self, parent: QGraphicsItem | None = None):
super().__init__(parent=parent)
self._signals = self.Signals()
@@ -73,7 +73,7 @@ class SchedulerEvent:
def installSceneEventFilters(self, filterItems: QGraphicsItem) -> None: ...
@overload
- def installSceneEventFilters(self, filterItems: List[QGraphicsItem]) -> None: ...
+ def installSceneEventFilters(self, filterItems: list[QGraphicsItem]) -> None: ...
def installSceneEventFilters(self, filterItems) -> None:
"""
@@ -90,7 +90,7 @@ class SchedulerEvent:
def removeSceneEventFilters(self, filterItems: QGraphicsItem) -> None: ...
@overload
- def removeSceneEventFilters(self, filterItems: List[QGraphicsItem]) -> None: ...
+ def removeSceneEventFilters(self, filterItems: list[QGraphicsItem]) -> None: ...
def removeSceneEventFilters(self, filterItems) -> None:
"""
diff --git a/b_asic/scheduler_gui/scheduler_item.py b/b_asic/scheduler_gui/scheduler_item.py
index 13af04a9..fbb84fcd 100644
--- a/b_asic/scheduler_gui/scheduler_item.py
+++ b/b_asic/scheduler_gui/scheduler_item.py
@@ -8,7 +8,7 @@ maintaining a schedule.
from collections import defaultdict
from math import floor
from pprint import pprint
-from typing import Dict, List, Optional, Set, cast
+from typing import cast
# QGraphics and QPainter imports
from qtpy.QtCore import Signal
@@ -58,18 +58,18 @@ class SchedulerItem(SchedulerEvent, QGraphicsItemGroup):
The parent. Passed to the constructor of QGraphicsItemGroup.
"""
- _axes: Optional[AxesItem]
- _operation_items: Dict[str, OperationItem]
+ _axes: AxesItem | None
+ _operation_items: dict[str, OperationItem]
_x_axis_indent: float
- _event_items: List[QGraphicsItem]
- _signal_dict: Dict[OperationItem, Set[SignalItem]]
+ _event_items: list[QGraphicsItem]
+ _signal_dict: dict[OperationItem, set[SignalItem]]
def __init__(
self,
schedule: Schedule,
warnings: bool = True,
show_port_numbers: bool = False,
- parent: Optional[QGraphicsItem] = None,
+ parent: QGraphicsItem | None = None,
):
"""
Construct a SchedulerItem.
@@ -292,15 +292,15 @@ class SchedulerItem(SchedulerEvent, QGraphicsItemGroup):
return self._schedule
@property
- def axes(self) -> Optional[AxesItem]:
+ def axes(self) -> AxesItem | None:
return self._axes
@property
- def components(self) -> List[OperationItem]:
+ def components(self) -> list[OperationItem]:
return list(component for component in self._operation_items.values())
@property
- def event_items(self) -> List[QGraphicsItem]:
+ def event_items(self) -> list[QGraphicsItem]:
"""Return a list of objects that receives events."""
return self._event_items
diff --git a/b_asic/scheduler_gui/timeline_item.py b/b_asic/scheduler_gui/timeline_item.py
index b9296cf3..5fd10d07 100644
--- a/b_asic/scheduler_gui/timeline_item.py
+++ b/b_asic/scheduler_gui/timeline_item.py
@@ -5,7 +5,7 @@ B-ASIC Scheduler-GUI Timeline Item Module.
Contains the scheduler_gui TimelineItem class for drawing and
maintain the timeline in a schedule.
"""
-from typing import List, Optional, overload
+from typing import overload
# QGraphics and QPainter imports
from qtpy.QtCore import QLineF, Qt
@@ -20,7 +20,7 @@ class TimelineItem(QGraphicsLineItem):
_delta_time_label: QGraphicsTextItem
@overload
- def __init__(self, line: QLineF, parent: Optional[QGraphicsItem] = None) -> None:
+ def __init__(self, line: QLineF, parent: QGraphicsItem | None = None) -> None:
"""
Constructs a TimelineItem out of *line*.
@@ -28,7 +28,7 @@ class TimelineItem(QGraphicsLineItem):
"""
@overload
- def __init__(self, parent: Optional[QGraphicsItem] = None) -> None:
+ def __init__(self, parent: QGraphicsItem | None = None) -> None:
"""Constructs a TimelineItem.
*parent* is passed to QGraphicsLineItem's constructor."""
@@ -40,7 +40,7 @@ class TimelineItem(QGraphicsLineItem):
y1: float,
x2: float,
y2: float,
- parent: Optional[QGraphicsItem] = None,
+ parent: QGraphicsItem | None = None,
) -> None:
"""
Constructs a TimelineItem from (x1, y1) to (x2, y2).
@@ -126,6 +126,6 @@ class TimelineItem(QGraphicsLineItem):
self._delta_time_label.setScale(scale)
@property
- def event_items(self) -> List[QGraphicsItem]:
+ def event_items(self) -> list[QGraphicsItem]:
"""Return a list of objects that receives events."""
return [self]
diff --git a/b_asic/sfg_generators.py b/b_asic/sfg_generators.py
index 2d68215c..ceb602bd 100644
--- a/b_asic/sfg_generators.py
+++ b/b_asic/sfg_generators.py
@@ -4,7 +4,8 @@ B-ASIC signal flow graph generators.
This module contains a number of functions generating SFGs for specific functions.
"""
-from typing import TYPE_CHECKING, Dict, Optional, Sequence, Union
+from collections.abc import Sequence
+from typing import TYPE_CHECKING
import numpy as np
@@ -28,10 +29,10 @@ if TYPE_CHECKING:
def wdf_allpass(
coefficients: Sequence[float],
- name: Optional[str] = None,
- latency: Optional[int] = None,
- latency_offsets: Optional[Dict[str, int]] = None,
- execution_time: Optional[int] = None,
+ name: str | None = None,
+ latency: int | None = None,
+ latency_offsets: dict[str, int] | None = None,
+ execution_time: int | None = None,
) -> SFG:
"""
Generate a signal flow graph of a WDF allpass section based on symmetric two-port\
@@ -131,9 +132,9 @@ def wdf_allpass(
def direct_form_fir(
coefficients: Sequence[complex],
- name: Optional[str] = None,
- mult_properties: Optional[Union[Dict[str, int], Dict[str, Dict[str, int]]]] = None,
- add_properties: Optional[Union[Dict[str, int], Dict[str, Dict[str, int]]]] = None,
+ name: str | None = None,
+ mult_properties: dict[str, int] | dict[str, dict[str, int]] | None = None,
+ add_properties: dict[str, int] | dict[str, dict[str, int]] | None = None,
) -> SFG:
r"""
Generate a signal flow graph of a direct form FIR filter.
@@ -199,9 +200,9 @@ def direct_form_fir(
def transposed_direct_form_fir(
coefficients: Sequence[complex],
- name: Optional[str] = None,
- mult_properties: Optional[Union[Dict[str, int], Dict[str, Dict[str, int]]]] = None,
- add_properties: Optional[Union[Dict[str, int], Dict[str, Dict[str, int]]]] = None,
+ name: str | None = None,
+ mult_properties: dict[str, int] | dict[str, dict[str, int]] | None = None,
+ add_properties: dict[str, int] | dict[str, dict[str, int]] | None = None,
) -> SFG:
r"""
Generate a signal flow graph of a transposed direct form FIR filter.
@@ -267,9 +268,9 @@ def transposed_direct_form_fir(
def direct_form_1_iir(
b: Sequence[complex],
a: Sequence[complex],
- name: Optional[str] = None,
- mult_properties: Optional[Union[Dict[str, int], Dict[str, Dict[str, int]]]] = None,
- add_properties: Optional[Union[Dict[str, int], Dict[str, Dict[str, int]]]] = None,
+ name: str | None = None,
+ mult_properties: dict[str, int] | dict[str, dict[str, int]] | None = None,
+ add_properties: dict[str, int] | dict[str, dict[str, int]] | None = None,
) -> SFG:
"""Generates a direct-form IIR filter of type I with coefficients a and b."""
if len(a) < 2 or len(b) < 2:
@@ -328,9 +329,9 @@ def direct_form_1_iir(
def direct_form_2_iir(
b: Sequence[complex],
a: Sequence[complex],
- name: Optional[str] = None,
- mult_properties: Optional[Union[Dict[str, int], Dict[str, Dict[str, int]]]] = None,
- add_properties: Optional[Union[Dict[str, int], Dict[str, Dict[str, int]]]] = None,
+ name: str | None = None,
+ mult_properties: dict[str, int] | dict[str, dict[str, int]] | None = None,
+ add_properties: dict[str, int] | dict[str, dict[str, int]] | None = None,
) -> SFG:
"""Generates a direct-form IIR filter of type II with coefficients a and b."""
if len(a) < 2 or len(b) < 2:
diff --git a/b_asic/signal.py b/b_asic/signal.py
index d755bfd5..ae4bed44 100644
--- a/b_asic/signal.py
+++ b/b_asic/signal.py
@@ -4,7 +4,8 @@ B-ASIC Signal Module.
Contains the class for representing the connections between operations.
"""
-from typing import TYPE_CHECKING, Iterable, Optional, Union
+from collections.abc import Iterable
+from typing import TYPE_CHECKING, Optional, Union
from b_asic.graph_component import AbstractGraphComponent, GraphComponent
from b_asic.types import Name, TypeName
@@ -44,9 +45,9 @@ class Signal(AbstractGraphComponent):
def __init__(
self,
- source: Optional[Union["OutputPort", "Signal", "Operation"]] = None,
- destination: Optional[Union["InputPort", "Signal", "Operation"]] = None,
- bits: Optional[int] = None,
+ source: Union["OutputPort", "Signal", "Operation"] | None = None,
+ destination: Union["InputPort", "Signal", "Operation"] | None = None,
+ bits: int | None = None,
name: Name = Name(""),
):
"""Construct a Signal."""
@@ -186,7 +187,7 @@ class Signal(AbstractGraphComponent):
return self._source is None or self._destination is None
@property
- def bits(self) -> Optional[int]:
+ def bits(self) -> int | None:
"""
Get the number of bits that this operation using this signal as an
input should quantize received values to.
@@ -195,7 +196,7 @@ class Signal(AbstractGraphComponent):
return self.param("bits")
@bits.setter
- def bits(self, bits: Optional[int]) -> None:
+ def bits(self, bits: int | None) -> None:
"""
Set the number of bits that operations using this signal as an input
should quantize received values to.
diff --git a/b_asic/signal_flow_graph.py b/b_asic/signal_flow_graph.py
index 903ab2af..abc380ec 100644
--- a/b_asic/signal_flow_graph.py
+++ b/b_asic/signal_flow_graph.py
@@ -8,20 +8,14 @@ import itertools
import re
import warnings
from collections import defaultdict, deque
+from collections.abc import Iterable, MutableSet, Sequence
from io import StringIO
from numbers import Number
from queue import PriorityQueue
from typing import (
DefaultDict,
Deque,
- Dict,
- Iterable,
- List,
- MutableSet,
Optional,
- Sequence,
- Set,
- Tuple,
Union,
cast,
)
@@ -42,7 +36,7 @@ from b_asic.signal import Signal
from b_asic.special_operations import Delay, Input, Output
from b_asic.types import GraphID, GraphIDNumber, Name, Num, TypeName
-DelayQueue = List[Tuple[str, ResultKey, OutputPort]]
+DelayQueue = list[tuple[str, ResultKey, OutputPort]]
_OPERATION_SHAPE: DefaultDict[TypeName, str] = defaultdict(lambda: "ellipse")
@@ -115,29 +109,29 @@ class SFG(AbstractOperation):
input_sources :
"""
- _components_by_id: Dict[GraphID, GraphComponent]
- _components_by_name: DefaultDict[Name, List[GraphComponent]]
- _components_dfs_order: List[GraphComponent]
- _operations_dfs_order: List[Operation]
- _operations_topological_order: List[Operation]
+ _components_by_id: dict[GraphID, GraphComponent]
+ _components_by_name: DefaultDict[Name, list[GraphComponent]]
+ _components_dfs_order: list[GraphComponent]
+ _operations_dfs_order: list[Operation]
+ _operations_topological_order: list[Operation]
_graph_id_generator: GraphIDGenerator
- _input_operations: List[Input]
- _output_operations: List[Output]
- _original_components_to_new: Dict[GraphComponent, GraphComponent]
- _original_input_signals_to_indices: Dict[Signal, int]
- _original_output_signals_to_indices: Dict[Signal, int]
- _precedence_list: Optional[List[List[OutputPort]]]
- _used_ids: Set[GraphID] = set()
+ _input_operations: list[Input]
+ _output_operations: list[Output]
+ _original_components_to_new: dict[GraphComponent, GraphComponent]
+ _original_input_signals_to_indices: dict[Signal, int]
+ _original_output_signals_to_indices: dict[Signal, int]
+ _precedence_list: list[list[OutputPort]] | None
+ _used_ids: set[GraphID] = set()
def __init__(
self,
- inputs: Optional[Sequence[Input]] = None,
- outputs: Optional[Sequence[Output]] = None,
- input_signals: Optional[Sequence[Signal]] = None,
- output_signals: Optional[Sequence[Signal]] = None,
+ inputs: Sequence[Input] | None = None,
+ outputs: Sequence[Output] | None = None,
+ input_signals: Sequence[Signal] | None = None,
+ output_signals: Sequence[Signal] | None = None,
id_number_offset: GraphIDNumber = GraphIDNumber(0),
name: Name = Name(""),
- input_sources: Optional[Sequence[Optional[SignalSourceProvider]]] = None,
+ input_sources: Sequence[SignalSourceProvider | None] | None = None,
):
input_signal_count = 0 if input_signals is None else len(input_signals)
input_operation_count = 0 if inputs is None else len(inputs)
@@ -323,7 +317,7 @@ class SFG(AbstractOperation):
return string_io.getvalue()
def __call__(
- self, *src: Optional[SignalSourceProvider], name: Name = Name("")
+ self, *src: SignalSourceProvider | None, name: Name = Name("")
) -> "SFG":
"""
Return a new independent SFG instance that is identical to this SFG
@@ -351,10 +345,10 @@ class SFG(AbstractOperation):
self,
index: int,
input_values: Sequence[Num],
- results: Optional[MutableResultMap] = None,
- delays: Optional[MutableDelayMap] = None,
+ results: MutableResultMap | None = None,
+ delays: MutableDelayMap | None = None,
prefix: str = "",
- bits_override: Optional[int] = None,
+ bits_override: int | None = None,
quantize: bool = True,
) -> Number:
# doc-string inherited
@@ -506,7 +500,7 @@ class SFG(AbstractOperation):
}
output_op = self._output_operations[output_index]
queue: Deque[Operation] = deque([output_op])
- visited: Set[Operation] = {output_op}
+ visited: set[Operation] = {output_op}
while queue:
op = queue.popleft()
if isinstance(op, Input):
@@ -542,12 +536,12 @@ class SFG(AbstractOperation):
return self._graph_id_generator.id_number_offset
@property
- def components(self) -> List[GraphComponent]:
+ def components(self) -> list[GraphComponent]:
"""Get all components of this graph in depth-first order."""
return self._components_dfs_order
@property
- def operations(self) -> List[Operation]:
+ def operations(self) -> list[Operation]:
"""Get all operations of this graph in depth-first order."""
return list(self._operations_dfs_order)
@@ -569,7 +563,7 @@ class SFG(AbstractOperation):
]
return components
- def find_by_id(self, graph_id: GraphID) -> Optional[GraphComponent]:
+ def find_by_id(self, graph_id: GraphID) -> GraphComponent | None:
"""
Find the graph component with the specified ID.
@@ -762,7 +756,7 @@ class SFG(AbstractOperation):
self,
input_comp_id: GraphID,
new_operation: Operation,
- port: Optional[int] = None,
+ port: int | None = None,
) -> Optional["SFG"]:
"""
Insert an operation in the SFG before a given source operation.
@@ -1093,7 +1087,7 @@ class SFG(AbstractOperation):
remaining_inports_per_operation = {op: op.input_count for op in self.operations}
# Maps number of input counts to a queue of seen objects with such a size.
- seen_with_inputs_dict: Dict[int, Deque] = defaultdict(deque)
+ seen_with_inputs_dict: dict[int, Deque] = defaultdict(deque)
seen = set()
top_order = []
@@ -1212,7 +1206,7 @@ class SFG(AbstractOperation):
cast(Operation, op).execution_time = execution_time
def set_latency_offsets_of_type(
- self, type_name: TypeName, latency_offsets: Dict[str, int]
+ self, type_name: TypeName, latency_offsets: dict[str, int]
) -> None:
"""
Set the latency offsets of all operations with the given type name.
@@ -1229,8 +1223,8 @@ class SFG(AbstractOperation):
cast(Operation, op).set_latency_offsets(latency_offsets)
def _traverse_for_precedence_list(
- self, first_iter_ports: List[OutputPort]
- ) -> List[List[OutputPort]]:
+ self, first_iter_ports: list[OutputPort]
+ ) -> list[list[OutputPort]]:
# Find dependencies of output ports and input ports.
remaining_inports_per_operation = {op: op.input_count for op in self.operations}
@@ -1474,7 +1468,7 @@ class SFG(AbstractOperation):
results: MutableResultMap,
delays: MutableDelayMap,
prefix: str,
- bits_override: Optional[int],
+ bits_override: int | None,
quantize: bool,
deferred_delays: DelayQueue,
) -> Num:
@@ -1519,7 +1513,7 @@ class SFG(AbstractOperation):
results: MutableResultMap,
delays: MutableDelayMap,
prefix: str,
- bits_override: Optional[int],
+ bits_override: int | None,
quantize: bool,
deferred_delays: DelayQueue,
) -> Num:
@@ -1550,7 +1544,7 @@ class SFG(AbstractOperation):
def sfg_digraph(
self,
show_signal_id: bool = False,
- engine: Optional[str] = None,
+ engine: str | None = None,
branch_node: bool = True,
port_numbering: bool = True,
splines: str = "spline",
@@ -1661,9 +1655,9 @@ class SFG(AbstractOperation):
def show(
self,
- fmt: Optional[str] = None,
+ fmt: str | None = None,
show_signal_id: bool = False,
- engine: Optional[str] = None,
+ engine: str | None = None,
branch_node: bool = True,
port_numbering: bool = True,
splines: str = "spline",
@@ -1755,7 +1749,7 @@ class SFG(AbstractOperation):
for input in inputs_used:
input_op = self._input_operations[input]
queue: Deque[Operation] = deque([input_op])
- visited: Set[Operation] = {input_op}
+ visited: set[Operation] = {input_op}
dict_of_sfg = {}
while queue:
op = queue.popleft()
@@ -1839,7 +1833,7 @@ class SFG(AbstractOperation):
for output in output_index_used:
output_op = self._output_operations[output]
queue: Deque[Operation] = deque([output_op])
- visited: Set[Operation] = {output_op}
+ visited: set[Operation] = {output_op}
while queue:
op = queue.popleft()
for input_port in op.inputs:
@@ -1855,7 +1849,7 @@ class SFG(AbstractOperation):
for input in input_index_used:
input_op = self._input_operations[input]
queue: Deque[Operation] = deque([input_op])
- visited: Set[Operation] = {input_op}
+ visited: set[Operation] = {input_op}
while queue:
op = queue.popleft()
if isinstance(op, Output):
@@ -2020,7 +2014,7 @@ class SFG(AbstractOperation):
paths.append(newpath)
return paths
- def edit(self) -> Dict[str, "SFG"]:
+ def edit(self) -> dict[str, "SFG"]:
"""Edit SFG in GUI."""
from b_asic.GUI.main_window import start_editor
@@ -2141,13 +2135,13 @@ class SFG(AbstractOperation):
def is_constant(self) -> bool:
return all(output.is_constant for output in self._output_operations)
- def get_used_type_names(self) -> List[TypeName]:
+ def get_used_type_names(self) -> list[TypeName]:
"""Get a list of all TypeNames used in the SFG."""
ret = list({op.type_name() for op in self.operations})
ret.sort()
return ret
- def get_used_graph_ids(self) -> Set[GraphID]:
+ def get_used_graph_ids(self) -> set[GraphID]:
"""Get a list of all GraphID:s used in the SFG."""
ret = set({op.graph_id for op in self.operations})
sorted(ret)
diff --git a/b_asic/signal_generator.py b/b_asic/signal_generator.py
index 45706d81..21381b48 100644
--- a/b_asic/signal_generator.py
+++ b/b_asic/signal_generator.py
@@ -11,9 +11,9 @@ This is handled by a number of private generator classes. Check out the source c
if you want more information.
"""
+from collections.abc import Sequence
from math import pi, sin
from numbers import Number
-from typing import List, Optional, Sequence, Union
import numpy as np
@@ -195,12 +195,12 @@ class FromFile(SignalGenerator):
__slots__ = ("_data", "_len", "_path")
_path: str
- _data: List[Num]
+ _data: list[Num]
_len: int
def __init__(self, path: str) -> None:
self._path = path
- data: List[Num] = np.loadtxt(path, dtype=complex).tolist()
+ data: list[Num] = np.loadtxt(path, dtype=complex).tolist()
self._data = data
self._len = len(data)
@@ -264,12 +264,12 @@ class Gaussian(SignalGenerator):
__slots__ = ("_rng", "_seed", "_loc", "_scale")
- _seed: Optional[int]
+ _seed: int | None
_loc: float
_scale: float
def __init__(
- self, seed: Optional[int] = None, loc: float = 0.0, scale: float = 1.0
+ self, seed: int | None = None, loc: float = 0.0, scale: float = 1.0
) -> None:
self._rng = np.random.default_rng(seed)
self._seed = seed
@@ -309,12 +309,12 @@ class Uniform(SignalGenerator):
__slots__ = ("_rng", "_seed", "_low", "_high")
- _seed: Optional[int]
+ _seed: int | None
_low: float
_high: float
def __init__(
- self, seed: Optional[int] = None, low: float = -1.0, high: float = 1.0
+ self, seed: int | None = None, low: float = -1.0, high: float = 1.0
) -> None:
self._rng = np.random.default_rng(seed)
self._seed = seed
@@ -495,13 +495,13 @@ class Upsample(SignalGenerator):
__slots__ = ("_generator", "_factor", "_phase")
- _generator: Union[SignalGenerator, Sequence[complex]]
+ _generator: SignalGenerator | Sequence[complex]
_factor: int
_phase: int
def __init__(
self,
- data: Union[SignalGenerator, Sequence[complex]],
+ data: SignalGenerator | Sequence[complex],
factor: int,
phase: int = 0,
) -> None:
@@ -540,13 +540,13 @@ class Downsample(SignalGenerator):
__slots__ = ("_generator", "_factor", "_phase")
- _generator: Union[SignalGenerator, Sequence[complex]]
+ _generator: SignalGenerator | Sequence[complex]
_factor: int
_phase: int
def __init__(
self,
- data: Union[SignalGenerator, Sequence[complex]],
+ data: SignalGenerator | Sequence[complex],
factor: int,
phase: int = 0,
) -> None:
diff --git a/b_asic/simulation.py b/b_asic/simulation.py
index 31f25910..4b300433 100644
--- a/b_asic/simulation.py
+++ b/b_asic/simulation.py
@@ -5,15 +5,9 @@ Contains a class for simulating the result of an SFG given a set of input values
"""
from collections import defaultdict
+from collections.abc import Callable, Mapping, MutableMapping, MutableSequence, Sequence
from numbers import Number
from typing import (
- Callable,
- List,
- Mapping,
- MutableMapping,
- MutableSequence,
- Optional,
- Sequence,
Union,
)
@@ -50,13 +44,13 @@ class Simulation:
_results: MutableResultArrayMap
_delays: MutableDelayMap
_iteration: int
- _input_functions: List[InputFunction]
- _input_length: Optional[int]
+ _input_functions: list[InputFunction]
+ _input_length: int | None
def __init__(
self,
sfg: SFG,
- input_providers: Optional[Sequence[Optional[InputProvider]]] = None,
+ input_providers: Sequence[InputProvider | None] | None = None,
):
"""Construct a Simulation of an SFG."""
if not isinstance(sfg, SFG):
@@ -103,7 +97,7 @@ class Simulation:
)
self._input_functions[index] = lambda n: input_provider[n]
- def set_inputs(self, input_providers: Sequence[Optional[InputProvider]]) -> None:
+ def set_inputs(self, input_providers: Sequence[InputProvider | None]) -> None:
"""
Set the input functions used to get values for the inputs to the internal SFG.
@@ -123,7 +117,7 @@ class Simulation:
def step(
self,
save_results: bool = True,
- bits_override: Optional[int] = None,
+ bits_override: int | None = None,
quantize: bool = True,
) -> Sequence[Num]:
"""
@@ -141,7 +135,7 @@ class Simulation:
self,
iteration: int,
save_results: bool = True,
- bits_override: Optional[int] = None,
+ bits_override: int | None = None,
quantize: bool = True,
) -> Sequence[Num]:
"""
@@ -184,7 +178,7 @@ class Simulation:
self,
iterations: int,
save_results: bool = True,
- bits_override: Optional[int] = None,
+ bits_override: int | None = None,
quantize: bool = True,
) -> Sequence[Num]:
"""
@@ -207,7 +201,7 @@ class Simulation:
def run(
self,
save_results: bool = True,
- bits_override: Optional[int] = None,
+ bits_override: int | None = None,
quantize: bool = True,
) -> Sequence[Num]:
"""
diff --git a/b_asic/special_operations.py b/b_asic/special_operations.py
index f6d0e84c..dbffb2af 100644
--- a/b_asic/special_operations.py
+++ b/b_asic/special_operations.py
@@ -5,7 +5,7 @@ Contains operations with special purposes that may be treated differently from
normal operations in an SFG.
"""
-from typing import Optional, Sequence, Tuple
+from collections.abc import Sequence
from b_asic.operation import (
AbstractOperation,
@@ -67,7 +67,7 @@ class Input(AbstractOperation):
def get_plot_coordinates(
self,
- ) -> Tuple[Tuple[Tuple[float, float], ...], Tuple[Tuple[float, float], ...]]:
+ ) -> tuple[tuple[tuple[float, float], ...], tuple[tuple[float, float], ...]]:
# Doc-string inherited
return (
(
@@ -88,11 +88,11 @@ class Input(AbstractOperation):
),
)
- def get_input_coordinates(self) -> Tuple[Tuple[float, float], ...]:
+ def get_input_coordinates(self) -> tuple[tuple[float, float], ...]:
# doc-string inherited
return tuple()
- def get_output_coordinates(self) -> Tuple[Tuple[float, float], ...]:
+ def get_output_coordinates(self) -> tuple[tuple[float, float], ...]:
# doc-string inherited
return ((0, 0.5),)
@@ -118,7 +118,7 @@ class Output(AbstractOperation):
def __init__(
self,
- src0: Optional[SignalSourceProvider] = None,
+ src0: SignalSourceProvider | None = None,
name: Name = Name(""),
):
"""Construct an Output operation."""
@@ -140,18 +140,18 @@ class Output(AbstractOperation):
def get_plot_coordinates(
self,
- ) -> Tuple[Tuple[Tuple[float, float], ...], Tuple[Tuple[float, float], ...]]:
+ ) -> tuple[tuple[tuple[float, float], ...], tuple[tuple[float, float], ...]]:
# Doc-string inherited
return (
((0, 0), (0, 1), (0.25, 1), (0.5, 0.5), (0.25, 0), (0, 0)),
((0, 0), (0, 1), (0.25, 1), (0.5, 0.5), (0.25, 0), (0, 0)),
)
- def get_input_coordinates(self) -> Tuple[Tuple[float, float], ...]:
+ def get_input_coordinates(self) -> tuple[tuple[float, float], ...]:
# doc-string inherited
return ((0, 0.5),)
- def get_output_coordinates(self) -> Tuple[Tuple[float, float], ...]:
+ def get_output_coordinates(self) -> tuple[tuple[float, float], ...]:
# doc-string inherited
return tuple()
@@ -181,7 +181,7 @@ class Delay(AbstractOperation):
def __init__(
self,
- src0: Optional[SignalSourceProvider] = None,
+ src0: SignalSourceProvider | None = None,
initial_value: Num = 0,
name: Name = Name(""),
):
@@ -203,8 +203,8 @@ class Delay(AbstractOperation):
return self.param("initial_value")
def current_output(
- self, index: int, delays: Optional[DelayMap] = None, prefix: str = ""
- ) -> Optional[Num]:
+ self, index: int, delays: DelayMap | None = None, prefix: str = ""
+ ) -> Num | None:
if delays is not None:
return delays.get(self.key(index, prefix), self.param("initial_value"))
return self.param("initial_value")
@@ -213,10 +213,10 @@ class Delay(AbstractOperation):
self,
index: int,
input_values: Sequence[Num],
- results: Optional[MutableResultMap] = None,
- delays: Optional[MutableDelayMap] = None,
+ results: MutableResultMap | None = None,
+ delays: MutableDelayMap | None = None,
prefix: str = "",
- bits_override: Optional[int] = None,
+ bits_override: int | None = None,
quantize: bool = True,
) -> Num:
if index != 0:
diff --git a/b_asic/utils.py b/b_asic/utils.py
index c128a32b..654d86a9 100644
--- a/b_asic/utils.py
+++ b/b_asic/utils.py
@@ -1,11 +1,11 @@
"""B-ASIC Utilities."""
-from typing import List, Sequence
+from collections.abc import Sequence
from b_asic.types import Num
-def interleave(*args) -> List[Num]:
+def interleave(*args) -> list[Num]:
"""
Interleave a number of arrays.
@@ -29,7 +29,7 @@ def interleave(*args) -> List[Num]:
return [val for tup in zip(*args) for val in tup]
-def downsample(a: Sequence[Num], factor: int, phase: int = 0) -> List[Num]:
+def downsample(a: Sequence[Num], factor: int, phase: int = 0) -> list[Num]:
"""
Downsample a sequence with an integer factor.
@@ -57,7 +57,7 @@ def downsample(a: Sequence[Num], factor: int, phase: int = 0) -> List[Num]:
return a[phase::factor]
-def upsample(a: Sequence[Num], factor: int, phase: int = 0) -> List[Num]:
+def upsample(a: Sequence[Num], factor: int, phase: int = 0) -> list[Num]:
"""
Upsample a sequence with an integer factor.
@@ -93,7 +93,7 @@ def upsample(a: Sequence[Num], factor: int, phase: int = 0) -> List[Num]:
return interleave(*args)
-def decompose(a: Sequence[Num], factor: int) -> List[List[Num]]:
+def decompose(a: Sequence[Num], factor: int) -> list[list[Num]]:
"""
Polyphase decompose signal *a* into *factor* parts.
--
GitLab