code: def _fetch_memray_profile(
      dask_worker: Worker,
      filename: str,
      report_args: Sequence[str] | Literal[False],
  ) -> bytes:
      """Generate and fetch the memray report"""
      if not hasattr(dask_worker, "_memray"):
          return b""
      path = pathlib.Path(dask_worker.local_directory) / (
          filename + str(dask_worker.id)
      )
      dask_worker._memray.close()
      del dask_worker._memray
  
      if not report_args:
          with open(path, "rb") as fd:
              return fd.read()
  
      report_filename = path.with_suffix(".html")
      if not report_args[0] == "memray":
          report_args = ["memray"] + list(report_args)
      assert (
          "-f" not in report_args
      ), "Cannot provide filename for report generation"
      assert (
          "-o" not in report_args
      ), "Cannot provide output filename for report generation"
      report_args = list(report_args) + [
          "-f",
          str(path),
          "-o",
          str(report_filename),
      ]
      subprocess.run(report_args)
      with open(report_filename, "rb") as fd:
          return fd.read()
  
file_path: distributed/distributed/diagnostics/memray.py
imports: from __future__ import annotations
  import contextlib
  import logging
  import pathlib
  import subprocess
  import time
  import uuid
  from collections.abc import Iterator, Sequence
  from typing import Any, Literal
  from urllib.parse import quote
  from toolz.itertoolz import partition
  from distributed import get_client
  from distributed.worker import Worker
label_desc: Use image node_id 2 for calling a global function with example usage: _fetch_memray_profile(dask_worker, filename, report_args) and returns: bytes, fd, fd