Support 3.9

Signed-off-by: Kurt Greaves <kurt.greaves@zepben.com>

Author: kgreav

src/zepben/evolve/services/network/tracing/feeder/assign_to_lv_feeders.py

@@ -3,7 +3,7 @@
 #  License, v. 2.0. If a copy of the MPL was not distributed with this
 #  file, You can obtain one at https://mozilla.org/MPL/2.0/.
 from collections.abc import Iterable
-from typing import Collection, List, Generator
+from typing import Collection, List, Generator, TypeVar
 
 from zepben.evolve import Switch, AuxiliaryEquipment, ProtectedSwitch, Traversal
 from zepben.evolve.model.cim.iec61970.base.core.conducting_equipment import ConductingEquipment
@@ -19,6 +19,8 @@
 from zepben.evolve.services.network.tracing.networktrace.tracing import Tracing
 from zepben.evolve.services.network.tracing.traversal.step_context import StepContext
 
+T = TypeVar("T")
+
 __all__ = ["AssignToLvFeeders"]
 
 
@@ -93,7 +95,7 @@ async def run_with_feeders(self,
     async def _create_trace(self,
                       terminal_to_aux_equipment: dict[Terminal, list[AuxiliaryEquipment]],
                       lv_feeder_start_points: Iterable[ConductingEquipment],
-                      lv_feeders_to_assign: list[LvFeeder]) -> NetworkTrace[...]:
+                      lv_feeders_to_assign: list[LvFeeder]) -> NetworkTrace[T]:
 
         def _reached_hv(ce: ConductingEquipment):
             return True if ce.base_voltage and ce.base_voltage.nominal_voltage >= 1000 else False

src/zepben/evolve/services/network/tracing/feeder/clear_direction.py

@@ -4,7 +4,7 @@
 #  file, You can obtain one at https://mozilla.org/MPL/2.0/.
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 from zepben.evolve.model.cim.iec61970.base.core.terminal import Terminal
 
@@ -49,7 +49,7 @@ async def run(self,
     @staticmethod
     def _create_trace(state_operators: NetworkStateOperators,
                       visited_feeder_head_terminals: list[Terminal]
-                      ) -> NetworkTrace[...]:
+                      ) -> NetworkTrace[Any]:
         def queue_condition(step: NetworkTraceStep, context: StepContext, _, __):
             return state_operators.get_direction(step.path.to_terminal) != FeederDirection.NONE
 

src/zepben/evolve/services/network/tracing/networktrace/actions/equipment_tree_builder.py

@@ -4,6 +4,7 @@
 #  file, You can obtain one at https://mozilla.org/MPL/2.0/.
 
 import uuid
+from typing import Any
 
 from zepben.evolve.model.cim.iec61970.base.core.conducting_equipment import ConductingEquipment
 from zepben.evolve.services.network.tracing.networktrace.actions.tree_node import TreeNode
@@ -40,20 +41,20 @@ def __init__(self):
     def roots(self):
         return self._roots.values()
 
-    def compute_initial_value(self, item: NetworkTraceStep[...]) -> EquipmentTreeNode:
+    def compute_initial_value(self, item: NetworkTraceStep[Any]) -> EquipmentTreeNode:
         node = self._roots.get(item.path.to_equipment)
         if node is None:
             node = TreeNode(item.path.to_equipment, None)
             self._roots[item.path.to_equipment] = node
         return node
 
-    def compute_next_value_typed(self, next_item: NetworkTraceStep[...], current_item: NetworkTraceStep[...], current_value: EquipmentTreeNode) -> EquipmentTreeNode:
+    def compute_next_value_typed(self, next_item: NetworkTraceStep[Any], current_item: NetworkTraceStep[Any], current_value: EquipmentTreeNode) -> EquipmentTreeNode:
         if next_item.path.traced_internally:
             return current_value
         else:
             return TreeNode(next_item.path.to_equipment, current_value)
 
-    def apply(self, item: NetworkTraceStep[...], context: StepContext):
+    def apply(self, item: NetworkTraceStep[Any], context: StepContext):
         current_node: TreeNode = self.get_context_value(context)
         if current_node.parent:
             current_node.parent.add_child(current_node)

