meerschaum.utils.daemon.Daemon

   1#! /usr/bin/env python3
   2# -*- coding: utf-8 -*-
   3# vim:fenc=utf-8
   4
   5"""
   6Manage running daemons via the Daemon class.
   7"""
   8
   9from __future__ import annotations
  10import os
  11import importlib
  12import pathlib
  13import json
  14import shutil
  15import signal
  16import time
  17import traceback
  18from functools import partial
  19from datetime import datetime, timezone
  20
  21import meerschaum as mrsm
  22from meerschaum.utils.typing import (
  23    Optional, Dict, Any, SuccessTuple, Callable, List, Union,
  24    is_success_tuple, Tuple,
  25)
  26from meerschaum.config import get_config
  27from meerschaum._internal.static import STATIC_CONFIG
  28from meerschaum.config._patch import apply_patch_to_config
  29from meerschaum.utils.warnings import warn, error
  30from meerschaum.utils.packages import attempt_import
  31from meerschaum.utils.venv import venv_exec
  32from meerschaum.utils.daemon._names import get_new_daemon_name
  33from meerschaum.utils.daemon.RotatingFile import RotatingFile
  34from meerschaum.utils.daemon.StdinFile import StdinFile
  35from meerschaum.utils.threading import RepeatTimer
  36from meerschaum.__main__ import _close_pools
  37
  38_daemons = []
  39_results = {}
  40
  41class Daemon:
  42    """
  43    Daemonize Python functions into background processes.
  44
  45    Examples
  46    --------
  47    >>> import meerschaum as mrsm
  48    >>> from meerschaum.utils.daemons import Daemon
  49    >>> daemon = Daemon(print, ('hi',))
  50    >>> success, msg = daemon.run()
  51    >>> print(daemon.log_text)
  52
  53    2024-07-29 18:03 | hi
  54    2024-07-29 18:03 |
  55    >>> daemon.run(allow_dirty_run=True)
  56    >>> print(daemon.log_text)
  57
  58    2024-07-29 18:03 | hi
  59    2024-07-29 18:03 |
  60    2024-07-29 18:05 | hi
  61    2024-07-29 18:05 |
  62    >>> mrsm.pprint(daemon.properties)
  63    {
  64        'label': 'print',
  65        'target': {'name': 'print', 'module': 'builtins', 'args': ['hi'], 'kw': {}},
  66        'result': None,
  67        'process': {'ended': '2024-07-29T18:03:33.752806'}
  68    }
  69
  70    """
  71
  72    def __new__(
  73        cls,
  74        *args,
  75        daemon_id: Optional[str] = None,
  76        **kw
  77    ):
  78        """
  79        If a daemon_id is provided and already exists, read from its pickle file.
  80        """
  81        instance = super(Daemon, cls).__new__(cls)
  82        if daemon_id is not None:
  83            instance.daemon_id = daemon_id
  84            if instance.pickle_path.exists():
  85                instance = instance.read_pickle()
  86        return instance
  87
  88    @classmethod
  89    def from_properties_file(cls, daemon_id: str) -> Daemon:
  90        """
  91        Return a Daemon from a properties dictionary.
  92        """
  93        properties_path = cls._get_properties_path_from_daemon_id(daemon_id)
  94        if not properties_path.exists():
  95            raise OSError(f"Properties file '{properties_path}' does not exist.")
  96
  97        try:
  98            with open(properties_path, 'r', encoding='utf-8') as f:
  99                properties = json.load(f)
 100        except Exception:
 101            properties = {}
 102
 103        if not properties:
 104            raise ValueError(f"No properties could be read for daemon '{daemon_id}'.")
 105
 106        daemon_id = properties_path.parent.name
 107        target_cf = properties.get('target', {})
 108        target_module_name = target_cf.get('module', None)
 109        target_function_name = target_cf.get('name', None)
 110        target_args = target_cf.get('args', None)
 111        target_kw = target_cf.get('kw', None)
 112        label = properties.get('label', None)
 113
 114        if None in [
 115            target_module_name,
 116            target_function_name,
 117            target_args,
 118            target_kw,
 119        ]:
 120            raise ValueError("Missing target function information.")
 121
 122        target_module = importlib.import_module(target_module_name)
 123        target_function = getattr(target_module, target_function_name)
 124
 125        return Daemon(
 126            daemon_id=daemon_id,
 127            target=target_function,
 128            target_args=target_args,
 129            target_kw=target_kw,
 130            properties=properties,
 131            label=label,
 132        )
 133
 134
 135    def __init__(
 136        self,
 137        target: Optional[Callable[[Any], Any]] = None,
 138        target_args: Union[List[Any], Tuple[Any], None] = None,
 139        target_kw: Optional[Dict[str, Any]] = None,
 140        env: Optional[Dict[str, str]] = None,
 141        daemon_id: Optional[str] = None,
 142        label: Optional[str] = None,
 143        properties: Optional[Dict[str, Any]] = None,
 144        pickle: bool = True,
 145    ):
 146        """
 147        Parameters
 148        ----------
 149        target: Optional[Callable[[Any], Any]], default None,
 150            The function to execute in a child process.
 151
 152        target_args: Union[List[Any], Tuple[Any], None], default None
 153            Positional arguments to pass to the target function.
 154
 155        target_kw: Optional[Dict[str, Any]], default None
 156            Keyword arguments to pass to the target function.
 157
 158        env: Optional[Dict[str, str]], default None
 159            If provided, set these environment variables in the daemon process.
 160
 161        daemon_id: Optional[str], default None
 162            Build a `Daemon` from an existing `daemon_id`.
 163            If `daemon_id` is provided, other arguments are ignored and are derived
 164            from the existing pickled `Daemon`.
 165
 166        label: Optional[str], default None
 167            Label string to help identifiy a daemon.
 168            If `None`, use the function name instead.
 169
 170        properties: Optional[Dict[str, Any]], default None
 171            Override reading from the properties JSON by providing an existing dictionary.
 172        """
 173        _pickle = self.__dict__.get('_pickle', False)
 174        if daemon_id is not None:
 175            self.daemon_id = daemon_id
 176            if not self.pickle_path.exists() and not target and ('target' not in self.__dict__):
 177
 178                if not self.properties_path.exists():
 179                    raise Exception(
 180                        f"Daemon '{self.daemon_id}' does not exist. "
 181                        + "Pass a target to create a new Daemon."
 182                    )
 183
 184                try:
 185                    new_daemon = self.from_properties_file(daemon_id)
 186                except Exception:
 187                    new_daemon = None
 188
 189                if new_daemon is not None:
 190                    new_daemon.write_pickle()
 191                    target = new_daemon.target
 192                    target_args = new_daemon.target_args
 193                    target_kw = new_daemon.target_kw
 194                    label = new_daemon.label
 195                    self._properties = new_daemon.properties
 196                else:
 197                    try:
 198                        self.properties_path.unlink()
 199                    except Exception:
 200                        pass
 201
 202                    raise Exception(
 203                        f"Could not recover daemon '{self.daemon_id}' "
 204                        + "from its properties file."
 205                    )
 206
 207        if 'target' not in self.__dict__:
 208            if target is None:
 209                error("Cannot create a Daemon without a target.")
 210            self.target = target
 211
 212        self.pickle = pickle
 213
 214        ### NOTE: We have to check self.__dict__ in case we un-pickling.
 215        if '_target_args' not in self.__dict__:
 216            self._target_args = target_args
 217        if '_target_kw' not in self.__dict__:
 218            self._target_kw = target_kw
 219
 220        if 'label' not in self.__dict__:
 221            if label is None:
 222                label = (
 223                    self.target.__name__ if '__name__' in self.target.__dir__()
 224                        else str(self.target)
 225                )
 226            self.label = label
 227        elif label is not None:
 228            self.label = label
 229
 230        if 'daemon_id' not in self.__dict__:
 231            self.daemon_id = get_new_daemon_name()
 232        if '_properties' not in self.__dict__:
 233            self._properties = properties
 234        elif properties:
 235            if self._properties is None:
 236                self._properties = {}
 237            self._properties.update(properties)
 238        if self._properties is None:
 239            self._properties = {}
 240
 241        self._properties.update({'label': self.label})
 242        if env:
 243            self._properties.update({'env': env})
 244
 245        ### Instantiate the process and if it doesn't exist, make sure the PID is removed.
 246        _ = self.process
 247
 248
 249    def _run_exit(
 250        self,
 251        keep_daemon_output: bool = True,
 252        allow_dirty_run: bool = False,
 253    ) -> Any:
 254        """Run the daemon's target function.
 255        NOTE: This WILL EXIT the parent process!
 256
 257        Parameters
 258        ----------
 259        keep_daemon_output: bool, default True
 260            If `False`, delete the daemon's output directory upon exiting.
 261
 262        allow_dirty_run, bool, default False:
 263            If `True`, run the daemon, even if the `daemon_id` directory exists.
 264            This option is dangerous because if the same `daemon_id` runs twice,
 265            the last to finish will overwrite the output of the first.
 266
 267        Returns
 268        -------
 269        Nothing — this will exit the parent process.
 270        """
 271        import platform
 272        import sys
 273        import os
 274        import traceback
 275        from meerschaum.utils.warnings import warn
 276        from meerschaum.config import get_config
 277        daemon = attempt_import('daemon')
 278        lines = get_config('jobs', 'terminal', 'lines')
 279        columns = get_config('jobs', 'terminal', 'columns')
 280
 281        if platform.system() == 'Windows':
 282            return False, "Windows is no longer supported."
 283
 284        self._setup(allow_dirty_run)
 285
 286        _daemons.append(self)
 287
 288        logs_cf = self.properties.get('logs', {})
 289        log_refresh_seconds = logs_cf.get('refresh_files_seconds', None)
 290        if log_refresh_seconds is None:
 291            log_refresh_seconds = get_config('jobs', 'logs', 'refresh_files_seconds')
 292        write_timestamps = logs_cf.get('write_timestamps', None)
 293        if write_timestamps is None:
 294            write_timestamps = get_config('jobs', 'logs', 'timestamps', 'enabled')
 295
 296        self._log_refresh_timer = RepeatTimer(
 297            log_refresh_seconds,
 298            partial(self.rotating_log.refresh_files, start_interception=write_timestamps),
 299        )
 300
 301        capture_stdin = logs_cf.get('stdin', True)
 302        cwd = self.properties.get('cwd', os.getcwd())
 303
 304        ### NOTE: The SIGINT handler has been removed so that child processes may handle
 305        ###       KeyboardInterrupts themselves.
 306        ###       The previous aggressive approach was redundant because of the SIGTERM handler.
 307        self._daemon_context = daemon.DaemonContext(
 308            pidfile=self.pid_lock,
 309            stdout=self.rotating_log,
 310            stderr=self.rotating_log,
 311            stdin=(self.stdin_file if capture_stdin else None),
 312            working_directory=cwd,
 313            detach_process=True,
 314            files_preserve=list(self.rotating_log.subfile_objects.values()),
 315            signal_map={
 316                signal.SIGTERM: self._handle_sigterm,
 317            },
 318        )
 319
 320        if capture_stdin and sys.stdin is None:
 321            raise OSError("Cannot daemonize without stdin.")
 322
 323        try:
 324            os.environ['LINES'], os.environ['COLUMNS'] = str(int(lines)), str(int(columns))
 325            with self._daemon_context:
 326                if capture_stdin:
 327                    sys.stdin = self.stdin_file
 328                _ = os.environ.pop(STATIC_CONFIG['environment']['systemd_stdin_path'], None)
 329                os.environ[STATIC_CONFIG['environment']['daemon_id']] = self.daemon_id
 330                os.environ['PYTHONUNBUFFERED'] = '1'
 331
 332                ### Allow the user to override environment variables.
 333                env = self.properties.get('env', {})
 334                if env and isinstance(env, dict):
 335                    os.environ.update({str(k): str(v) for k, v in env.items()})
 336
 337                self.rotating_log.refresh_files(start_interception=True)
 338                result = None
 339                try:
 340                    with open(self.pid_path, 'w+', encoding='utf-8') as f:
 341                        f.write(str(os.getpid()))
 342
 343                    ### NOTE: The timer fails to start for remote actions to localhost.
 344                    try:
 345                        if not self._log_refresh_timer.is_running():
 346                            self._log_refresh_timer.start()
 347                    except Exception:
 348                        pass
 349
 350                    self.properties['result'] = None
 351                    self._capture_process_timestamp('began')
 352                    result = self.target(*self.target_args, **self.target_kw)
 353                    self.properties['result'] = result
 354                except (BrokenPipeError, KeyboardInterrupt, SystemExit):
 355                    result = False, traceback.format_exc()
 356                except Exception as e:
 357                    warn(
 358                        f"Exception in daemon target function: {traceback.format_exc()}",
 359                    )
 360                    result = False, str(e)
 361                finally:
 362                    _results[self.daemon_id] = result
 363                    self.properties['result'] = result
 364
 365                    if keep_daemon_output:
 366                        self._capture_process_timestamp('ended')
 367                    else:
 368                        self.cleanup()
 369
 370                    self._log_refresh_timer.cancel()
 371                    try:
 372                        if self.pid is None and self.pid_path.exists():
 373                            self.pid_path.unlink()
 374                    except Exception:
 375                        pass
 376
 377                    if is_success_tuple(result):
 378                        try:
 379                            mrsm.pprint(result)
 380                        except BrokenPipeError:
 381                            pass
 382
 383        except Exception:
 384            daemon_error = traceback.format_exc()
 385            import meerschaum.config.paths as paths
 386            with open(paths.DAEMON_ERROR_LOG_PATH, 'a+', encoding='utf-8') as f:
 387                f.write(
 388                    f"Error in Daemon '{self}':\n\n"
 389                    f"{sys.stdin=}\n"
 390                    f"{self.stdin_file_path=}\n"
 391                    f"{self.stdin_file_path.exists()=}\n\n"
 392                    f"{daemon_error}\n\n"
 393                )
 394            warn(f"Encountered an error while running the daemon '{self}':\n{daemon_error}")
 395
 396    def _capture_process_timestamp(
 397        self,
 398        process_key: str,
 399        write_properties: bool = True,
 400    ) -> None:
 401        """
 402        Record the current timestamp to the parameters `process:<process_key>`.
 403
 404        Parameters
 405        ----------
 406        process_key: str
 407            Under which key to store the timestamp.
 408
 409        write_properties: bool, default True
 410            If `True` persist the properties to disk immediately after capturing the timestamp.
 411        """
 412        if 'process' not in self.properties:
 413            self.properties['process'] = {}
 414
 415        if process_key not in ('began', 'ended', 'paused', 'stopped'):
 416            raise ValueError(f"Invalid key '{process_key}'.")
 417
 418        self.properties['process'][process_key] = (
 419            datetime.now(timezone.utc).replace(tzinfo=None).isoformat()
 420        )
 421        if write_properties:
 422            self.write_properties()
 423
 424    def run(
 425        self,
 426        keep_daemon_output: bool = True,
 427        allow_dirty_run: bool = False,
 428        wait: bool = False,
 429        timeout: Union[int, float] = 4,
 430        debug: bool = False,
 431    ) -> SuccessTuple:
 432        """Run the daemon as a child process and continue executing the parent.
 433
 434        Parameters
 435        ----------
 436        keep_daemon_output: bool, default True
 437            If `False`, delete the daemon's output directory upon exiting.
 438
 439        allow_dirty_run: bool, default False
 440            If `True`, run the daemon, even if the `daemon_id` directory exists.
 441            This option is dangerous because if the same `daemon_id` runs concurrently,
 442            the last to finish will overwrite the output of the first.
 443
 444        wait: bool, default True
 445            If `True`, block until `Daemon.status` is running (or the timeout expires).
 446
 447        timeout: Union[int, float], default 4
 448            If `wait` is `True`, block for up to `timeout` seconds before returning a failure.
 449
 450        Returns
 451        -------
 452        A SuccessTuple indicating success.
 453
 454        """
 455        import platform
 456        if platform.system() == 'Windows':
 457            return False, "Cannot run background jobs on Windows."
 458
 459        ### The daemon might exist and be paused.
 460        if self.status == 'paused':
 461            return self.resume()
 462
 463        self._remove_stop_file()
 464        if self.status == 'running':
 465            return True, f"Daemon '{self}' is already running."
 466
 467        self.mkdir_if_not_exists(allow_dirty_run)
 468        _write_pickle_success_tuple = self.write_pickle()
 469        if not _write_pickle_success_tuple[0]:
 470            return _write_pickle_success_tuple
 471
 472        _launch_daemon_code = (
 473            "from meerschaum.utils.daemon import Daemon, _daemons; "
 474            f"daemon = Daemon(daemon_id='{self.daemon_id}'); "
 475            f"_daemons['{self.daemon_id}'] = daemon; "
 476            f"daemon._run_exit(keep_daemon_output={keep_daemon_output}, "
 477            "allow_dirty_run=True)"
 478        )
 479        env = dict(os.environ)
 480        _launch_success_bool = venv_exec(_launch_daemon_code, debug=debug, venv=None, env=env)
 481        msg = (
 482            "Success"
 483            if _launch_success_bool
 484            else f"Failed to start daemon '{self.daemon_id}'."
 485        )
 486        if not wait or not _launch_success_bool:
 487            return _launch_success_bool, msg
 488
 489        timeout = self.get_timeout_seconds(timeout)
 490        check_timeout_interval = self.get_check_timeout_interval_seconds()
 491
 492        if not timeout:
 493            success = self.status == 'running'
 494            msg = "Success" if success else f"Failed to run daemon '{self.daemon_id}'."
 495            if success:
 496                self._capture_process_timestamp('began')
 497            return success, msg
 498
 499        begin = time.perf_counter()
 500        while (time.perf_counter() - begin) < timeout:
 501            if self.status == 'running':
 502                self._capture_process_timestamp('began')
 503                return True, "Success"
 504            time.sleep(check_timeout_interval)
 505
 506        return False, (
 507            f"Failed to start daemon '{self.daemon_id}' within {timeout} second"
 508            + ('s' if timeout != 1 else '') + '.'
 509        )
 510
 511
 512    def kill(self, timeout: Union[int, float, None] = 8) -> SuccessTuple:
 513        """
 514        Forcibly terminate a running daemon.
 515        Sends a SIGTERM signal to the process.
 516
 517        Parameters
 518        ----------
 519        timeout: Optional[int], default 3
 520            How many seconds to wait for the process to terminate.
 521
 522        Returns
 523        -------
 524        A SuccessTuple indicating success.
 525        """
 526        if self.status != 'paused':
 527            success, msg = self._send_signal(signal.SIGTERM, timeout=timeout)
 528            if success:
 529                self._write_stop_file('kill')
 530                self.stdin_file.close()
 531                self._remove_blocking_stdin_file()
 532                return success, msg
 533
 534        if self.status == 'stopped':
 535            self._write_stop_file('kill')
 536            self.stdin_file.close()
 537            self._remove_blocking_stdin_file()
 538            return True, "Process has already stopped."
 539
 540        psutil = attempt_import('psutil')
 541        process = self.process
 542        try:
 543            process.terminate()
 544            process.kill()
 545            process.wait(timeout=timeout)
 546        except Exception as e:
 547            return False, f"Failed to kill job {self} ({process}) with exception: {e}"
 548
 549        try:
 550            if process.status():
 551                return False, "Failed to stop daemon '{self}' ({process})."
 552        except psutil.NoSuchProcess:
 553            pass
 554
 555        if self.pid_path.exists():
 556            try:
 557                self.pid_path.unlink()
 558            except Exception:
 559                pass
 560
 561        self._write_stop_file('kill')
 562        self.stdin_file.close()
 563        self._remove_blocking_stdin_file()
 564        return True, "Success"
 565
 566    def quit(self, timeout: Union[int, float, None] = None) -> SuccessTuple:
 567        """Gracefully quit a running daemon."""
 568        if self.status == 'paused':
 569            return self.kill(timeout)
 570
 571        signal_success, signal_msg = self._send_signal(signal.SIGINT, timeout=timeout)
 572        if signal_success:
 573            self._write_stop_file('quit')
 574            self.stdin_file.close()
 575            self._remove_blocking_stdin_file()
 576        return signal_success, signal_msg
 577
 578    def pause(
 579        self,
 580        timeout: Union[int, float, None] = None,
 581        check_timeout_interval: Union[float, int, None] = None,
 582    ) -> SuccessTuple:
 583        """
 584        Pause the daemon if it is running.
 585
 586        Parameters
 587        ----------
 588        timeout: Union[float, int, None], default None
 589            The maximum number of seconds to wait for a process to suspend.
 590
 591        check_timeout_interval: Union[float, int, None], default None
 592            The number of seconds to wait between checking if the process is still running.
 593
 594        Returns
 595        -------
 596        A `SuccessTuple` indicating whether the `Daemon` process was successfully suspended.
 597        """
 598        self._remove_blocking_stdin_file()
 599
 600        if self.process is None:
 601            return False, f"Daemon '{self.daemon_id}' is not running and cannot be paused."
 602
 603        if self.status == 'paused':
 604            return True, f"Daemon '{self.daemon_id}' is already paused."
 605
 606        self._write_stop_file('pause')
 607        self.stdin_file.close()
 608        self._remove_blocking_stdin_file()
 609        try:
 610            self.process.suspend()
 611        except Exception as e:
 612            return False, f"Failed to pause daemon '{self.daemon_id}':\n{e}"
 613
 614        timeout = self.get_timeout_seconds(timeout)
 615        check_timeout_interval = self.get_check_timeout_interval_seconds(
 616            check_timeout_interval
 617        )
 618
 619        psutil = attempt_import('psutil')
 620
 621        if not timeout:
 622            try:
 623                success = self.process.status() == 'stopped'
 624            except psutil.NoSuchProcess:
 625                success = True
 626            msg = "Success" if success else f"Failed to suspend daemon '{self.daemon_id}'."
 627            if success:
 628                self._capture_process_timestamp('paused')
 629            return success, msg
 630
 631        begin = time.perf_counter()
 632        while (time.perf_counter() - begin) < timeout:
 633            try:
 634                if self.process.status() == 'stopped':
 635                    self._capture_process_timestamp('paused')
 636                    return True, "Success"
 637            except psutil.NoSuchProcess as e:
 638                return False, f"Process exited unexpectedly. Was it killed?\n{e}"
 639            time.sleep(check_timeout_interval)
 640
 641        return False, (
 642            f"Failed to pause daemon '{self.daemon_id}' within {timeout} second"
 643            + ('s' if timeout != 1 else '') + '.'
 644        )
 645
 646    def resume(
 647        self,
 648        timeout: Union[int, float, None] = None,
 649        check_timeout_interval: Union[float, int, None] = None,
 650    ) -> SuccessTuple:
 651        """
 652        Resume the daemon if it is paused.
 653
 654        Parameters
 655        ----------
 656        timeout: Union[float, int, None], default None
 657            The maximum number of seconds to wait for a process to resume.
 658
 659        check_timeout_interval: Union[float, int, None], default None
 660            The number of seconds to wait between checking if the process is still stopped.
 661
 662        Returns
 663        -------
 664        A `SuccessTuple` indicating whether the `Daemon` process was successfully resumed.
 665        """
 666        if self.status == 'running':
 667            return True, f"Daemon '{self.daemon_id}' is already running."
 668
 669        if self.status == 'stopped':
 670            return False, f"Daemon '{self.daemon_id}' is stopped and cannot be resumed."
 671
 672        self._remove_stop_file()
 673        try:
 674            if self.process is None:
 675                return False, f"Cannot resume daemon '{self.daemon_id}'."
 676
 677            self.process.resume()
 678        except Exception as e:
 679            return False, f"Failed to resume daemon '{self.daemon_id}':\n{e}"
 680
 681        timeout = self.get_timeout_seconds(timeout)
 682        check_timeout_interval = self.get_check_timeout_interval_seconds(
 683            check_timeout_interval
 684        )
 685
 686        if not timeout:
 687            success = self.status == 'running'
 688            msg = "Success" if success else f"Failed to resume daemon '{self.daemon_id}'."
 689            if success:
 690                self._capture_process_timestamp('began')
 691            return success, msg
 692
 693        begin = time.perf_counter()
 694        while (time.perf_counter() - begin) < timeout:
 695            if self.status == 'running':
 696                self._capture_process_timestamp('began')
 697                return True, "Success"
 698            time.sleep(check_timeout_interval)
 699
 700        return False, (
 701            f"Failed to resume daemon '{self.daemon_id}' within {timeout} second"
 702            + ('s' if timeout != 1 else '') + '.'
 703        )
 704
 705    def _write_stop_file(self, action: str) -> SuccessTuple:
 706        """Write the stop file timestamp and action."""
 707        if action not in ('quit', 'kill', 'pause'):
 708            return False, f"Unsupported action '{action}'."
 709
 710        if not self.stop_path.parent.exists():
 711            self.stop_path.parent.mkdir(parents=True, exist_ok=True)
 712
 713        with open(self.stop_path, 'w+', encoding='utf-8') as f:
 714            json.dump(
 715                {
 716                    'stop_time': datetime.now(timezone.utc).isoformat(),
 717                    'action': action,
 718                },
 719                f
 720            )
 721
 722        return True, "Success"
 723
 724    def _remove_stop_file(self) -> SuccessTuple:
 725        """Remove the stop file"""
 726        if not self.stop_path.exists():
 727            return True, "Stop file does not exist."
 728
 729        try:
 730            self.stop_path.unlink()
 731        except Exception as e:
 732            return False, f"Failed to remove stop file:\n{e}"
 733
 734        return True, "Success"
 735
 736    def _read_stop_file(self) -> Dict[str, Any]:
 737        """
 738        Read the stop file if it exists.
 739        """
 740        if not self.stop_path.exists():
 741            return {}
 742
 743        try:
 744            with open(self.stop_path, 'r', encoding='utf-8') as f:
 745                data = json.load(f)
 746            return data
 747        except Exception:
 748            return {}
 749
 750    def _remove_blocking_stdin_file(self) -> mrsm.SuccessTuple:
 751        """
 752        Remove the blocking STDIN file if it exists.
 753        """
 754        try:
 755            if self.blocking_stdin_file_path.exists():
 756                self.blocking_stdin_file_path.unlink()
 757        except Exception as e:
 758            return False, str(e)
 759
 760        return True, "Success"
 761
 762    def _handle_sigterm(self, signal_number: int, stack_frame: 'frame') -> None:
 763        """
 764        Handle `SIGTERM` within the `Daemon` context.
 765        This method is injected into the `DaemonContext`.
 766        """
 767        from meerschaum.utils.process import signal_handler
 768        from meerschaum.utils.threading import request_stop, interrupt_threads
 769        signal_handler(signal_number, stack_frame)
 770
 771        ### Tell cooperative loops (e.g. `sync pipes`) to stop, then actively unwind
 772        ### any worker threads so they cannot keep the process alive as a zombie.
 773        request_stop()
 774        interrupt_threads(SystemExit)
 775
 776        timer = self.__dict__.get('_log_refresh_timer', None)
 777        if timer is not None:
 778            timer.cancel()
 779
 780        daemon_context = self.__dict__.get('_daemon_context', None)
 781        if daemon_context is not None:
 782            daemon_context.close()
 783
 784        _close_pools()
 785        raise SystemExit(0)
 786
 787    def _send_signal(
 788        self,
 789        signal_to_send,
 790        timeout: Union[float, int, None] = None,
 791        check_timeout_interval: Union[float, int, None] = None,
 792    ) -> SuccessTuple:
 793        """Send a signal to the daemon process.
 794
 795        Parameters
 796        ----------
 797        signal_to_send:
 798            The signal the send to the daemon, e.g. `signals.SIGINT`.
 799
 800        timeout: Union[float, int, None], default None
 801            The maximum number of seconds to wait for a process to terminate.
 802
 803        check_timeout_interval: Union[float, int, None], default None
 804            The number of seconds to wait between checking if the process is still running.
 805
 806        Returns
 807        -------
 808        A SuccessTuple indicating success.
 809        """
 810        try:
 811            pid = self.pid
 812            if pid is None:
 813                return (
 814                    False,
 815                    f"Daemon '{self.daemon_id}' is not running, "
 816                    + f"cannot send signal '{signal_to_send}'."
 817                )
 818            
 819            os.kill(pid, signal_to_send)
 820        except Exception:
 821            return False, f"Failed to send signal {signal_to_send}:\n{traceback.format_exc()}"
 822
 823        timeout = self.get_timeout_seconds(timeout)
 824        check_timeout_interval = self.get_check_timeout_interval_seconds(
 825            check_timeout_interval
 826        )
 827
 828        if not timeout:
 829            return True, f"Successfully sent '{signal_to_send}' to daemon '{self.daemon_id}'."
 830
 831        begin = time.perf_counter()
 832        while (time.perf_counter() - begin) < timeout:
 833            if not self.status == 'running':
 834                return True, "Success"
 835            time.sleep(check_timeout_interval)
 836
 837        return False, (
 838            f"Failed to stop daemon '{self.daemon_id}' (PID: {pid}) within {timeout} second"
 839            + ('s' if timeout != 1 else '') + '.'
 840        )
 841
 842    def mkdir_if_not_exists(self, allow_dirty_run: bool = False):
 843        """Create the Daemon's directory.
 844        If `allow_dirty_run` is `False` and the directory already exists,
 845        raise a `FileExistsError`.
 846        """
 847        try:
 848            self.path.mkdir(parents=True, exist_ok=True)
 849            _already_exists = any(os.scandir(self.path))
 850        except FileExistsError:
 851            _already_exists = True
 852
 853        if _already_exists and not allow_dirty_run:
 854            error(
 855                f"Daemon '{self.daemon_id}' already exists. " +
 856                "To allow this daemon to run, do one of the following:\n"
 857                + "  - Execute `daemon.cleanup()`.\n"
 858                + f"  - Delete the directory '{self.path}'.\n"
 859                + "  - Pass `allow_dirty_run=True` to `daemon.run()`.\n",
 860                FileExistsError,
 861            )
 862
 863    @property
 864    def process(self) -> Union['psutil.Process', None]:
 865        """
 866        Return the psutil process for the Daemon.
 867        """
 868        psutil = attempt_import('psutil')
 869        pid = self.pid
 870        if pid is None:
 871            return None
 872        if '_process' not in self.__dict__ or self.__dict__['_process'].pid != int(pid):
 873            try:
 874                self._process = psutil.Process(int(pid))
 875                process_exists = True
 876            except Exception:
 877                process_exists = False
 878            if not process_exists:
 879                _ = self.__dict__.pop('_process', None)
 880                try:
 881                    if self.pid_path.exists():
 882                        self.pid_path.unlink()
 883                except Exception:
 884                    pass
 885                return None
 886        return self._process
 887
 888    @property
 889    def status(self) -> str:
 890        """
 891        Return the running status of this Daemon.
 892        """
 893        if self.process is None:
 894            return 'stopped'
 895
 896        psutil = attempt_import('psutil', lazy=False)
 897        try:
 898            if self.process.status() == 'stopped':
 899                return 'paused'
 900            if self.process.status() == 'zombie':
 901                raise psutil.NoSuchProcess(self.process.pid)
 902        except (psutil.NoSuchProcess, AttributeError):
 903            if self.pid_path.exists():
 904                try:
 905                    self.pid_path.unlink()
 906                except Exception:
 907                    pass
 908            return 'stopped'
 909
 910        return 'running'
 911
 912    @classmethod
 913    def _get_path_from_daemon_id(cls, daemon_id: str) -> pathlib.Path:
 914        """
 915        Return a Daemon's path from its `daemon_id`.
 916        """
 917        import meerschaum.config.paths as paths
 918        return paths.DAEMON_RESOURCES_PATH / daemon_id
 919
 920    @property
 921    def path(self) -> pathlib.Path:
 922        """
 923        Return the path for this Daemon's directory.
 924        """
 925        return self._get_path_from_daemon_id(self.daemon_id)
 926
 927    @classmethod
 928    def _get_properties_path_from_daemon_id(cls, daemon_id: str) -> pathlib.Path:
 929        """
 930        Return the `properties.json` path for a given `daemon_id`.
 931        """
 932        return cls._get_path_from_daemon_id(daemon_id) / 'properties.json'
 933
 934    @property
 935    def properties_path(self) -> pathlib.Path:
 936        """
 937        Return the `propterties.json` path for this Daemon.
 938        """
 939        return self._get_properties_path_from_daemon_id(self.daemon_id)
 940
 941    @property
 942    def stop_path(self) -> pathlib.Path:
 943        """
 944        Return the path for the stop file (created when manually stopped).
 945        """
 946        return self.path / '.stop.json'
 947
 948    @property
 949    def log_path(self) -> pathlib.Path:
 950        """
 951        Return the log path.
 952        """
 953        logs_cf = self.properties.get('logs', None) or {}
 954        if 'path' not in logs_cf:
 955            import meerschaum.config.paths as paths
 956            return paths.LOGS_RESOURCES_PATH / (self.daemon_id + '.log')
 957
 958        return pathlib.Path(logs_cf['path'])
 959
 960    @property
 961    def stdin_file_path(self) -> pathlib.Path:
 962        """
 963        Return the stdin file path.
 964        """
 965        return self.path / 'input.stdin'
 966
 967    @property
 968    def blocking_stdin_file_path(self) -> pathlib.Path:
 969        """
 970        Return the stdin file path.
 971        """
 972        if '_blocking_stdin_file_path' in self.__dict__:
 973            return self._blocking_stdin_file_path
 974
 975        return self.path / 'input.stdin.block'
 976
 977    @property
 978    def prompt_kwargs_file_path(self) -> pathlib.Path:
 979        """
 980        Return the file path to the kwargs for the invoking `prompt()`.
 981        """
 982        return self.path / 'prompt_kwargs.json'
 983
 984    @property
 985    def log_offset_path(self) -> pathlib.Path:
 986        """
 987        Return the log offset file path.
 988        """
 989        import meerschaum.config.paths as paths
 990        return paths.LOGS_RESOURCES_PATH / ('.' + self.daemon_id + '.log.offset')
 991
 992    @property
 993    def log_offset_lock(self) -> 'fasteners.InterProcessLock':
 994        """
 995        Return the process lock context manager.
 996        """
 997        if '_log_offset_lock' in self.__dict__:
 998            return self._log_offset_lock
 999
