code: def test_lsdd_online(lsdd_online_params, seed):
      (
          n_features,
          ert,
          window_size,
          preprocess,
          n_bootstraps,
      ) = lsdd_online_params
  
      with fixed_seed(seed):
          x_ref = (
              np.random.randn(n * n_features)
              .reshape(n, n_features)
              .astype(np.float32)
          )
      preprocess_fn, preprocess_kwargs = preprocess
      to_list = False
      if (
          hasattr(preprocess_fn, "__name__")
          and preprocess_fn.__name__ == "preprocess_list"
      ):
          to_list = True
          x_ref = [_[None, :] for _ in x_ref]
      elif (
          isinstance(preprocess_fn, Callable)
          and "layer" in list(preprocess_kwargs.keys())
          and preprocess_kwargs["model"].__name__ == "HiddenOutput"
      ):
          model = MyModel(n_features)
          layer = preprocess_kwargs["layer"]
          preprocess_fn = partial(
              preprocess_fn,
              model=HiddenOutput(model=model, layer=layer),
          )
      else:
          preprocess_fn = None
  
      with fixed_seed(seed):
          cd = LSDDDriftOnlineTorch(
              x_ref=x_ref,
              ert=ert,
              window_size=window_size,
              preprocess_fn=preprocess_fn,
              n_bootstraps=n_bootstraps,
          )
          x_h0 = (
              np.random.randn(n * n_features)
              .reshape(n, n_features)
              .astype(np.float32)
          )
          x_h1 = (
              np.random.randn(n * n_features)
              .reshape(n, n_features)
              .astype(np.float32)
              + 1
          )
  
      detection_times_h0 = []
      test_stats_h0 = []
      for x_t in x_h0:
          if to_list:
              x_t = [x_t]
          pred_t = cd.predict(x_t, return_test_stat=True)
          test_stats_h0.append(pred_t["data"]["test_stat"])
          if pred_t["data"]["is_drift"]:
              detection_times_h0.append(pred_t["data"]["time"])
              cd.reset_state()
      average_delay_h0 = np.array(detection_times_h0).mean()
      test_stats_h0 = [ts for ts in test_stats_h0 if ts is not None]
      assert ert / 3 < average_delay_h0 < 3 * ert
  
      cd.reset_state()
  
      detection_times_h1 = []
      test_stats_h1 = []
      for x_t in x_h1:
          if to_list:
              x_t = [x_t]
          pred_t = cd.predict(x_t, return_test_stat=True)
          test_stats_h1.append(pred_t["data"]["test_stat"])
          if pred_t["data"]["is_drift"]:
              detection_times_h1.append(pred_t["data"]["time"])
              cd.reset_state()
      average_delay_h1 = np.array(detection_times_h1).mean()
      test_stats_h1 = [ts for ts in test_stats_h1 if ts is not None]
      assert np.abs(average_delay_h1) < ert / 2
  
      assert np.mean(test_stats_h1) > np.mean(test_stats_h0)
  
file_path: alibi-detect/alibi_detect/cd/pytorch/tests/test_lsdd_online_pt.py
imports: from functools import partial
  from itertools import product
  import numpy
  import pytest
  import torch
  import torch.nn
  from typing import Callable, List
  from alibi_detect.cd.pytorch.lsdd_online import LSDDDriftOnlineTorch
  from alibi_detect.cd.pytorch.preprocess import HiddenOutput, preprocess_drift
  from alibi_detect.utils._random import fixed_seed
label_desc: Use image node_id 3 for calling a global function with example usage: test_lsdd_online(lsdd_online_params, seed) without return types