src/zepben/evolve/services/network/tracing/networktrace/actions/tree_node.py

@@ -3,20 +3,21 @@
 #  License, v. 2.0. If a copy of the MPL was not distributed with this
 #  file, You can obtain one at https://mozilla.org/MPL/2.0/.
 
-from typing import Self, List, TypeVar, Generic
+from typing import List, TypeVar, Generic
 
 from zepben.evolve import IdentifiedObject
 
 T = TypeVar('T')
 
+
 class TreeNode(Generic[T]):
     """
     represents a node in the NetworkTrace tree
     """
     def __init__(self, identified_object: IdentifiedObject, parent=None):
         self.identified_object = identified_object
-        self._parent: Self = parent
-        self._children: List[Self] = []
+        self._parent: TreeNode = parent
+        self._children: List[TreeNode] = []
 
     @property
     def parent(self) -> 'TreeNode[T]':
@@ -26,7 +27,7 @@ def parent(self) -> 'TreeNode[T]':
     def children(self):
         return list(self._children)
 
-    def add_child(self, child: Self):
+    def add_child(self, child: 'TreeNode'):
         self._children.append(child)
 
     def __str__(self):

src/zepben/evolve/services/network/tracing/networktrace/compute_data.py

@@ -2,7 +2,7 @@
 #  This Source Code Form is subject to the terms of the Mozilla Public
 #  License, v. 2.0. If a copy of the MPL was not distributed with this
 #  file, You can obtain one at https://mozilla.org/MPL/2.0/.
-from typing import TypeVar, Generic
+from typing import TypeVar, Generic, Any
 
 from zepben.evolve.services.network.tracing.networktrace.network_trace_step import NetworkTraceStep
 from zepben.evolve.services.network.tracing.traversal.step_context import StepContext
@@ -38,7 +38,7 @@ class ComputeDataWithPaths(Generic[T]):
     def __init__(self, func):
         self._func = func or (lambda *args: None)
 
-    def compute_next(self, current_step: NetworkTraceStep[T], current_context: StepContext, next_path: NetworkTraceStep.Path, next_paths: list[NetworkTraceStep.Path, ...]) -> T:
+    def compute_next(self, current_step: NetworkTraceStep[T], current_context: StepContext, next_path: NetworkTraceStep.Path, next_paths: list[NetworkTraceStep.Path, Any]) -> T:
         """
         Called for each new NetworkTraceStep in a NetworkTrace. The value returned from this function
         will be stored against the next step within NetworkTraceStep. data.

src/zepben/evolve/services/network/tracing/networktrace/network_trace_tracker.py

@@ -2,7 +2,7 @@
 #  This Source Code Form is subject to the terms of the Mozilla Public
 #  License, v. 2.0. If a copy of the MPL was not distributed with this
 #  file, You can obtain one at https://mozilla.org/MPL/2.0/.
-from typing import Set
+from typing import Set, Any
 
 from zepben.evolve import Terminal, SinglePhaseKind
 
@@ -31,7 +31,7 @@ def clear(self):
         self._visited.clear()
 
     @staticmethod
-    def _get_key(terminal: Terminal, phases: Set[SinglePhaseKind]) -> ... :
+    def _get_key(terminal: Terminal, phases: Set[SinglePhaseKind]) -> Any:
         if phases and len(phases) < 1:
             return terminal
         else:

src/zepben/evolve/services/network/tracing/traversal/weighted_priority_queue.py

@@ -3,7 +3,7 @@
 #  License, v. 2.0. If a copy of the MPL was not distributed with this
 #  file, You can obtain one at https://mozilla.org/MPL/2.0/.
 from collections import defaultdict
-from typing import TypeVar, Callable, Iterable
+from typing import TypeVar, Callable, Iterable, Any
 
 from zepben.evolve.services.network.tracing.traversal.queue import TraversalQueue
 
@@ -26,7 +26,7 @@ class WeightedPriorityQueue(TraversalQueue[T]):
     :param queue_provider: A queue provider. This allows you to customise the priority of items with the same weight.
     :param get_weight:     A method to extract the weight of an item being added to the queue.
     """
