added type aliases

src/tickit/core/components/component.py

@@ -3,13 +3,15 @@
 import traceback
 from abc import abstractmethod
 from dataclasses import dataclass
-from typing import Dict, Optional, Type, Union
+from typing import Dict, Optional, Type
 
 from tickit.core.state_interfaces.state_interface import StateConsumer, StateProducer
 from tickit.core.typedefs import (
     Changes,
     ComponentException,
     ComponentID,
+    ComponentInput,
+    ComponentOutput,
     ComponentPort,
     Input,
     Interrupt,
@@ -75,14 +77,14 @@ def __call__(self) -> Component:
 class BaseComponent(Component):
     """A base class for components, implementing state interface related methods."""
 
-    state_consumer: StateConsumer[Union[Input, StopComponent]]
-    state_producer: StateProducer[Union[Interrupt, Output, ComponentException]]
+    state_consumer: StateConsumer[ComponentInput]
+    state_producer: StateProducer[ComponentOutput]
 
-    async def handle_input(self, message: Union[Input, StopComponent]):
+    async def handle_input(self, message: ComponentInput):
         """Call on_tick when an input is received.
 
         Args:
-            message (Union[Input, StopComponent])): An immutable data container for any
+            message (ComponentInput): An immutable data container for any
                 message a component receives.
         """
         if isinstance(message, Input):

src/tickit/core/management/schedulers/base.py

@@ -9,6 +9,8 @@
 from tickit.core.typedefs import (
     ComponentException,
     ComponentID,
+    ComponentInput,
+    ComponentOutput,
     Input,
     Interrupt,
     Output,
@@ -67,9 +69,7 @@ async def update_component(self, input: Input) -> None:
         """
         await self.state_producer.produce(input_topic(input.target), input)
 
-    async def handle_message(
-        self, message: Union[Interrupt, Output, ComponentException]
-    ) -> None:
+    async def handle_message(self, message: ComponentOutput) -> None:
         """Handle messages received by the state consumer.
 
         An asynchronous callback which handles Interrupt, Output and ComponentException
@@ -100,15 +100,13 @@ async def setup(self) -> None:
         producer to produce component inputs.
         """
         self.ticker = Ticker(self._wiring, self.update_component)
-        self.state_consumer: StateConsumer[
-            Union[Interrupt, Output, ComponentException]
-        ] = self._state_consumer_cls(self.handle_message)
+        self.state_consumer: StateConsumer[ComponentOutput] = self._state_consumer_cls(
+            self.handle_message
+        )
         await self.state_consumer.subscribe(
             {output_topic(component) for component in self.ticker.components}
         )
-        self.state_producer: StateProducer[
-            Union[Input, StopComponent, ComponentException]
-        ] = self._state_producer_cls()
+        self.state_producer: StateProducer[ComponentInput] = self._state_producer_cls()
 
     def add_wakeup(self, component: ComponentID, when: SimTime) -> None:
         """Adds a wakeup to the mapping.

src/tickit/core/typedefs.py

@@ -1,5 +1,5 @@
 from dataclasses import dataclass
-from typing import Hashable, Iterator, Mapping, NewType, Optional
+from typing import Hashable, Iterator, Mapping, NewType, Optional, Union
 
 from apischema import deserializer, serializer
 from immutables import Map
@@ -126,3 +126,7 @@ class ComponentException:
     source: ComponentID
     error: Exception
     traceback: str
+
+
+ComponentOutput = Union[Interrupt, Output, ComponentException]
+ComponentInput = Union[Input, StopComponent, ComponentException]