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

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 _publish has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def _publish(self, msg: typing.Union[float, bool, int, str, np.ndarray, np.number], header: Header) -> None:
        if not self._unregistered:
            # todo: check if dtype(msg) == self._dtype?
            # Convert python native types to numpy arrays.
            if isinstance(msg, float):

• Read up
  Read up

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]

Avoid deeply nested control flow statements.

Open

                        if cname_address in self.disconnected[node_name][key]:
                            color = "red"
                        if cname_address in self.connected_rx[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 (

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 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_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 _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 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

Avoid deeply nested control flow statements.

Open

                        with self.feedthroughs as d:
                            d[cname] = mapping_ft
                elif component == "inputs":

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 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 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 callback has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    def callback(self, t_n: float, image: Optional[eagerx.utils.utils.Msg] = None):
        # Fill output_msg with 'done' output --> signals that we are done rendering
        output_msgs = dict(done=0)
        # Grab latest image
        self.last_image = image.msgs[-1] if self.encoding == "rgb" else cv2.cvtColor(image.msgs[-1], cv2.COLOR_BGR2RGB)

• 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

Avoid too many return statements within this function.

Open

        return float(value)

Avoid too many return statements within this function.

Open

            return True

Avoid too many return statements within this function.

Open

            return logsilent