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

Open

def color_nodes(G):
    # Color nodes based on in/out going edges
    for n, data in G.nodes(data=True):
        if data["is_stale"]:
            facecolor = "lightgrey"

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

Open

def wait_for_node_initialization(is_initialized, backend, wait_time=0.3):
    iter = 0

    # Wait for nodes to be initialized
    while True:

• Read up
  Read up

Function __init__ has 33 lines of code (exceeds 25 allowed). Consider refactoring.

Open

    def __init__(
        self,
        name: str,
        rate: float,
        graph: Graph,

Function _convert_type has a Cognitive Complexity of 11 (exceeds 5 allowed). Consider refactoring.

Open

def _convert_type(param):
    if type(param) in supported_types or param is None:
        return param
    else:
        if isinstance(param, SpecView):

• Read up
  Read up

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

Open

    def callback(self, msg):
        if not isinstance(msg, tuple(_ROS_TO_NUMPY)):
            Ros1.logerr(f"[subscriber][{self._name}]: type(recv)={type(msg)}")
            time.sleep(10000000)

• Read up
  Read up

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

Open

    def initialize(self, spec_cls):
        import eagerx.core.register as register

        agnostic = register.LOOKUP_TYPES[spec_cls.make]

• Read up
  Read up

Function split has a Cognitive Complexity of 11 (exceeds 5 allowed). Consider refactoring.

Open

def split(a: typing.Any):
    if isinstance(a, dict):
        for key in list(a):
            value = a.pop(key)
            value = split(value)

• Read up
  Read up

Function switch_to_reset has a Cognitive Complexity of 10 (exceeds 5 allowed). Consider refactoring.

Open

def switch_to_reset():
    def _switch_to_reset(source):
        def subscribe(observer, scheduler=None):
            reset_mode = [False]

• Read up
  Read up

Function render has a Cognitive Complexity of 10 (exceeds 5 allowed). Consider refactoring.

Open

    def render(self) -> Optional[np.ndarray]:
        """A method to start rendering (i.e. open the render window).

        A bool message to topic address ":attr:`~eagerx.core.env.BaseEnv.name` */env/render/toggle*",
        which toggles the rendering on/off.

• Read up
  Read up

Function _prepare_io_topics has a Cognitive Complexity of 10 (exceeds 5 allowed). Consider refactoring.

Open

    def _prepare_io_topics(self, name: str, env_node: EnvNode):
        params = get_param_with_blocking(name, self.backend)

        # Get info from engine on reactive properties
        sync = get_param_with_blocking(self.ns + "/sync", self.backend)

• Read up
  Read up

Function supported_types has a Cognitive Complexity of 10 (exceeds 5 allowed). Consider refactoring.

Open

def supported_types(*types: Tuple, is_classmethod=True):
    # Check if we support NoneType
    none_support = False
    for a in types:
        if a is None:

• Read up
  Read up

Function init_supervisor has 27 lines of code (exceeds 25 allowed). Consider refactoring.

Open

def init_supervisor(ns, node, outputs=tuple(), state_outputs=tuple()):
    # Initialize schedulers
    tp_scheduler = ThreadPoolScheduler(max_workers=5)
    reset_disp = CompositeDisposable()

Consider simplifying this complex logical expression.

Open

        if self._space is None:
            return (
                isinstance(other, Space)
                and other._space is None
                and (self.shape == other.shape)

Function _register_types has a Cognitive Complexity of 9 (exceeds 5 allowed). Consider refactoring.

Open

def _register_types(TYPE_REGISTER, component, cnames, func, space_only=True):
    name_split = func.__qualname__.split(".")
    cls_name = name_split[0]
    fn_name = name_split[1]
    entity_id = func.__module__ + "/" + cls_name

• Read up
  Read up

Function filter_dict has a Cognitive Complexity of 9 (exceeds 5 allowed). Consider refactoring.

Open

def filter_dict():
    def _filter_dict(source):
        def subscribe(observer, scheduler=None):
            def on_next(value):
                d = dict()

• Read up
  Read up

Function _resolve_args has a Cognitive Complexity of 9 (exceeds 5 allowed). Consider refactoring.

Open

def _resolve_args(arg_str, context, resolve_anon, commands):
    ros_valid = ["find", "env", "optenv", "dirname", "anon", "arg"]
    valid = ros_valid + ["ns", "config"]
    resolved = arg_str
    if isinstance(arg_str, (str, list)):

• 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

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

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]: