Avoid deeply nested control flow statements.

Open

                    if not G.nodes[v]["always_active"]:
                        new_stale_nodes.append(v)
                        G.nodes[v]["is_stale"] = True
    remain_active = [n for n, is_stale in nx.get_node_attributes(G, "remain_active").items() if is_stale]

Function engine has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

def engine(engine_cls: "Engine", entity=None) -> Callable:
    """A decorator to register an engine implementation of an :class:`~eagerx.core.entities.Object`.

    .. note:: In our running example, the :func:`~eagerx.core.entities.Object.example_engine` method would be decorated.

• Read up
  Read up

Function init_node has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

def init_node(
    ns,
    rate_node,
    node,
    inputs,

• Read up
  Read up

Function init_state_resets has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

def init_state_resets(ns, state_inputs, trigger, scheduler, tp_scheduler, node):
    if len(state_inputs) > 0:
        channels = []

        for s in state_inputs:

• Read up
  Read up

Function init_engine has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

def init_engine(
    ns,
    rate_node,
    node,
    inputs_init,

• Read up
  Read up

Function initialize has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def initialize(self, spec: NodeSpec):
        # Define observation buffers
        self.observation_buffer = dict()
        for cname, i in self.inputs.items():
            if isinstance(i["space"], dict):

• Read up
  Read up

Function _ns has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

def _ns(resolved, a, args, context):
    """
    process $(ns) arg

    :returns: updated resolved argument, ``str``

• Read up
  Read up

Function get_param_with_blocking has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

def get_param_with_blocking(name, backend, default=_unspecified, timeout=2.0):
    params = Unspecified()
    start = time.time()
    while isinstance(params, Unspecified):
        if time.time() - start > timeout:

• Read up
  Read up

Avoid deeply nested control flow statements.

Open

                        if cname_address in self.connected_rx[node_name][key]:
                            assert color is None, f"Duplicate connection status for address ({cname_address})."
                            color = "green"
                        if cname_address in self.connected_bnd[node_name][key]:

Avoid deeply nested control flow statements.

Open

                        if cname_address in self.connected_bnd[node_name][key]:
                            assert color is None, f"Duplicate connection status for address ({cname_address})."
                            color = "blue"
                        assert (

Function node_shutdown has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def node_shutdown(self):
        if not self.has_shutdown:
            self.backend.logdebug(f"[{self.name}] RxEngine.node_shutdown() called.")
            for address, node in self.engine.launch_nodes.items():
                self.backend.loginfo(f"[{self.name}] Send termination signal to '{address}'.")

• Read up
  Read up

Function make has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def make(
        cls,
        name: str,
        sensors: Optional[List[str]] = None,
        env_id: str = "Pendulum-v1",

• Read up
  Read up

Function _connect has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def _connect(
        self,
        source: SpecView = None,
        target: SpecView = None,
        window: Optional[int] = None,

• Read up
  Read up

Function _config has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

def _config(resolved, a, args, context):
    """
    process $(config) arg

    :returns: updated resolved argument, ``str``

• Read up
  Read up

Function init_state_inputs_channel has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

def init_state_inputs_channel(ns, state_inputs, scheduler, node):
    if len(state_inputs) > 0:
        channels = []
        for s in state_inputs:
            d = s["done"].pipe(

• Read up
  Read up

Avoid deeply nested control flow statements.

Open

                        if mode == 0:
                            i[key] = filter_info_for_printing(msg.info)
                        else:
                            i[key] = msg.msgs
        else:

Function cb_ft has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

def cb_ft(cb_input, sync):
    # Fill output msg with number of node ticks
    output_msgs = dict()
    for key, msg in cb_input.items():
        if key not in ["node_tick", "t_n"]:

• Read up
  Read up

Function init_real_reset has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

def init_real_reset(
    ns,
    Nc,
    rate_node,
    RR,

• Read up
  Read up

Function callback has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    def callback(self, t_n: float, tick: Optional[Msg] = None):
        assert isinstance(self.simulator, dict), (
            'Simulator object "%s" is not compatible with this engine node.' % self.simulator
        )
        with self._cond:

• Read up
  Read up

Function _add_engine_states has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    def _add_engine_states(self, name: str, spec: ObjectSpec):
        # Pop states that were not implemented.
        assert name in self.objects, f"There is no Object called `{name}' in engine.objects. First add the Object."
        states = spec.engine.states
        for cname in list(states.keys()):

• Read up
  Read up