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

Avoid too many return statements within this function.

Open

            return self.logfatal

Avoid too many return statements within this function.

Open

            return logsilent

Avoid too many return statements within this function.

Open

        return float(value)

Avoid too many return statements within this function.

Open

            return False

Avoid too many return statements within this function.

Open

            return yaml.safe_load(value)

Avoid too many return statements within this function.

Open

        return int(value)

Avoid too many return statements within this function.

Open

        return value

Avoid too many return statements within this function.

Open

            return True

Function connect has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

    def connect(
        self,
        source: Optional[SpecView] = None,
        target: Optional[SpecView] = None,
        actuator: str = None,

• Read up
  Read up

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

Open

    def initialize(self, spec, simulator):
        # We will probably use self.simulator in callback & reset.
        assert (
            self.process == process.ENGINE
        ), "Simulation node requires a reference to the simulator, hence it must be launched in the Engine process"

• Read up
  Read up

Function delete_param has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

    def delete_param(self, param: str, level: int = 1) -> None:
        try:
            keys = [k for k in param.split("/") if len(k) > 0]
            val = self._pserver
            for key in keys[:-1]:

• Read up
  Read up

Function reset_graph has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

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

• Read up
  Read up

Function _observation_space has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

    def _observation_space(self) -> gym.spaces.Dict:
        """Infers the observation space from the space of every observation.

        This space defines the format of valid observations.

• Read up
  Read up

Function __getitem__ has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

    def __getitem__(self, spec_lookup):
        for _entity_cls, entity_id in self._dict.items():
            for entity_idd, entry in entity_id.items():
                if entry["spec"] == spec_lookup:
                    return entity_idd

• Read up
  Read up

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

Open

    def build(self, ns: str):
        params = super().build(ns)

        # Convert engine states
        for name, obj in params[self.config.name]["objects"].items():

• Read up
  Read up

Function _publish has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

    def _publish(self, msg: typing.Union[float, bool, int, str, np.ndarray, np.number], header: Header) -> None:
        # Convert python native types to numpy arrays.
        if isinstance(msg, float):
            msg = np.array(msg, dtype="float32")
        elif isinstance(msg, bool):

• Read up
  Read up

Function __eq__ has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

    def __eq__(self, other):
        if self._space is None:
            return (
                isinstance(other, Space)
                and other._space is None

• Read up
  Read up

Function _create_supervisor has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

    def _create_supervisor(environment: NodeSpec, engine: EngineSpec, nodes: List[NodeSpec]) -> BaseNodeSpec:
        entity_type = f"{SupervisorNode.__module__}/{SupervisorNode.__name__}"
        spec = Node.pre_make("N/a", entity_type)
        spec.add_output("step", space=Space(shape=(), dtype="int64"))

• Read up
  Read up

Function get_log_fn has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

    def get_log_fn(self, log_level: int) -> Callable:
        if log_level == DEBUG:
            return self.logdebug
        elif log_level == INFO:
            return self.loginfo

• Read up
  Read up