class_docstr: A Pipenet object connects transformers, detectors and aggregators.
  
  Parameters
  ----------
  steps: dicts
      Components of the pipenet. Each key-value item represents a step (
      transformer, detector, or aggregator), where key is the unique name of
      the step and the value is a dict with the following key-value pairs:
  
          - input (str or list of str): Input to the model, which must be
            either 'original' (i.e. the input time series), or the name of
            a upstream component.
          - subset (str, list of str, or list of lists of str, optional): If
            a model does not use all series from an input component, use this
            field to specify which series should be included. If not given or
            "all", all series from the input component will be used.
          - model (object): A detector, transformer, or aggregator object.
  
  Attributes
  ----------
  steps_graph_: OrderedDict
      Order of steps to be executed. Keys are step names, values are 2-tuple
      (i, j) where i is the index of execution round and j is the the index
      within a round.
  
  final_step_: str
      Name of the final step to be executed. It is the single step in the
      last round of execution in attribute `steps_graph_`.
  
  Examples
  --------
  The following example show how to use a Pipenet to build a level shift
  detector with some basic transformers, detectors, and aggregator.
  
  >>> from adtk.detector import QuantileAD, ThresholdAD
  >>> from adtk.transformer import DoubleRollingAggregate
  >>> from adtk.aggregator import AndAggregator
  >>> from adtk.pipe import Pipenet
  >>> steps = {
          "diff_abs": {
              "input": "original",
              "model": DoubleRollingAggregate(
                  agg="median",
                  window=20,
                  center=True,
                  diff="l1",
              ),
          },
          "quantile_ad": {
              "input": "diff_abs",
              "model": QuantileAD(high=0.99, low=0),
          },
          "diff": {
              "input": "original",
              "model": DoubleRollingAggregate(
                  agg="median",
                  window=20,
                  center=True,
                  diff="diff",
              ),
          },
          "sign_check": {
              "input": "diff",
              "model": ThresholdAD(high=0.0, low=-float("inf")),
          },
          "and": {
              "model": AndAggregator(),
              "input": ["quantile_ad", "sign_check"],
          },
      }
  >>> myPipenet = Pipenet(steps)
docstr: Private method for detect, transform, fit_detect, fit_transform
  
  Parameters
  ----------
  fit: bool
      Whether this call is for fit_detect/fit_transform or
      detect/transform.
  
  detect: bool
      Whether this call is for detect/fit_detect or
      transform/fit_transform.
  
  Others:
      Same as higher-level calls
  
  Returns
  -------
      Same as higher-level calls
