class_docstr: This thread runs alongside LightningWork and periodically checks for state changes. If the state changed from
  one interval to the next, it will compute the delta and add it to the queue which is connected to the Flow. This
  enables state changes to be captured that are not triggered through a setattr call.
  
  Args:
      work: The LightningWork for which the state should be monitored
      delta_queue: The queue to send deltas to when state changes occur
      interval: The interval at which to check for state changes.
  
  Example:
  
      class Work(LightningWork):
          ...
  
          def run(self):
              # This update gets sent to the Flow once the thread compares the new state with the previous one
              self.list.append(1)
class_docstr_tok: ["This","thread","runs","alongside","LightningWork","and","periodically","checks","for","state","changes",".","If","the","state","changed","from","one","interval","to","the","next",",","it","will","compute","the","delta","and","add","it","to","the","queue","which","is","connected","to","the","Flow",".","This","enables","state","changes","to","be","captured","that","are","not","triggered","through","a","setattr","call",".","Args",":","work",":","The","LightningWork","for","which","the","state","should","be","monitored","delta_queue",":","The","queue","to","send","deltas","to","when","state","changes","occur","interval",":","The","interval","at","which","to","check","for","state","changes",".","Example",":","class","Work","(","LightningWork",")",":","...","def","run","(","self",")",":","#","This","update","gets","sent","to","the","Flow","once","the","thread","compares","the","new","state","with","the","previous","one","self.list.append","(","1",")"]
code: def run_once(self) -> None:
      with _state_observer_lock:
          # Add all deltas the LightningWorkSetAttrProxy has processed and sent to the Flow already while
          # the WorkStateObserver was sleeping
          for delta in self._delta_memory:
              self._last_state += delta
          self._delta_memory.clear()
  
          # The remaining delta is the result of state updates triggered outside the setattr, e.g, by a list append
          delta = Delta(
              DeepDiff(
                  self._last_state, self._work.state, verbose_level=2
              )
          )
          if not delta.to_dict():
              return
          self._last_state = deepcopy(self._work.state)
          self._delta_queue.put(
              ComponentDelta(id=self._work.name, delta=delta)
          )
  
      if self._flow_to_work_delta_queue:
          while True:
              deep_diff = self.get_state_changed_from_queue(
                  self._flow_to_work_delta_queue
              )
              if not isinstance(deep_diff, dict):
                  break
              try:
                  with _state_observer_lock:
                      self._work.apply_flow_delta(
                          Delta(deep_diff, raise_errors=True)
                      )
              except Exception as ex:
                  print(traceback.print_exc())
                  self._error_queue.put(ex)
                  raise ex
  
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 WorkStateObserver obj's underlying member method code with example usage: obj.run_once() without return types