From 32ad8a9b683bd7625b5f9aff57e2cedaa4a33bb3 Mon Sep 17 00:00:00 2001 From: Blake Harnden <32446120+bharnden@users.noreply.github.com> Date: Tue, 9 Jun 2020 20:03:32 -0700 Subject: [PATCH] daemon: added type hinting to Session --- daemon/core/emulator/session.py | 70 ++++++++++++++++++--------------- 1 file changed, 39 insertions(+), 31 deletions(-) diff --git a/daemon/core/emulator/session.py b/daemon/core/emulator/session.py index a1e71612..a2b2670b 100644 --- a/daemon/core/emulator/session.py +++ b/daemon/core/emulator/session.py @@ -15,9 +15,17 @@ import time from typing import Any, Callable, Dict, Iterable, List, Optional, Tuple, Type, TypeVar from core import constants, utils +from core.configservice.manager import ConfigServiceManager from core.emane.emanemanager import EmaneManager from core.emane.nodes import EmaneNet -from core.emulator.data import ConfigData, EventData, ExceptionData, FileData, LinkData +from core.emulator.data import ( + ConfigData, + EventData, + ExceptionData, + FileData, + LinkData, + NodeData, +) from core.emulator.distributed import DistributedController from core.emulator.emudata import InterfaceData, LinkOptions, NodeOptions from core.emulator.enumerations import ( @@ -89,62 +97,62 @@ class Session: :param config: session configuration :param mkdir: flag to determine if a directory should be made """ - self.id = _id + self.id: int = _id # define and create session directory when desired - self.session_dir = os.path.join(tempfile.gettempdir(), f"pycore.{self.id}") + self.session_dir: str = os.path.join(tempfile.gettempdir(), f"pycore.{self.id}") if mkdir: os.mkdir(self.session_dir) - self.name = None - self.file_name = None - self.thumbnail = None - self.user = None - self.event_loop = EventLoop() - self.link_colors = {} + self.name: Optional[str] = None + self.file_name: Optional[str] = None + self.thumbnail: Optional[str] = None + self.user: Optional[str] = None + self.event_loop: EventLoop = EventLoop() + self.link_colors: Dict[int, str] = {} # dict of nodes: all nodes and nets - self.nodes = {} + self.nodes: Dict[int, NodeBase] = {} self._nodes_lock = threading.Lock() - self.state = EventTypes.DEFINITION_STATE - self._state_time = time.monotonic() - self._state_file = os.path.join(self.session_dir, "state") + self.state: EventTypes = EventTypes.DEFINITION_STATE + self._state_time: float = time.monotonic() + self._state_file: str = os.path.join(self.session_dir, "state") # hooks handlers - self._hooks = {} - self._state_hooks = {} + self._hooks: Dict[EventTypes, Tuple[str, str]] = {} + self._state_hooks: Dict[EventTypes, Callable[[int], None]] = {} self.add_state_hook( state=EventTypes.RUNTIME_STATE, hook=self.runtime_state_hook ) # handlers for broadcasting information - self.event_handlers = [] - self.exception_handlers = [] - self.node_handlers = [] - self.link_handlers = [] - self.file_handlers = [] - self.config_handlers = [] - self.shutdown_handlers = [] + self.event_handlers: List[Callable[[EventData], None]] = [] + self.exception_handlers: List[Callable[[ExceptionData], None]] = [] + self.node_handlers: List[Callable[[NodeData], None]] = [] + self.link_handlers: List[Callable[[LinkData], None]] = [] + self.file_handlers: List[Callable[[FileData], None]] = [] + self.config_handlers: List[Callable[[ConfigData], None]] = [] + self.shutdown_handlers: List[Callable[[Session], None]] = [] # session options/metadata - self.options = SessionConfig() + self.options: SessionConfig = SessionConfig() if not config: config = {} for key in config: value = config[key] self.options.set_config(key, value) - self.metadata = {} + self.metadata: Dict[str, str] = {} # distributed support and logic - self.distributed = DistributedController(self) + self.distributed: DistributedController = DistributedController(self) # initialize session feature helpers - self.location = GeoLocation() - self.mobility = MobilityManager(session=self) - self.services = CoreServices(session=self) - self.emane = EmaneManager(session=self) - self.sdt = Sdt(session=self) + self.location: GeoLocation = GeoLocation() + self.mobility: MobilityManager = MobilityManager(self) + self.services: CoreServices = CoreServices(self) + self.emane: EmaneManager = EmaneManager(self) + self.sdt: Sdt = Sdt(self) # initialize default node services self.services.default_services = { @@ -156,7 +164,7 @@ class Session: } # config services - self.service_manager = None + self.service_manager: Optional[ConfigServiceManager] = None @classmethod def get_node_class(cls, _type: NodeTypes) -> Type[NodeBase]: