Replace obsolete types

benchmarks/power_distribution/power_distributor.py

@@ -7,8 +7,9 @@
 import csv
 import random
 import timeit
+from collections.abc import Coroutine
 from datetime import timedelta
-from typing import Any, Coroutine, Dict, List, Set  # pylint: disable=unused-import
+from typing import Any
 
 from frequenz.channels import Broadcast
 
@@ -32,7 +33,7 @@
 PORT = 61060
 
 
-async def send_requests(batteries: Set[int], request_num: int) -> List[Result]:
+async def send_requests(batteries: set[int], request_num: int) -> list[Result]:
     """Send requests to the PowerDistributingActor and wait for the response.
 
     Args:
@@ -47,7 +48,7 @@ async def send_requests(batteries: Set[int], request_num: int) -> List[Result]:
     """
     battery_pool = microgrid.battery_pool(batteries)
     results_rx = battery_pool.power_distribution_results()
-    result: List[Result] = []
+    result: list[Result] = []
     for _ in range(request_num):
         await battery_pool.set_power(Power(float(random.randrange(100000, 1000000))))
         try:
@@ -61,7 +62,7 @@ async def send_requests(batteries: Set[int], request_num: int) -> List[Result]:
     return result
 
 
-def parse_result(result: List[List[Result]]) -> Dict[str, float]:
+def parse_result(result: list[list[Result]]) -> dict[str, float]:
     """Parse result.
 
     Args:
@@ -91,8 +92,8 @@ def parse_result(result: List[List[Result]]) -> Dict[str, float]:
 
 async def run_test(  # pylint: disable=too-many-locals
     num_requests: int,
-    batteries: Set[int],
-) -> Dict[str, Any]:
+    batteries: set[int],
+) -> dict[str, Any]:
     """Run test.
 
     Args:
@@ -112,7 +113,7 @@ async def run_test(  # pylint: disable=too-many-locals
         requests_receiver=power_request_channel.new_receiver(),
         battery_status_sender=battery_status_channel.new_sender(),
     ):
-        tasks: List[Coroutine[Any, Any, List[Result]]] = []
+        tasks: list[Coroutine[Any, Any, list[Result]]] = []
         tasks.append(send_requests(batteries, num_requests))
 
         result = await asyncio.gather(*tasks)
@@ -133,7 +134,7 @@ async def run() -> None:
         HOST, PORT, ResamplerConfig(resampling_period=timedelta(seconds=1.0))
     )
 
-    all_batteries: Set[Component] = connection_manager.get().component_graph.components(
+    all_batteries: set[Component] = connection_manager.get().component_graph.components(
         component_category={ComponentCategory.BATTERY}
     )
     batteries_ids = {c.component_id for c in all_batteries}

benchmarks/timeseries/benchmark_datasourcing.py

@@ -14,7 +14,7 @@
 import sys
 import tracemalloc
 from time import perf_counter
-from typing import Any, Tuple
+from typing import Any
 
 from frequenz.channels import Broadcast, Receiver, ReceiverStoppedError
 
@@ -131,7 +131,7 @@ async def consume(channel: Receiver[Any]) -> None:
         )
 
 