1000        fasteners = attempt_import('fasteners')
1001        self._log_offset_lock = fasteners.InterProcessLock(self.log_offset_path)
1002        return self._log_offset_lock
1003
1004    @property
1005    def rotating_log(self) -> RotatingFile:
1006        """
1007        The rotating log file for the daemon's output.
1008        """
1009        if '_rotating_log' in self.__dict__:
1010            return self._rotating_log
1011
1012        logs_cf = self.properties.get('logs', None) or {}
1013        write_timestamps = logs_cf.get('write_timestamps', None)
1014        if write_timestamps is None:
1015            write_timestamps = get_config('jobs', 'logs', 'timestamps', 'enabled')
1016
1017        timestamp_format = logs_cf.get('timestamp_format', None)
1018        if timestamp_format is None:
1019            timestamp_format = get_config('jobs', 'logs', 'timestamps', 'format')
1020
1021        num_files_to_keep = logs_cf.get('num_files_to_keep', None)
1022        if num_files_to_keep is None:
1023            num_files_to_keep = get_config('jobs', 'logs', 'num_files_to_keep')
1024
1025        max_file_size = logs_cf.get('max_file_size', None)
1026        if max_file_size is None:
1027            max_file_size = get_config('jobs', 'logs', 'max_file_size')
1028
1029        redirect_streams = logs_cf.get('redirect_streams', True)
1030
1031        self._rotating_log = RotatingFile(
1032            self.log_path,
1033            redirect_streams=redirect_streams,
1034            write_timestamps=write_timestamps,
1035            timestamp_format=timestamp_format,
1036            num_files_to_keep=num_files_to_keep,
1037            max_file_size=max_file_size,
1038        )
1039        return self._rotating_log
1040
1041    @property
1042    def stdin_file(self):
1043        """
1044        Return the file handler for the stdin file.
1045        """
1046        if (_stdin_file := self.__dict__.get('_stdin_file', None)):
1047            return _stdin_file
1048
1049        self._stdin_file = StdinFile(
1050            self.stdin_file_path,
1051            lock_file_path=self.blocking_stdin_file_path,
1052        )
1053        return self._stdin_file
1054
1055    @property
1056    def log_text(self) -> Union[str, None]:
1057        """
1058        Read the log files and return their contents.
1059        Returns `None` if the log file does not exist.
1060        """
1061        logs_cf = self.properties.get('logs', None) or {}
1062        write_timestamps = logs_cf.get('write_timestamps', None)
1063        if write_timestamps is None:
1064            write_timestamps = get_config('jobs', 'logs', 'timestamps', 'enabled')
1065
1066        timestamp_format = logs_cf.get('timestamp_format', None)
1067        if timestamp_format is None:
1068            timestamp_format = get_config('jobs', 'logs', 'timestamps', 'format')
1069
1070        num_files_to_keep = logs_cf.get('num_files_to_keep', None)
1071        if num_files_to_keep is None:
1072            num_files_to_keep = get_config('jobs', 'logs', 'num_files_to_keep')
1073
1074        max_file_size = logs_cf.get('max_file_size', None)
1075        if max_file_size is None:
1076            max_file_size = get_config('jobs', 'logs', 'max_file_size')
1077
1078        new_rotating_log = RotatingFile(
1079            self.rotating_log.file_path,
1080            num_files_to_keep=num_files_to_keep,
1081            max_file_size=max_file_size,
1082            write_timestamps=write_timestamps,
1083            timestamp_format=timestamp_format,
1084        )
1085        return new_rotating_log.read()
1086
1087    def readlines(self) -> List[str]:
1088        """
1089        Read the next log lines, persisting the cursor for later use.
1090        Note this will alter the cursor of `self.rotating_log`.
1091        """
1092        self.rotating_log._cursor = self._read_log_offset()
1093        lines = self.rotating_log.readlines()
1094        self._write_log_offset()
1095        return lines
1096
1097    def _read_log_offset(self) -> Tuple[int, int]:
1098        """
1099        Return the current log offset cursor.
1100
1101        Returns
1102        -------
1103        A tuple of the form (`subfile_index`, `position`).
1104        """
1105        if not self.log_offset_path.exists():
1106            return 0, 0
1107
1108        try:
1109            with open(self.log_offset_path, 'r', encoding='utf-8') as f:
1110                cursor_text = f.read()
1111            cursor_parts = cursor_text.split(' ')
1112            subfile_index, subfile_position = int(cursor_parts[0]), int(cursor_parts[1])
1113            return subfile_index, subfile_position
1114        except Exception as e:
1115            warn(f"Failed to read cursor:\n{e}")
1116        return 0, 0
1117
1118    def _write_log_offset(self) -> None:
1119        """
1120        Write the current log offset file.
1121        """
1122        with self.log_offset_lock:
1123            with open(self.log_offset_path, 'w+', encoding='utf-8') as f:
1124                subfile_index = self.rotating_log._cursor[0]
1125                subfile_position = self.rotating_log._cursor[1]
1126                f.write(f"{subfile_index} {subfile_position}")
1127
1128    @property
1129    def pid(self) -> Union[int, None]:
1130        """
1131        Read the PID file and return its contents.
1132        Returns `None` if the PID file does not exist.
1133        """
1134        if not self.pid_path.exists():
1135            return None
1136        try:
1137            with open(self.pid_path, 'r', encoding='utf-8') as f:
1138                text = f.read()
1139            if len(text) == 0:
1140                return None
1141            pid = int(text.rstrip())
1142        except Exception as e:
1143            warn(e)
1144            text = None
1145            pid = None
1146        return pid
1147
1148    @property
1149    def pid_path(self) -> pathlib.Path:
1150        """
1151        Return the path to a file containing the PID for this Daemon.
1152        """
1153        return self.path / 'process.pid'
1154
1155    @property
1156    def pid_lock(self) -> 'fasteners.InterProcessLock':
1157        """
1158        Return the process lock context manager.
1159        """
1160        if '_pid_lock' in self.__dict__:
1161            return self._pid_lock
1162
1163        fasteners = attempt_import('fasteners')
1164        self._pid_lock = fasteners.InterProcessLock(self.pid_path)
1165        return self._pid_lock
1166
1167    @property
1168    def pickle_path(self) -> pathlib.Path:
1169        """
1170        Return the path for the pickle file.
1171        """
1172        return self.path / 'pickle.pkl'
1173
1174    def read_properties(self) -> Optional[Dict[str, Any]]:
1175        """Read the properties JSON file and return the dictionary."""
1176        if not self.properties_path.exists():
1177            return None
1178        try:
1179            with open(self.properties_path, 'r', encoding='utf-8') as file:
1180                properties = json.load(file)
1181        except Exception:
1182            properties = {}
1183        
1184        return properties or {}
1185
1186    def read_pickle(self) -> Daemon:
1187        """Read a Daemon's pickle file and return the `Daemon`."""
1188        import pickle
1189        import traceback
1190        if not self.pickle_path.exists():
1191            error(f"Pickle file does not exist for daemon '{self.daemon_id}'.")
1192
1193        if self.pickle_path.stat().st_size == 0:
1194            error(f"Pickle was empty for daemon '{self.daemon_id}'.")
1195
1196        try:
1197            with open(self.pickle_path, 'rb') as pickle_file:
1198                daemon = pickle.load(pickle_file)
1199            success, msg = True, 'Success'
1200        except Exception as e:
1201            success, msg = False, str(e)
1202            daemon = None
1203            traceback.print_exception(type(e), e, e.__traceback__)
1204        if not success:
1205            error(msg)
1206        return daemon
1207
1208    @property
1209    def properties(self) -> Dict[str, Any]:
1210        """
1211        Return the contents of the properties JSON file.
1212        """
1213        try:
1214            _file_properties = self.read_properties() or {}
1215        except Exception:
1216            traceback.print_exc()
1217            _file_properties = {}
1218
1219        if not self._properties:
1220            self._properties = _file_properties
1221
1222        if self._properties is None:
1223            self._properties = {}
1224
1225        if (
1226            self._properties.get('result', None) is None
1227            and _file_properties.get('result', None) is not None
1228        ):
1229            _ = self._properties.pop('result', None)
1230
1231        if _file_properties is not None:
1232            self._properties = apply_patch_to_config(
1233                _file_properties,
1234                self._properties,
1235            )
1236
1237        return self._properties
1238
1239    @property
1240    def hidden(self) -> bool:
1241        """
1242        Return a bool indicating whether this Daemon should be displayed.
1243        """
1244        return self.daemon_id.startswith('_') or self.daemon_id.startswith('.')
1245
1246    def write_properties(self) -> SuccessTuple:
1247        """Write the properties dictionary to the properties JSON file
1248        (only if self.properties exists).
1249        """
1250        from meerschaum.utils.misc import generate_password
1251        success, msg = (
1252            False,
1253            f"No properties to write for daemon '{self.daemon_id}'."
1254        )
1255        backup_path = self.properties_path.parent / (generate_password(8) + '.json')
1256        props = self.properties
1257        if props is not None:
1258            try:
1259                self.path.mkdir(parents=True, exist_ok=True)
1260                if self.properties_path.exists():
1261                    self.properties_path.rename(backup_path)
1262                with open(self.properties_path, 'w+', encoding='utf-8') as properties_file:
1263                    json.dump(props, properties_file)
1264                success, msg = True, 'Success'
1265            except Exception as e:
1266                success, msg = False, str(e)
1267
1268        try:
1269            if backup_path.exists():
1270                if not success:
1271                    backup_path.rename(self.properties_path)
1272                else:
1273                    backup_path.unlink()
1274        except Exception as e:
1275            success, msg = False, str(e)
1276
1277        return success, msg
1278
1279    def write_pickle(self) -> SuccessTuple:
1280        """Write the pickle file for the daemon."""
1281        import pickle
1282        import traceback
1283        from meerschaum.utils.misc import generate_password
1284
1285        if not self.pickle:
1286            return True, "Success"
1287
1288        from meerschaum._internal.entry import _shells
1289        if _shells:
1290            from meerschaum._internal.shell.Shell import revert_input
1291            revert_input()
1292
1293        backup_path = self.pickle_path.parent / (generate_password(7) + '.pkl')
1294        try:
1295            self.path.mkdir(parents=True, exist_ok=True)
1296            if self.pickle_path.exists():
1297                self.pickle_path.rename(backup_path)
1298            with open(self.pickle_path, 'wb+') as pickle_file:
1299                pickle.dump(self, pickle_file)
1300            success, msg = True, "Success"
1301        except Exception as e:
1302            success, msg = False, str(e)
1303            traceback.print_exception(type(e), e, e.__traceback__)
1304        try:
1305            if backup_path.exists():
1306                if not success:
1307                    backup_path.rename(self.pickle_path)
1308                else:
1309                    backup_path.unlink()
1310        except Exception as e:
1311            success, msg = False, str(e)
1312        return success, msg
1313
1314
1315    def _setup(
1316        self,
1317        allow_dirty_run: bool = False,
1318    ) -> None:
1319        """
1320        Update properties before starting the Daemon.
1321        """
1322        if self.properties is None:
1323            self._properties = {}
1324
1325        self._properties.update({
1326            'target': {
1327                'name': self.target.__name__,
1328                'module': self.target.__module__,
1329                'args': self.target_args,
1330                'kw': self.target_kw,
1331            },
1332        })
1333        self.mkdir_if_not_exists(allow_dirty_run)
1334        _write_properties_success_tuple = self.write_properties()
1335        if not _write_properties_success_tuple[0]:
1336            error(_write_properties_success_tuple[1])
1337
1338        _write_pickle_success_tuple = self.write_pickle()
1339        if not _write_pickle_success_tuple[0]:
1340            error(_write_pickle_success_tuple[1])
1341
1342    def cleanup(self, keep_logs: bool = False) -> SuccessTuple:
1343        """
1344        Remove a daemon's directory after execution.
1345
1346        Parameters
1347        ----------
1348        keep_logs: bool, default False
1349            If `True`, skip deleting the daemon's log files.
1350
1351        Returns
1352        -------
1353        A `SuccessTuple` indicating success.
1354        """
1355        if self.path.exists():
1356            try:
1357                shutil.rmtree(self.path)
1358            except Exception as e:
1359                msg = f"Failed to clean up '{self.daemon_id}':\n{e}"
1360                warn(msg)
1361                return False, msg
1362        if not keep_logs:
1363            self.rotating_log.delete()
1364            try:
1365                if self.log_offset_path.exists():
1366                    self.log_offset_path.unlink()
1367            except Exception as e:
1368                msg = f"Failed to remove offset file for '{self.daemon_id}':\n{e}"
1369                warn(msg)
1370                return False, msg
1371        return True, "Success"
1372
1373
1374    def get_timeout_seconds(self, timeout: Union[int, float, None] = None) -> Union[int, float]:
1375        """
1376        Return the timeout value to use. Use `--timeout-seconds` if provided,
1377        else the configured default (8).
1378        """
1379        if isinstance(timeout, (int, float)):
1380            return timeout
1381        return get_config('jobs', 'timeout_seconds')
1382
1383
1384    def get_check_timeout_interval_seconds(
1385        self,
1386        check_timeout_interval: Union[int, float, None] = None,
1387    ) -> Union[int, float]:
1388        """
1389        Return the interval value to check the status of timeouts.
1390        """
1391        if isinstance(check_timeout_interval, (int, float)):
1392            return check_timeout_interval
1393        return get_config('jobs', 'check_timeout_interval_seconds')
1394
1395    @property
1396    def target_args(self) -> Union[Tuple[Any], None]:
1397        """
1398        Return the positional arguments to pass to the target function.
1399        """
1400        target_args = (
1401            self.__dict__.get('_target_args', None)
1402            or self.properties.get('target', {}).get('args', None)
1403        )
1404        if target_args is None:
1405            return tuple([])
1406
1407        return tuple(target_args)
1408
1409    @property
1410    def target_kw(self) -> Union[Dict[str, Any], None]:
1411        """
1412        Return the keyword arguments to pass to the target function.
1413        """
1414        target_kw = (
1415            self.__dict__.get('_target_kw', None)
1416            or self.properties.get('target', {}).get('kw', None)
1417        )
1418        if target_kw is None:
1419            return {}
1420
1421        return {key: val for key, val in target_kw.items()}
1422
1423    @staticmethod
1424    def _get_target_reference(target) -> Union[Dict[str, str], None]:
1425        """
1426        If `target` is an importable, top-level function (e.g. `entry`), return a
1427        ``{'module': ..., 'qualname': ...}`` reference so it can be re-imported in the
1428        daemon process instead of pickled by value.
1429
1430        Pickling such a target by value serializes its entire global graph, which can
1431        reach unpicklable live state (e.g. a connector caching an `_asyncio.Task`) and
1432        raise `TypeError: cannot pickle '_asyncio.Task' object`. dill also falls back to
1433        by-value pickling whenever it cannot match the function by identity — which
1434        happens when two copies of a package are importable (a stale install shadowing
1435        a venv), so `byref=True` alone is not enough. Returns `None` for closures,
1436        lambdas, and anything not importable, which fall back to dill.
1437        """
1438        module = getattr(target, '__module__', None)
1439        qualname = getattr(target, '__qualname__', None)
1440        if not module or not qualname:
1441            return None
1442        if '<locals>' in qualname or '<lambda>' in qualname:
1443            return None
1444        return {'module': module, 'qualname': qualname}
1445
1446    @staticmethod
1447    def _load_target_reference(target_ref: Dict[str, str]):
1448        """
1449        Re-import a target from a `{'module': ..., 'qualname': ...}` reference.
1450        """
1451        import importlib
1452        obj = importlib.import_module(target_ref['module'])
1453        for part in target_ref['qualname'].split('.'):
1454            obj = getattr(obj, part)
1455        return obj
1456
1457    def __getstate__(self):
1458        """
1459        Pickle this Daemon.
1460        """
1461        dill = attempt_import('dill')
1462        state = {
1463            'target_args': self.target_args,
1464            'target_kw': self.target_kw,
1465            'daemon_id': self.daemon_id,
1466            'label': self.label,
1467            'properties': self.properties,
1468        }
1469        target_ref = self._get_target_reference(self.target)
1470        if target_ref is not None:
1471            ### Store by reference (re-imported in the daemon process), not by value.
1472            state['target'] = None
1473            state['target_ref'] = target_ref
1474        else:
1475            state['target'] = dill.dumps(self.target, byref=True)
1476        return state
1477
1478    def __setstate__(self, _state: Dict[str, Any]):
1479        """
1480        Restore this Daemon from a pickled state.
1481        If the properties file exists, skip the old pickled version.
1482        """
1483        dill = attempt_import('dill')
1484        target_ref = _state.pop('target_ref', None)
1485        if target_ref is not None and _state.get('target', None) is None:
1486            _state['target'] = self._load_target_reference(target_ref)
1487        else:
1488            _state['target'] = dill.loads(_state['target'])
1489        self._pickle = True
1490        daemon_id = _state.get('daemon_id', None)
1491        if not daemon_id:
1492            raise ValueError("Need a daemon_id to un-pickle a Daemon.")
1493
1494        properties_path = self._get_properties_path_from_daemon_id(daemon_id)
1495        ignore_properties = properties_path.exists()
1496        if ignore_properties:
1497            _state = {
1498                key: val
1499                for key, val in _state.items()
1500                if key != 'properties'
1501            }
1502        self.__init__(**_state)
1503
1504
1505    def __repr__(self):
1506        return str(self)
1507
1508    def __str__(self):
1509        return self.daemon_id
1510
1511    def __eq__(self, other):
1512        if not isinstance(other, Daemon):
1513            return False
1514        return self.daemon_id == other.daemon_id
1515
1516    def __hash__(self):
1517        return hash(self.daemon_id)