Skip to content

Commit

Permalink
resolve missing type hints (work in progress)
Browse files Browse the repository at this point in the history
  • Loading branch information
ClaasRostock committed Jan 25, 2025
1 parent f61fa62 commit 9f1b3fe
Show file tree
Hide file tree
Showing 4 changed files with 52 additions and 11 deletions.
24 changes: 18 additions & 6 deletions src/sim_explorer/case.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@
from collections.abc import Callable, Generator, Iterable, Iterator, Sequence
from datetime import datetime
from pathlib import Path
from typing import Any
from typing import TYPE_CHECKING, Any

import matplotlib.pyplot as plt
import numpy as np
Expand All @@ -23,7 +23,9 @@
from sim_explorer.system_interface_osp import SystemInterfaceOSP
from sim_explorer.utils.misc import from_xml
from sim_explorer.utils.paths import get_path, relative_path
from sim_explorer.utils.types import TValue

if TYPE_CHECKING:
from sim_explorer.utils.types import TActionArgs, TValue

"""
sim_explorer module for definition and execution of simulation experiments
Expand Down Expand Up @@ -51,7 +53,7 @@ class Case:
spec (dict): the dictionary of the case specification
"""

def __init__(
def __init__( # noqa: C901
self,
cases: Cases,
name: str,
Expand Down Expand Up @@ -403,7 +405,7 @@ def get_from_config(element: str, default: float | None = None) -> float | None:
raise CaseInitError("'stepSize' should be specified as part of the 'base' specification.") from None
return special

def run(self, dump: str | None = "") -> None:
def run(self, dump: str | None = "") -> None: # noqa: C901, PLR0912, PLR0915
"""Set up case and run it.
All get action are recorded in results and get actions always concern whole case variables.
Expand All @@ -417,7 +419,7 @@ def run(self, dump: str | None = "") -> None:

def do_actions(
_t: float,
actions: list[tuple[Any, ...]],
actions: list[TActionArgs],
_iter: Iterator[tuple[float, list[tuple[Any, ...]]]],
time: int | float,
) -> tuple[float, list[tuple[Any, ...]]]:
Expand Down Expand Up @@ -462,13 +464,23 @@ def do_actions(
self.add_results_object(Results(self))

while True:
t_get: float | int
a_get: list[TActionArgs]
try:
t_get, a_get = next(get_iter)
except StopIteration:
t_get, a_get = (tstop + 1, [])
if t_get < 0: # negative time indicates 'always'
for a in a_get:
act_step.append((*a, self.cases.simulator.action_step(a, self.cases.variables[a[0]]["type"])))
act_step.append(
(
*a,
self.cases.simulator.action_step(
act_info=a,
typ=self.cases.variables[a[0]]["type"],
),
)
)
else:
break

Expand Down
6 changes: 3 additions & 3 deletions src/sim_explorer/system_interface.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@
from sim_explorer.json5 import Json5
from sim_explorer.utils.misc import from_xml, match_with_wildcard
from sim_explorer.utils.osp import read_system_structure_xml
from sim_explorer.utils.types import TValue
from sim_explorer.utils.types import TActionArgs, TValue

if TYPE_CHECKING:
from xml.etree.ElementTree import Element
Expand Down Expand Up @@ -610,11 +610,11 @@ def add_actions( # noqa: PLR0913
else:
raise KeyError(f"Unknown action type {act_type} at time {at_time}")

def do_action(self, time: int | float, act_info: tuple[Any, ...], typ: type) -> bool:
def do_action(self, time: int | float, act_info: TActionArgs, typ: type) -> bool:
"""Do the action described by the tuple using OSP functions."""
raise NotImplementedError("The method 'do_action()' cannot be used in SystemInterface") from None

def action_step(self, act_info: tuple[Any, ...], typ: type) -> Callable[..., Any]:
def action_step(self, act_info: TActionArgs, typ: type) -> Callable[..., Any]:
"""Pre-compile the step action and return the partial function
so that it can be called at communication points.
"""
Expand Down
5 changes: 3 additions & 2 deletions src/sim_explorer/system_interface_osp.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
from libcosimpy.CosimObserver import CosimObserver

from sim_explorer.system_interface import SystemInterface
from sim_explorer.utils.types import TActionArgs


class SystemInterfaceOSP(SystemInterface):
Expand Down Expand Up @@ -84,7 +85,7 @@ def _action_func(self, act_type: int, var_type: type) -> Callable[..., Any]:
str: self.observer.last_string_values,
}[var_type]

def do_action(self, time: int | float, act_info: tuple[Any, ...], typ: type) -> bool:
def do_action(self, time: int | float, act_info: TActionArgs, typ: type) -> bool:
"""Do the action described by the tuple using OSP functions."""
if len(act_info) == 4: # set action # noqa: PLR2004
cvar, comp, refs, values = act_info
Expand All @@ -100,7 +101,7 @@ def do_action(self, time: int | float, act_info: tuple[Any, ...], typ: type) ->
assert time >= 0, "Get actions for all communication points shall be pre-compiled"
return self._action_func(2, typ)(_comp, refs)

def action_step(self, act_info: tuple[Any, ...], typ: type) -> Callable[..., Any]:
def action_step(self, act_info: TActionArgs, typ: type) -> Callable[..., Any]:
"""Pre-compile the step action and return the partial function
so that it can be called at communication points.
"""
Expand Down
28 changes: 28 additions & 0 deletions src/sim_explorer/utils/types.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,3 +12,31 @@
TDataTable: TypeAlias = Sequence[TDataRow] # X table
TArguments: TypeAlias = Mapping[str, TValue] # X row with variable names
TArgs: TypeAlias = dict[str, TValue]


# ===== System Interface =====================================================================
#: Arguments for 'get' action functions (component_variable_name, component_name, variable_references)
TGetActionArgs: TypeAlias = tuple[str, str, tuple[int, ...]]
#: Arguments for 'set' action functions (component_variable_name, component_name, variable_references, variable_values)
TSetActionArgs: TypeAlias = tuple[str, str, tuple[int, ...], tuple[TValue, ...]]
#: Arguments for action functions
TActionArgs: TypeAlias = TGetActionArgs | TSetActionArgs

#: Function signature for action functions
TActionFunc: TypeAlias = callable[[int | float, TActionArgs, type], bool]
#: Function signature for action step functions
TActionStepFunc: TypeAlias = callable[[TActionArgs, type], callable[..., TValue]]

#: Function signature for initial action functions
TInitialActionFunc: TypeAlias = callable[[int | float, TSetActionArgs], bool]
#: Function signature for initial action functions
TInitialActionStepFunc: TypeAlias = callable[[TSetActionArgs], callable[..., TValue]]

#: Function signature for get action functions
TGetActionFunc: TypeAlias = callable[[TGetActionArgs], TValue]
#: Function signature for set action functions
TSetActionFunc: TypeAlias = callable[[TSetActionArgs], bool]
#: Function signature for get action step functions
TGetActionStepFunc: TypeAlias = callable[[TGetActionArgs], callable[..., TValue]]
#: Function signature for set action step functions
TSetActionStepFunc: TypeAlias = callable[[TSetActionArgs], callable[..., TValue]]

0 comments on commit 9f1b3fe

Please sign in to comment.