def __call__()

in core/maxframe/dataframe/misc/map.py [0:0]

• 55 lines of code
• 16 McCabe index (conditional complexity)

    def __call__(self, series, dtype, skip_infer=False):
        if dtype is None and not skip_infer:
            inferred_dtype = None
            if callable(self.arg):
                # arg is a function, try to inspect the signature
                sig = inspect.signature(self.arg)
                return_type = sig.return_annotation
                if return_type is not inspect._empty:
                    inferred_dtype = np.dtype(return_type)
                else:
                    try:
                        with quiet_stdio():
                            # try to infer dtype by calling the function
                            inferred_dtype = (
                                build_series(series)
                                .map(self.arg, na_action=self.na_action)
                                .dtype
                            )
                    except:  # noqa: E722  # nosec
                        pass
            else:
                if isinstance(self.arg, MutableMapping):
                    inferred_dtype = pd.Series(self.arg).dtype
                else:
                    inferred_dtype = self.arg.dtype
            if inferred_dtype is not None and np.issubdtype(inferred_dtype, np.number):
                if np.issubdtype(inferred_dtype, np.inexact):
                    # for the inexact e.g. float
                    # we can make the decision,
                    # but for int, due to the nan which may occur,
                    # we cannot infer the dtype
                    dtype = inferred_dtype
            else:
                dtype = inferred_dtype

        if dtype is None:
            if not skip_infer:
                raise ValueError(
                    "cannot infer dtype, it needs to be specified manually for `map`"
                )
        else:
            dtype = np.int64 if dtype is int else dtype
            dtype = np.dtype(dtype)

        inputs = [series]
        if isinstance(self.arg, SERIES_TYPE):
            inputs.append(self.arg)

        if isinstance(series, SERIES_TYPE):
            return self.new_series(
                inputs,
                shape=series.shape,
                dtype=dtype,
                index_value=series.index_value,
                name=series.name,
            )
        else:
            return self.new_index(
                inputs,
                shape=series.shape,
                dtype=dtype,
                index_value=series.index_value,
                name=series.name,
            )