def _run_r_forecast()

in src/gluonts/model/r_forecast/_predictor.py [0:0]

• 51 lines of code
• 11 McCabe index (conditional complexity)

    def _run_r_forecast(self, d, params, save_info):
        buf = []

        def save_to_buf(x):
            buf.append(x)

        def dont_save(x):
            pass

        f = save_to_buf if save_info else dont_save

        # save output from the R console in buf
        self._rinterface.set_writeconsole_regular(f)
        self._rinterface.set_writeconsole_warnerror(f)

        make_ts = self._stats_pkg.ts
        r_params = self._robjects.vectors.ListVector(params)
        vec = self._robjects.FloatVector(d["target"])
        ts = make_ts(vec, frequency=self.period)
        forecast = self._r_method(ts, r_params)
        forecast_dict = dict(
            zip(forecast.names, map(self._unlist, list(forecast)))
        )

        if "quantiles" in forecast_dict or "upper_quantiles" in forecast_dict:

            def from_interval_to_level(interval: int, side: str):
                if side == "upper":
                    level = 50 + interval / 2
                elif side == "lower":
                    level = 50 - interval / 2
                else:
                    raise ValueError
                return level / 100

            # Post-processing quantiles on then Python side for the convenience of asserting and debugging.
            upper_quantiles = [
                str(from_interval_to_level(interval, side="upper"))
                for interval in params["intervals"]
            ]

            lower_quantiles = [
                str(from_interval_to_level(interval, side="lower"))
                for interval in params["intervals"]
            ]

            # Median forecasts would be available at two places: Lower 0 and Higher 0 (0-prediction interval)
            forecast_dict["quantiles"] = dict(
                zip(
                    lower_quantiles + upper_quantiles[1:],
                    forecast_dict["lower_quantiles"]
                    + forecast_dict["upper_quantiles"][1:],
                )
            )

            # `QuantileForecast` allows "mean" as the key; we store them as well since they can differ from median.
            forecast_dict["quantiles"].update(
                {"mean": forecast_dict.pop("mean")}
            )

        self._rinterface.set_writeconsole_regular(
            self._rinterface.consolePrint
        )
        self._rinterface.set_writeconsole_warnerror(
            self._rinterface.consolePrint
        )
        return forecast_dict, buf