-def parse_args() -> Tuple[int, int, bool]:
+def parse_args() -> tuple[int, int, bool]:
     """Parse the command line arguments.
 
     Returns:

benchmarks/timeseries/benchmark_ringbuffer.py

@@ -6,7 +6,7 @@
 import random
 import timeit
 from datetime import datetime, timedelta, timezone
-from typing import Any, Dict, TypeVar
+from typing import Any, TypeVar
 
 import numpy as np
 
@@ -66,7 +66,7 @@ def test_slices(days: int, buffer: OrderedRingBuffer[Any], median: bool) -> None
                 total += float(np.average(minutes))
 
 
-def test_29_days_list(num_runs: int) -> Dict[str, float]:
+def test_29_days_list(num_runs: int) -> dict[str, float]:
     """Run the 29 day test on the list backend."""
     days = 29
     buffer = OrderedRingBuffer([0.0] * MINUTES_IN_29_DAYS, timedelta(minutes=1))
@@ -76,7 +76,7 @@ def test_29_days_list(num_runs: int) -> Dict[str, float]:
     return {"fill": fill_time, "test": test_time}
 
 
-def test_29_days_array(num_runs: int) -> Dict[str, float]:
+def test_29_days_array(num_runs: int) -> dict[str, float]:
     """Run the 29 day test on the array backend."""
     days = 29
     buffer = OrderedRingBuffer(
@@ -91,7 +91,7 @@ def test_29_days_array(num_runs: int) -> Dict[str, float]:
     return {"fill": fill_time, "test": test_time}
 
 
-def test_29_days_slicing_list(num_runs: int) -> Dict[str, float]:
+def test_29_days_slicing_list(num_runs: int) -> dict[str, float]:
     """Run slicing tests on list backend."""
     days = 29
     buffer = OrderedRingBuffer([0.0] * MINUTES_IN_29_DAYS, timedelta(minutes=1))
@@ -107,7 +107,7 @@ def test_29_days_slicing_list(num_runs: int) -> Dict[str, float]:
     return {"fill": fill_time, "median": median_test_time, "avg": avg_test_time}
 
 
-def test_29_days_slicing_array(num_runs: int) -> Dict[str, float]:
+def test_29_days_slicing_array(num_runs: int) -> dict[str, float]:
     """Run slicing tests on array backend."""
     days = 29
     buffer = OrderedRingBuffer(

benchmarks/timeseries/periodic_feature_extractor.py

@@ -9,7 +9,6 @@
 implementation.
 """
 
-from __future__ import annotations
 
 import asyncio
 import collections.abc
@@ -18,7 +17,6 @@
 from datetime import datetime, timedelta, timezone
 from functools import partial
 from timeit import timeit
-from typing import List
 
 import numpy as np
 from frequenz.channels import Broadcast
@@ -79,7 +77,7 @@ def _calculate_avg_window_py(
     feature_extractor: PeriodicFeatureExtractor,
     window: NDArray[np.float_],
     window_size: int,
-    weights: List[float] | None = None,
+    weights: list[float] | None = None,
 ) -> NDArray[np.float_]:
     """
     Plain python version of the average calculator.

benchmarks/timeseries/resampling.py

@@ -3,9 +3,9 @@
 
 """Benchmark resampling."""
 
+from collections.abc import Sequence
 from datetime import datetime, timedelta, timezone
 from timeit import timeit
-from typing import Sequence
 
 from frequenz.sdk.timeseries import Sample
 from frequenz.sdk.timeseries._quantities import Quantity

benchmarks/timeseries/ringbuffer_memusage.py

@@ -3,7 +3,6 @@
 
 """Memory allocation benchmark for the ringbuffer."""
 
-from __future__ import annotations
 
 import argparse
 import tracemalloc

benchmarks/timeseries/ringbuffer_serialization.py

@@ -3,7 +3,6 @@
 
 """Benchmarks the serialization of the `OrderedRingBuffer` class."""
 
-from __future__ import annotations
 
 import fnmatch
 import os

examples/battery_pool.py

@@ -3,12 +3,11 @@
 
 """Script with an example how to use BatteryPool."""
 
-from __future__ import annotations
 
 import asyncio
 import logging
 from datetime import timedelta
-from typing import Any, Dict
+from typing import Any
 
 from frequenz.channels import Receiver
 from frequenz.channels.util import MergeNamed
@@ -32,7 +31,7 @@ async def main() -> None:
     )
 
     battery_pool = microgrid.battery_pool()
-    receivers: Dict[str, Receiver[Any]] = {
+    receivers: dict[str, Receiver[Any]] = {
         "soc": battery_pool.soc.new_receiver(maxsize=1),
         "capacity": battery_pool.capacity.new_receiver(maxsize=1),
         "power_bounds": battery_pool.power_bounds.new_receiver(maxsize=1),

src/frequenz/sdk/_internal/_asyncio.py

@@ -3,7 +3,6 @@
 
 """General purpose async tools."""
 
-from __future__ import annotations
 
 import asyncio
 from abc import ABC

src/frequenz/sdk/_internal/_singleton_meta.py

@@ -4,13 +4,13 @@
 """Definition of Singleton metaclass."""
 
 from threading import Lock
-from typing import Any, Dict
+from typing import Any
 
 
 class SingletonMeta(type):
     """This is a thread-safe implementation of Singleton."""
 
-    _instances: Dict[Any, type] = {}
+    _instances: dict[Any, type] = {}
     """The dictionary of instances of the singleton classes."""
 
     _lock: Lock = Lock()

src/frequenz/sdk/actor/_channel_registry.py

@@ -3,7 +3,7 @@
 
 """A class that would dynamically create, own and provide access to channels."""
 
-from typing import Any, Dict
+from typing import Any
 
 from frequenz.channels import Broadcast, Receiver, Sender
 
@@ -22,7 +22,7 @@ def __init__(self, *, name: str) -> None:
             name: A unique name for the registry.
         """
         self._name = name
