code: def run_once(self):
      # 1. Wait for the caller queue data.
      called: Dict[str, Any] = self.caller_queue.get()
      logger.debug(f"Work {self.work_name} {called}")
  
      # 2. Extract the info from the caller queue data and process the input arguments. Arguments can contain
      # Lightning Path objects; if they don't have a consumer, the current Work will become one.
      call_hash = called["call_hash"]
      args, kwargs = self._process_call_args(
          called["args"], called["kwargs"]
      )
  
      # 3. Register the signal handler for spot instances.
      # `SIGUSR1` signal isn't supported on windows.
      # TODO (tchaton) Add support for windows
      if sys.platform != "win32":
          signal.signal(
              signal.SIGTERM,
              partial(
                  self._sigterm_signal_handler, call_hash=call_hash
              ),
          )
  
      # 4. Set the received state to the work.
      self.work.set_state(called["state"])
  
      # 5. Transfer all paths in the state automatically if they have an origin and exist
      self._transfer_path_attributes()
  
      # 6. Create the state observer thread.
      if self.run_executor_cls.enable_start_observer:
          self.state_observer = WorkStateObserver(
              self.work,
              delta_queue=self.delta_queue,
              flow_to_work_delta_queue=self.flow_to_work_delta_queue,
              error_queue=self.error_queue,
          )
  
      # 7. Deepcopy the work state and send the first `RUNNING` status delta to the flow.
      reference_state = deepcopy(self.work.state)
  
      # Set the internal IP address.
      # Set this here after the state observer is initialized, since it needs to record it as a change and send
      # it back to the flow
      default_internal_ip = (
          "127.0.0.1"
          if constants.LIGHTNING_CLOUDSPACE_HOST is None
          else "0.0.0.0"
      )  # noqa: S104
      self.work._internal_ip = os.environ.get(
          "LIGHTNING_NODE_PRIVATE_IP", default_internal_ip
      )
      self.work._public_ip = os.environ.get("LIGHTNING_NODE_IP", "")
  
      # 8. Patch the setattr method of the work. This needs to be done after step 4, so we don't
      # send delta while calling `set_state`.
      self._proxy_setattr()
  
      if self._is_starting(called, reference_state, call_hash):
          return
  
      # 9. Inform the flow the work is running and add the delta to the deepcopy.
      self.work._calls[CacheCallsKeys.LATEST_CALL_HASH] = call_hash
      self.work._calls[call_hash]["statuses"].append(
          make_status(WorkStageStatus.RUNNING)
      )
      delta = Delta(DeepDiff(reference_state, self.work.state))
      self.delta_queue.put(
          ComponentDelta(id=self.work_name, delta=delta)
      )
  
      # 10. Unwrap the run method if wrapped.
      work_run = self.work.run
      if hasattr(work_run, "__wrapped__"):
          work_run = work_run.__wrapped__
  
      # 11. Start the state observer thread. It will look for state changes and send them back to the Flow
      # The observer has to be initialized here, after the set_state call above so that the thread can start with
      # the proper initial state of the work
      if self.run_executor_cls.enable_start_observer:
          self.state_observer.start()
  
      # 12. Run the `work_run` method.
      # If an exception is raised, send a `FAILED` status delta to the flow and call the `on_exception` hook.
      try:
          ret = self.run_executor_cls(
              self.work, work_run, self.delta_queue
          )(*args, **kwargs)
      except LightningSigtermStateException as ex:
          raise ex
      except BaseException as ex:
          # 10.2 Send failed delta to the flow.
          reference_state = deepcopy(self.work.state)
          exp, val, tb = sys.exc_info()
          listing = traceback.format_exception(exp, val, tb)
          user_exception = False
          used_runpy = False
          trace = []
          for p in listing:
              if "runpy.py" in p:
                  trace = []
                  used_runpy = True
              if user_exception:
                  trace.append(p)
              if "ret = self.run_executor_cls(" in p:
                  user_exception = True
  
          if used_runpy:
              trace = trace[1:]
  
          self.work._calls[call_hash]["statuses"].append(
              make_status(
                  WorkStageStatus.FAILED,
                  message=str("\n".join(trace)),
                  reason=WorkFailureReasons.USER_EXCEPTION,
              )
          )
          self.delta_queue.put(
              ComponentDelta(
                  id=self.work_name,
                  delta=Delta(
                      DeepDiff(
                          reference_state,
                          self.work.state,
                          verbose_level=2,
                      )
                  ),
              )
          )
          self.work.on_exception(ex)
          print("########## CAPTURED EXCEPTION ###########")
          print(traceback.print_exc())
          print("########## CAPTURED EXCEPTION ###########")
          return
  
      # 13. Destroy the state observer.
      if (
          self.run_executor_cls.enable_start_observer
          and self.state_observer.started
      ):
          self.state_observer.join(0)
      self.state_observer = None
  
      # 14. Copy all artifacts to the shared storage so other Works can access them while this Work gets scaled down
      persist_artifacts(work=self.work)
  
      # 15. An asynchronous work shouldn't return a return value.
      if ret is not None:
          raise RuntimeError(
              f"Your work {self.work} shouldn't have a return value. Found {ret}."
              "HINT: Use the Payload API instead."
          )
  
      # 17. DeepCopy the state and send the latest delta to the flow.
      # use the latest state as we have already sent delta
      # during its execution.
      # inform the task has completed
      reference_state = deepcopy(self.work.state)
      self.work._calls[call_hash]["statuses"].append(
          make_status(WorkStageStatus.SUCCEEDED)
      )
      self.work._calls[call_hash]["ret"] = ret
      self.delta_queue.put(
          ComponentDelta(
              id=self.work_name,
              delta=Delta(
                  DeepDiff(
                      reference_state, self.work.state, verbose_level=2
                  )
              ),
          )
      )
  
      # 18. Update the work for the next delta if any.
      self._proxy_setattr(cleanup=True)
  
imports: import os
  import pathlib
  import queue
  import signal
  import sys
  import threading
  import time
  import traceback
  import warnings
  from contextlib import contextmanager
  from copy import deepcopy
  from dataclasses import dataclass, field
  from functools import partial
  from threading import Event, Thread
  from typing import TYPE_CHECKING, Any, Callable, Dict, Generator, Optional, Set, Tuple, Type, Union
  from deepdiff import DeepDiff, Delta
  from lightning_utilities.core.apply_func import apply_to_collection
  from lightning.app.core import constants
  from lightning.app.core.queues import MultiProcessQueue
  from lightning.app.storage.copier import _Copier, _copy_files
  from lightning.app.storage.drive import Drive, _maybe_create_drive
  from lightning.app.storage.path import Path, _path_to_work_artifact
  from lightning.app.storage.payload import Payload
  from lightning.app.utilities.app_helpers import affiliation
  from lightning.app.utilities.component import _set_work_context
  from lightning.app.utilities.enum import CacheCallsKeys, WorkFailureReasons, WorkStageStatus, WorkStopReasons, make_status
  from lightning.app.utilities.exceptions import CacheMissException, LightningSigtermStateException
  from lightning.app.utilities.app_helpers import Logger
label_desc: Use image node_id 4 for calling the WorkRunner obj's underlying member method code with example usage: obj.run_once() without return types