docstr_tok: ["Private","method","for","detect",",","transform",",","fit_detect",",","fit_transform","Parameters","--","--","--","--","--","fit",":","bool","Whether","this","call","is","for","fit_detect\/fit_transform","or","detect\/transform",".","detect",":","bool","Whether","this","call","is","for","detect\/fit_detect","or","transform\/fit_transform",".","Others",":","Same","as","higher-level","calls","Returns","--","--","--","-","Same","as","higher-level","calls"]
code: def _predict(
      self,
      ts: Union[pd.Series, pd.DataFrame],
      fit: bool,
      detect: bool,
      skip_fit: Optional[List[str]],
      return_intermediate: bool,
      return_list: bool = False,
  ) -> Union[
      Union[
          pd.Series,
          pd.DataFrame,
          List[Union[Tuple[pd.Timestamp, pd.Timestamp], pd.Timestamp]],
          Dict[
              str,
              List[
                  Union[Tuple[pd.Timestamp, pd.Timestamp], pd.Timestamp]
              ],
          ],
      ],
      Dict[
          str,
          Union[
              pd.Series,
              pd.DataFrame,
              List[
                  Union[Tuple[pd.Timestamp, pd.Timestamp], pd.Timestamp]
              ],
              Dict[
                  str,
                  List[
                      Union[
                          Tuple[pd.Timestamp, pd.Timestamp],
                          pd.Timestamp,
                      ]
                  ],
              ],
          ],
      ],
  ]:
      """
      Private method for detect, transform, fit_detect, fit_transform
  
      Parameters
      ----------
      fit: bool
          Whether this call is for fit_detect/fit_transform or
          detect/transform.
  
      detect: bool
          Whether this call is for detect/fit_detect or
          transform/fit_transform.
  
      Others:
          Same as higher-level calls
  
      Returns
      -------
          Same as higher-level calls
  
      """
      self._validate()
  
      if skip_fit is None:
          skip_fit = []
      if not isinstance(skip_fit, list):
          raise TypeError("Parameter `skip_fit` must be a list.")
      if not set(skip_fit) <= set(self.steps.keys()):
          raise ValueError(
              "Name(s) in `skip_fit` is not valid model name."
          )
  
      last_step_name = list(self.steps_graph_.keys())[-1]
  
      if detect:
          if isinstance(
              self.steps[last_step_name]["model"], _Transformer
          ):
              raise RuntimeError(
                  "This seems a transformation pipenet, "
                  "because model at the final step '{}' is a transformer. "
                  "Please use method `{}transform` instead.".format(
                      last_step_name, "fit_" if fit else ""
                  )
              )
      else:
          if isinstance(
              self.steps[last_step_name]["model"],
              (_Detector, _Aggregator),
          ):
              raise RuntimeError(
                  "This seems a detection pipenet, "
                  "because model at the final step '{}' is "
                  "either a detector or an aggregator. "
                  "Please use method `{}detect` instead.".format(
                      last_step_name, "fit_" if fit else ""
                  )
              )
  
      results = {
          "original": ts.copy()
      }  # type: Dict[str,Union[pd.Series,pd.DataFrame,List[Union[Tuple[pd.Timestamp, pd.Timestamp], pd.Timestamp]], Dict[str, List[Union[Tuple[pd.Timestamp, pd.Timestamp], pd.Timestamp]]]]]
      for step_name in list(self.steps_graph_.keys())[1:]:
          step = self.steps[step_name]
          input = self._get_input(step, results)
          results.update(
              {
                  step_name: (
                      (
                          step["model"].fit_predict(
                              input, return_list=return_list
                          )
                          if isinstance(
                              step["model"],
                              (
                                  _TrainableUnivariateDetector,
                                  _TrainableMultivariateDetector,
                              ),
                          )
                          else step["model"].predict(
                              input, return_list=return_list
                          )
                      )
                      if isinstance(step["model"], _Detector)
                      else (
                          step["model"].fit_predict(input)
                          if isinstance(
                              step["model"],
                              (
                                  _TrainableUnivariateTransformer,
                                  _TrainableMultivariateTransformer,
                              ),
                          )
                          else step["model"].predict(input)
                      )
                  )
                  if fit and (step_name not in skip_fit)
                  else (
                      step["model"].predict(
                          input, return_list=return_list
                      )
                      if isinstance(step["model"], _Detector)
                      else step["model"].predict(input)
                  )
              }
          )
  
      if return_intermediate:
          return results
      else:
          return results[last_step_name]
  
imports: from collections import OrderedDict
  from typing import Any, Callable, Dict, List, Optional, Tuple, Union
  import matplotlib.pyplot
  import pandas
  from matplotlib.collections import PatchCollection
  from matplotlib.lines import Line2D
  from matplotlib.patches import Circle
  from tabulate import tabulate
  from .._aggregator_base import _Aggregator
  from .._base import _Model, _TrainableModel
  from .._detector_base import _NonTrainableUnivariateDetector, _TrainableMultivariateDetector, _TrainableUnivariateDetector
  from .._transformer_base import _NonTrainableMultivariateTransformer, _NonTrainableUnivariateTransformer, _TrainableMultivariateTransformer, _TrainableUnivariateTransformer
  from ..metrics import f1_score, iou, precision, recall
label_desc: Use image node_id 5 for calling the Pipenet obj's underlying member method code with example usage: obj._predict(ts, fit, detect, skip_fit, return_intermediate, return_list) and returns: results, results