-        self._channels: Dict[str, Broadcast[Any]] = {}
+        self._channels: dict[str, Broadcast[Any]] = {}
 
     def new_sender(self, key: str) -> Sender[Any]:
         """Get a sender to a dynamically created channel with the given key.

src/frequenz/sdk/actor/_data_sourcing/microgrid_api_source.py

@@ -5,7 +5,8 @@
 
 import asyncio
 import logging
-from typing import Any, Callable, Dict, List, Optional, Tuple
+from collections.abc import Callable
+from typing import Any
 
 from frequenz.channels import Receiver, Sender
 
@@ -23,7 +24,7 @@
 from .._channel_registry import ChannelRegistry
 from ._component_metric_request import ComponentMetricRequest
 
-_MeterDataMethods: Dict[ComponentMetricId, Callable[[MeterData], float]] = {
+_MeterDataMethods: dict[ComponentMetricId, Callable[[MeterData], float]] = {
     ComponentMetricId.ACTIVE_POWER: lambda msg: msg.active_power,
     ComponentMetricId.CURRENT_PHASE_1: lambda msg: msg.current_per_phase[0],
     ComponentMetricId.CURRENT_PHASE_2: lambda msg: msg.current_per_phase[1],
@@ -34,7 +35,7 @@
     ComponentMetricId.FREQUENCY: lambda msg: msg.frequency,
 }
 
-_BatteryDataMethods: Dict[ComponentMetricId, Callable[[BatteryData], float]] = {
+_BatteryDataMethods: dict[ComponentMetricId, Callable[[BatteryData], float]] = {
     ComponentMetricId.SOC: lambda msg: msg.soc,
     ComponentMetricId.SOC_LOWER_BOUND: lambda msg: msg.soc_lower_bound,
     ComponentMetricId.SOC_UPPER_BOUND: lambda msg: msg.soc_upper_bound,
@@ -54,7 +55,7 @@
     ComponentMetricId.TEMPERATURE: lambda msg: msg.temperature,
 }
 
-_InverterDataMethods: Dict[ComponentMetricId, Callable[[InverterData], float]] = {
+_InverterDataMethods: dict[ComponentMetricId, Callable[[InverterData], float]] = {
     ComponentMetricId.ACTIVE_POWER: lambda msg: msg.active_power,
     ComponentMetricId.ACTIVE_POWER_INCLUSION_LOWER_BOUND: lambda msg: (
         msg.active_power_inclusion_lower_bound
@@ -71,7 +72,7 @@
     ComponentMetricId.FREQUENCY: lambda msg: msg.frequency,
 }
 
-_EVChargerDataMethods: Dict[ComponentMetricId, Callable[[EVChargerData], float]] = {
+_EVChargerDataMethods: dict[ComponentMetricId, Callable[[EVChargerData], float]] = {
     ComponentMetricId.ACTIVE_POWER: lambda msg: msg.active_power,
     ComponentMetricId.CURRENT_PHASE_1: lambda msg: msg.current_per_phase[0],
     ComponentMetricId.CURRENT_PHASE_2: lambda msg: msg.current_per_phase[1],
@@ -99,22 +100,20 @@ def __init__(
             registry: A channel registry.  To be replaced by a singleton
                 instance.
         """
-        self._comp_categories_cache: Dict[int, ComponentCategory] = {}
+        self._comp_categories_cache: dict[int, ComponentCategory] = {}
 
