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