code: def test_weight_stream():
      def _cascader(cached_func, const_dict, sch):
          weight = cached_func.inputs[1]
          scale_bias = cached_func.inputs[2]
          out = cached_func.outputs[0]
          conv_compute = OperatorCompute.from_output(out)
          co = conv_compute.split(sch, 3, 10)
          cache_weight = sch.cache_read(
              weight, "global", [conv_compute.op]
          )
          cache_scale_bias = sch.cache_read(
              scale_bias, "global", [conv_compute.op]
          )
          sch[cache_weight].compute_at(sch[out], co)
          sch[cache_scale_bias].compute_at(sch[out], co)
  
      def _get_func():
          ifm = relay.var("ifm", shape=(1, 16, 16, 32), dtype="int8")
          conv = make_ethosu_conv2d(
              ifm,
              32,
              16,
              (1, 1),
              (0, 0),
              (1, 1),
              (1, 1),
          )
          func = relay.Function(relay.analysis.free_vars(conv), conv)
          func = run_opt_pass(func, relay.transform.InferType())
          return func
  
      func = _get_func()
      mod, _ = _lower_to_tir(func, cascader=_cascader)
  
      script = mod.script()
      test_mod = tvm.script.from_source(script)
      reference_mod = WeightStream
      tvm.ir.assert_structural_equal(
          test_mod["main"], reference_mod["main"], True
      )
  
file_path: tvm/tests/python/contrib/test_ethosu/test_replace_copy.py
imports: import pytest
  import tvm
  from tvm import relay
  from tvm.relay.backend.contrib.ethosu.tir.compiler import _lower_to_tir
  from tvm.relay.backend.contrib.ethosu.tir.scheduler import OperatorCompute, copy_constants
  from tvm.relay.testing import run_opt_pass
  from tvm.script import tir
  from .infra import make_ethosu_conv2d
label_desc: Use image node_id 2 for calling a global function with example usage: test_weight_stream() and returns: func