Function remove has a Cognitive Complexity of 19 (exceeds 5 allowed). Consider refactoring.

Open

    def remove(self, names: Union[str, EntitySpec, List[Union[str, EntitySpec]]], remove: bool = False) -> List:
        """Removes nodes/objects from the graph.

        - First, all associated connections are disconnected.

• Read up
  Read up

Function build has a Cognitive Complexity of 18 (exceeds 5 allowed). Consider refactoring.

Open

    def build(self, ns: str):
        params = self.params  # Creates a deepcopy
        name = self.config.name
        entity_id = self.config.entity_id

• Read up
  Read up

Function resolve_args has a Cognitive Complexity of 18 (exceeds 5 allowed). Consider refactoring.

Open

def resolve_args(arg_str, context=None, resolve_anon=True, filename=None, only=None):
    """
    Resolves substitution args (see wiki spec U{http://ros.org/wiki/roslaunch}).

    @param arg_str: string to resolve zero or more substitution args

• Read up
  Read up

Consider simplifying this complex logical expression.

Open

        if shape is not None:
            shape = tuple(shape)
            assert low is None or np.isscalar(low) or low.shape == shape, "low.shape doesn't match provided shape"
            assert high is None or np.isscalar(high) or high.shape == shape, "high.shape doesn't match provided shape"
        elif low is not None and not np.isscalar(low):

Function substitute_args has a Cognitive Complexity of 17 (exceeds 5 allowed). Consider refactoring.

Open

def substitute_args(
    param: Union[str, Dict],
    context: Optional[Dict] = None,
    only: Optional[List[str]] = None,
):

• Read up
  Read up

Function episode_graph has a Cognitive Complexity of 17 (exceeds 5 allowed). Consider refactoring.

Open

def episode_graph(G):
    # Create a shallow copy graph that excludes "skip" edges
    H = nx.MultiDiGraph(G)
    for u, v, key, data in G.edges(data=True, keys=True):
        if data["skip"]:

• Read up
  Read up

Function __init__ has a Cognitive Complexity of 17 (exceeds 5 allowed). Consider refactoring.

Open

    def __init__(
        self,
        low: t.Optional[t.Union[t.List, np.ndarray, int, float, bool, t.Tuple]] = None,
        high: t.Optional[t.Union[t.List, np.ndarray, int, float, bool, t.Tuple]] = None,
        shape: t.Optional[t.Tuple] = None,

• Read up
  Read up

Function merge has a Cognitive Complexity of 16 (exceeds 5 allowed). Consider refactoring.

Open

def merge(a, b, path=None):
    """merges b into a"""
    # If it is a spec, convert to params
    if path is None:
        path = []

• Read up
  Read up

Function is_supported_type has a Cognitive Complexity of 15 (exceeds 5 allowed). Consider refactoring.

Open

def is_supported_type(param: Any, types: Tuple, none_support):
    if isinstance(param, types) or (param is None and none_support):
        if isinstance(param, dict):
            for key, value in param.items():
                assert isinstance(key, str), f'Invalid key "{key}". Only type "str" is supported as dictionary key.'

• Read up
  Read up

Function init_node has 44 lines of code (exceeds 25 allowed). Consider refactoring.

Open

def init_node(
    ns,
    rate_node,
    node,
    inputs,

Function create has a Cognitive Complexity of 14 (exceeds 5 allowed). Consider refactoring.

Open

    def create(cls, actuators: Optional[List[Dict]] = None, sensors: Optional[List[Dict]] = None):
        nodes = []

        from eagerx.core.entities import EngineNode

• Read up
  Read up

Function __getattr__ has a Cognitive Complexity of 14 (exceeds 5 allowed). Consider refactoring.

Open

    def __getattr__(self, name):
        if keys_exists(self._spec._params, *self._depth, name):
            new_depth = self._depth.copy() + [name]
            d = get_dict(self._spec._params, new_depth)
            if isinstance(d, dict):

• Read up
  Read up

Function _is_supported_type has a Cognitive Complexity of 14 (exceeds 5 allowed). Consider refactoring.

Open

def _is_supported_type(param):
    try:
        is_supported_type(param, supported_types, none_support=True)
    except TypeError as t:
        from eagerx.core.specs import EntitySpec

• Read up
  Read up

Function to_dict has a Cognitive Complexity of 13 (exceeds 5 allowed). Consider refactoring.

Open

    def to_dict(self) -> t.Dict:
        """Convert the space to a dict representation

        :return: Dict representation of the space.
        """

• Read up
  Read up

Function with_latest_from has a Cognitive Complexity of 13 (exceeds 5 allowed). Consider refactoring.

Open

def with_latest_from(*sources: Observable):
    def _with_latest_from(parent: Observable) -> Observable:
        NO_VALUE = NotSet()

        def subscribe(observer, scheduler=None):

• Read up
  Read up

Function run has a Cognitive Complexity of 13 (exceeds 5 allowed). Consider refactoring.

Open

def run(LOG_DIR, rate, sync, rtf, num_eps, num_steps, actions):
    eagerx.set_log_level(eagerx.DEBUG)

    # Initialize empty graph
    graph = eagerx.Graph.create()

• Read up
  Read up

Function run has 36 lines of code (exceeds 25 allowed). Consider refactoring.

Open

def run(LOG_DIR, rate, sync, rtf, num_eps, num_steps, actions):
    eagerx.set_log_level(eagerx.DEBUG)

    # Initialize empty graph
    graph = eagerx.Graph.create()

Function _shutdown has a Cognitive Complexity of 12 (exceeds 5 allowed). Consider refactoring.

Open

    def _shutdown(self):
        if not self.has_shutdown:
            self._shutdown_srv.unregister()
            for address, node in self._launch_nodes.items():
                self.backend.logdebug(f"[{self.name}] Send termination signal to '{address}'.")

• Read up
  Read up

Function initialize_nodes has a Cognitive Complexity of 12 (exceeds 5 allowed). Consider refactoring.

Open

def initialize_nodes(
    nodes: Union["BaseNodeSpec", List["BaseNodeSpec"]],
    process_id: int,
    ns: str,
    message_broker: Any,

• Read up
  Read up

Function merge has a Cognitive Complexity of 12 (exceeds 5 allowed). Consider refactoring.

Open

def merge(a: typing.Dict, b: typing.Dict, path=None):
    """merges b into a"""
    if path is None:
        path = []
    for key in b:

• Read up
  Read up