code: def make_xm_docker_resources(
      program, requirements: Optional[str] = None
  ):
      """Returns Docker XManager resources for each program's node.
  
      For each node of the Launchpad's program appropriate hardware requirements are
      specified (CPU, memory...), while the list of PyPi packages specified in
      the requirements file will be installed inside the Docker images.
  
      Args:
        program: program for which to construct Docker XManager resources.
        requirements: file containing additional requirements to use.
          If not specified, default Acme dependencies are used instead.
      """
      if (
          FLAGS.lp_launch_type != "vertex_ai"
          and FLAGS.lp_launch_type != "local_docker"
      ):
          # Avoid importing 'xmanager' for local runs.
          return None
  
      # Avoid importing Launchpad until it is actually used.
      import launchpad as lp  # pylint: disable=g-import-not-at-top
  
      # Reference lp.DockerConfig to force lazy import of xmanager by Launchpad and
      # then import it. It is done this way to avoid heavy imports by default.
      lp.DockerConfig  # pylint: disable=pointless-statement
      from xmanager import xm  # pylint: disable=g-import-not-at-top
  
      # Get number of each type of node.
      num_nodes = {k: len(v) for k, v in program.groups.items()}
  
      xm_resources = {}
  
      acme_location = os.path.dirname(os.path.dirname(__file__))
      if not requirements:
          # Acme requirements are located in the Acme directory (when installed
          # with pip), or need to be extracted from setup.py when using Acme codebase
          # from GitHub without PyPi installation.
          requirements = os.path.join(acme_location, "requirements.txt")
          if not os.path.isfile(requirements):
              # Try to generate requirements.txt from setup.py
              setup = os.path.join(
                  os.path.dirname(acme_location), "setup.py"
              )
              if os.path.isfile(setup):
                  # Generate requirements.txt file using setup.py.
                  import importlib.util  # pylint: disable=g-import-not-at-top
  
                  spec = importlib.util.spec_from_file_location(
                      "setup", setup
                  )
                  setup = importlib.util.module_from_spec(spec)
                  try:
                      spec.loader.exec_module(
                          setup
                      )  # pytype: disable=attribute-error
                  except SystemExit:
                      pass
                  atexit.register(os.remove, requirements)
                  setup.generate_requirements_file(requirements)
  
      # Extend PYTHONPATH with paths used by the launcher.
      python_path = []
      for path in sys.path:
          if path.startswith(acme_location) and acme_location != path:
              python_path.append(path[len(acme_location) :])
  
      if "replay" in num_nodes:
          replay_cpu = 6 + num_nodes.get("actor", 0) * 0.01
          replay_cpu = min(40, replay_cpu)
  
          xm_resources["replay"] = lp.DockerConfig(
              acme_location,
              requirements,
              hw_requirements=xm.JobRequirements(
                  cpu=replay_cpu, ram=10 * xm.GiB
              ),
              python_path=python_path,
          )
  
      if "evaluator" in num_nodes:
          xm_resources["evaluator"] = lp.DockerConfig(
              acme_location,
              requirements,
              hw_requirements=xm.JobRequirements(cpu=2, ram=4 * xm.GiB),
              python_path=python_path,
          )
  
      if "actor" in num_nodes:
          xm_resources["actor"] = lp.DockerConfig(
              acme_location,
              requirements,
              hw_requirements=xm.JobRequirements(cpu=2, ram=4 * xm.GiB),
              python_path=python_path,
          )
  
      if "learner" in num_nodes:
          learner_cpu = 6 + num_nodes.get("actor", 0) * 0.01
          learner_cpu = min(40, learner_cpu)
          xm_resources["learner"] = lp.DockerConfig(
              acme_location,
              requirements,
              hw_requirements=xm.JobRequirements(
                  cpu=learner_cpu, ram=6 * xm.GiB, P100=1
              ),
              python_path=python_path,
          )
  
      if "environment_loop" in num_nodes:
          xm_resources["environment_loop"] = lp.DockerConfig(
              acme_location,
              requirements,
              hw_requirements=xm.JobRequirements(
                  cpu=6, ram=6 * xm.GiB, P100=1
              ),
              python_path=python_path,
          )
  
      if "counter" in num_nodes:
          xm_resources["counter"] = lp.DockerConfig(
              acme_location,
              requirements,
              hw_requirements=xm.JobRequirements(cpu=3, ram=4 * xm.GiB),
              python_path=python_path,
          )
  
      if "cacher" in num_nodes:
          xm_resources["cacher"] = lp.DockerConfig(
              acme_location,
              requirements,
              hw_requirements=xm.JobRequirements(cpu=3, ram=6 * xm.GiB),
              python_path=python_path,
          )
  
      return xm_resources
  
file_path: acme/acme/utils/lp_utils.py
imports: import atexit
  import functools
  import inspect
  import os
  import sys
  import time
  from typing import Any, Callable, Optional
  from absl import flags
  from absl import logging
  from acme.utils import counting
  from acme.utils import signals
label_desc: Use image node_id 3 for calling a global function with example usage: make_xm_docker_resources(program, requirements) and returns: xm_resources, None