-        self.comp_data_receivers: Dict[int, Receiver[Any]] = {}
+        self.comp_data_receivers: dict[int, Receiver[Any]] = {}
         """The dictionary of component IDs to data receivers."""
 
-        self.comp_data_tasks: Dict[int, asyncio.Task[None]] = {}
+        self.comp_data_tasks: dict[int, asyncio.Task[None]] = {}
         """The dictionary of component IDs to asyncio tasks."""
 
         self._registry = registry
-        self._req_streaming_metrics: Dict[
-            int, Dict[ComponentMetricId, List[ComponentMetricRequest]]
+        self._req_streaming_metrics: dict[
+            int, dict[ComponentMetricId, list[ComponentMetricRequest]]
         ] = {}
 
-    async def _get_component_category(
-        self, comp_id: int
-    ) -> Optional[ComponentCategory]:
+    async def _get_component_category(self, comp_id: int) -> ComponentCategory | None:
         """Get the component category of the given component.
 
         Args:
@@ -139,7 +138,7 @@ async def _get_component_category(
     async def _check_battery_request(
         self,
         comp_id: int,
-        requests: Dict[ComponentMetricId, List[ComponentMetricRequest]],
+        requests: dict[ComponentMetricId, list[ComponentMetricRequest]],
     ) -> None:
         """Check if the requests are valid Battery metrics.
 
@@ -162,7 +161,7 @@ async def _check_battery_request(
     async def _check_ev_charger_request(
         self,
         comp_id: int,
-        requests: Dict[ComponentMetricId, List[ComponentMetricRequest]],
+        requests: dict[ComponentMetricId, list[ComponentMetricRequest]],
     ) -> None:
         """Check if the requests are valid EV Charger metrics.
 
@@ -185,7 +184,7 @@ async def _check_ev_charger_request(
     async def _check_inverter_request(
         self,
         comp_id: int,
-        requests: Dict[ComponentMetricId, List[ComponentMetricRequest]],
+        requests: dict[ComponentMetricId, list[ComponentMetricRequest]],
     ) -> None:
         """Check if the requests are valid Inverter metrics.
 
@@ -208,7 +207,7 @@ async def _check_inverter_request(
     async def _check_meter_request(
         self,
         comp_id: int,
-        requests: Dict[ComponentMetricId, List[ComponentMetricRequest]],
+        requests: dict[ComponentMetricId, list[ComponentMetricRequest]],
     ) -> None:
         """Check if the requests are valid Meter metrics.
 
@@ -232,7 +231,7 @@ async def _check_requested_component_and_metrics(
         self,
         comp_id: int,
         category: ComponentCategory,
-        requests: Dict[ComponentMetricId, List[ComponentMetricRequest]],
+        requests: dict[ComponentMetricId, list[ComponentMetricRequest]],
     ) -> None:
         """Check if the requested component and metrics are valid.
 
@@ -289,8 +288,8 @@ def _get_data_extraction_method(
     def _get_metric_senders(
         self,
         category: ComponentCategory,
-        requests: Dict[ComponentMetricId, List[ComponentMetricRequest]],
-    ) -> List[Tuple[Callable[[Any], float], List[Sender[Sample[Quantity]]]]]:
+        requests: dict[ComponentMetricId, list[ComponentMetricRequest]],
+    ) -> list[tuple[Callable[[Any], float], list[Sender[Sample[Quantity]]]]]:
         """Get channel senders from the channel registry for each requested metric.
 
         Args:

src/frequenz/sdk/actor/_resampling.py

@@ -3,7 +3,6 @@
 
 """An actor to resample microgrid component metrics."""
 
-from __future__ import annotations
 
 import asyncio
 import dataclasses

src/frequenz/sdk/actor/_run_utils.py

@@ -28,7 +28,7 @@ async def run(*actors: Actor) -> None:
             actor.start()
 
     # Wait until all actors are done
-    pending_tasks = set(asyncio.create_task(a.wait(), name=str(a)) for a in actors)
+    pending_tasks = {asyncio.create_task(a.wait(), name=str(a)) for a in actors}
     while pending_tasks:
         done_tasks, pending_tasks = await asyncio.wait(
             pending_tasks, return_when=asyncio.FIRST_COMPLETED