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

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