def gen_ode_system_convergence()

in src/envs/ode.py [0:0]

• 93 lines of code
• 37 McCabe index (conditional complexity)

    def gen_ode_system_convergence(self, return_system=False):
        """
        Generate systems of functions, and the corresponding convergence speed in zero.
        Start by generating a random system S, use SymPy to compute formal jacobian
        and evaluate it in zero, find largest eigenvalue
        Encode this as a prefix sensence
        """
        degree = self.rng.randint(self.min_degree, self.max_degree + 1)
        nb_ops = self.rng.randint(
            self.min_expr_len_factor_cspeed * degree + 3,
            self.max_expr_len_factor_cspeed * degree + 3,
            size=(degree,),
        )

        while True:
            system = []
            i = 0
            ngen = 0
            while i < degree:
                # generate expression
                expr = self.generate_tree(nb_ops[i], degree)
                ngen += 1
                # sympy zone
                try:
                    expr_sp = sp.S(expr, locals=self.local_dict)
                    # skip constant or invalid expressions
                    if len(expr_sp.free_symbols) == 0 or has_inf_nan(expr_sp):
                        continue
                    # evaluate gradient in point
                    values = self.compute_gradient(expr_sp, self.eval_point, degree)
                    if np.isnan(values).any() or np.isinf(values).any():
                        continue
                    if self.skip_zero_gradient and not values.any():
                        continue
                except TimeoutError:
                    continue
                except (ValueError, TypeError):
                    continue
                except Exception as e:
                    logger.error(
                        "An unknown exception of type {0} occurred in line {1} "
                        'for expression "{2}". Arguments:{3!r}.'.format(
                            type(e).__name__,
                            sys.exc_info()[-1].tb_lineno,
                            expr_sp,
                            e.args,
                        )
                    )
                    continue

                system.append(expr)
                if i == 0:
                    jacobian = values
                else:
                    jacobian = np.vstack((jacobian, values))
                i += 1
            if self.skip_zero_gradient:
                skip = False
                for i in range(degree):
                    if not jacobian[:, [i]].any():
                        skip = True
                        break
                if skip:
                    continue

            cspeed = -max(np.linalg.eigvals(jacobian).real)

            if self.prob_positive == 0 and cspeed > 0:
                continue
            if self.prob_positive == 1 and cspeed <= 0:
                continue
            if (
                self.prob_positive > 0
                and self.prob_positive < 1
                and self.np_total[degree] > 10
            ):
                proportion = self.np_positive[degree] / self.np_total[degree]
                if cspeed > 0 and proportion > self.prob_positive:
                    continue
                if cspeed <= 0 and proportion < self.prob_positive:
                    continue

            self.np_total[degree] += 1
            if cspeed > 0:
                self.np_positive[degree] += 1
            break

        # # debug
        # logger.info(str(cspeed))
        # logger.info(str(cspeed) + "\t" + " ||||| ".join(str(s) for s in system[:3]))
        # print(degree, str(ngen) + " : " + str((ngen - degree) / ngen * 100.0))

        # encode input
        x = self.write_int(degree)
        for s in system:
            x.append(self.func_separator)
            x.extend(self.encode_expr(s))

        # encode output: eigenvalue, and optionally the Jacobian matrix
        eigenvalue = self.write_float(cspeed)
        if self.predict_jacobian:
            y = []
            for row in jacobian:
                for value in row:
                    y.extend(
                        self.write_complex(value, precision=self.jacobian_precision)
                    )
                    y.append(self.list_separator)
                y.append(self.line_separator)
            y.append(self.mtrx_separator)
            y.extend(eigenvalue)
        else:
            y = eigenvalue

        if return_system:
            return x, y, system
        else:
            return x, y