-    def __init__(self, queue_provider: Callable[[], TraversalQueue[T]], get_weight: Callable[[...], int]):
+    def __init__(self, queue_provider: Callable[[], TraversalQueue[T]], get_weight: Callable[[Any], int]):
         self._queue_provider = queue_provider
         self._get_weight = get_weight
         super().__init__(queue=SortedDefaultDict(self._queue_provider))

src/zepben/evolve/streaming/data/set_current_states_status.py

@@ -7,7 +7,7 @@
 
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
-from typing import List, Optional, Tuple
+from typing import List, Optional, Tuple, Any
 
 from zepben.protobuf.ns.data.change_status_pb2 import BatchSuccessful as PBBatchSuccessful, BatchFailure as PBBatchFailure, \
     BatchNotProcessed as PBBatchNotProcessed, StateEventFailure as PBStateEventFailure, StateEventUnknownMrid as PBStateEventUnknownMrid, \
@@ -88,7 +88,7 @@ class BatchFailure(SetCurrentStatesStatus):
         failures: The status of each item processed in the batch that failed.
     """
 
-    def __init__(self, batch_id: int, partial_failure: bool, failures: Tuple['StateEventFailure', ...]):
+    def __init__(self, batch_id: int, partial_failure: bool, failures: Tuple['StateEventFailure', Any]):
         super().__init__(batch_id)
         self.partial_failure = partial_failure
         self.failures = failures

src/zepben/evolve/streaming/mutations/update_network_state_service.py

@@ -4,7 +4,7 @@
 #  file, You can obtain one at https://mozilla.org/MPL/2.0/.
 __all__ = ["UpdateNetworkStateService"]
 
-from typing import Tuple, Callable, AsyncGenerator
+from typing import Tuple, Callable, AsyncGenerator, Any
 
 from zepben.protobuf.ns.network_state_pb2_grpc import UpdateNetworkStateServiceServicer
 from zepben.protobuf.ns.network_state_requests_pb2 import SetCurrentStatesRequest as PBSetCurrentStatesRequest
@@ -29,7 +29,7 @@ class UpdateNetworkStateService(UpdateNetworkStateServiceServicer):
     """
 
     def __init__(self, on_set_current_states: Callable[
-        [AsyncGenerator[Tuple[int, Tuple[CurrentStateEvent, ...]], None]], AsyncGenerator[SetCurrentStatesStatus, None]]):
+        [AsyncGenerator[Tuple[int, Tuple[CurrentStateEvent, Any]], None]], AsyncGenerator[SetCurrentStatesStatus, None]]):
         self.on_set_current_states = on_set_current_states
 
     async def setCurrentStates(self, request_iterator: AsyncGenerator[PBSetCurrentStatesRequest, None], context) -> AsyncGenerator[
@@ -52,7 +52,7 @@ async def setCurrentStates(self, request_iterator: AsyncGenerator[PBSetCurrentSt
             A stream of protobuf SetCurrentStatesResponse sent back.
         """
 
-        async def request_generator() -> AsyncGenerator[Tuple[int, Tuple[CurrentStateEvent, ...]], None]:
+        async def request_generator() -> AsyncGenerator[Tuple[int, Tuple[CurrentStateEvent, Any]], None]:
             async for request in request_iterator:
                 yield request.messageId, tuple([CurrentStateEvent.from_pb(event) for event in request.event])
 

src/zepben/evolve/testing/test_network_builder.py

@@ -5,7 +5,7 @@
 from zepben.evolve.services.network.tracing.networktrace.operators.network_state_operators import NetworkStateOperators
 from zepben.evolve.services.network.tracing.networktrace.tracing import Tracing
 try:
-    from typing import Protocol
+    from typing import Protocol, Any
 except ImportError:
     Protocol = object
 
@@ -27,7 +27,7 @@ def null_action(_):
 class OtherCreator(Protocol):
     """Type hint class"""
 
-    def __call__(self, mrid: str, *args, **kwargs) -> ConductingEquipment: ...
+    def __call__(self, mrid: str, *args, **kwargs) -> ConductingEquipment: Any
 
 
 class TestNetworkBuilder: