diff --git a/dashboard.py b/dashboard.py index 6d6255bf..e68a621d 100644 --- a/dashboard.py +++ b/dashboard.py @@ -11,54 +11,44 @@ import json import sys import time import tkinter as tk -from calendar import timegm -from os.path import expanduser +from datetime import datetime from pathlib import Path -from typing import Any, cast -from watchdog.observers.api import BaseObserver +from typing import Any +from watchdog.events import FileSystemEventHandler +from watchdog.observers import Observer +from watchdog.observers.polling import PollingObserver from config import config from EDMCLogging import get_main_logger logger = get_main_logger() -if sys.platform == 'win32': - from watchdog.events import FileSystemEventHandler - from watchdog.observers import Observer -else: - # Linux's inotify doesn't work over CIFS or NFS, so poll - class FileSystemEventHandler: # type: ignore - """Dummy class to represent a file system event handler on platforms other than Windows.""" - class Dashboard(FileSystemEventHandler): - """Status.json handler.""" - - _POLL = 1 # Fallback polling interval + """Status.json handler leveraging unified Watchdog observers.""" def __init__(self) -> None: - FileSystemEventHandler.__init__(self) # futureproofing - not need for current version of watchdog + super().__init__() self.session_start: int = int(time.time()) self.root: tk.Tk = None # type: ignore - self.currentdir: str = None # type: ignore # The actual logdir that we're monitoring - self.observer: Observer | None = None # type: ignore - self.observed = None # a watchdog ObservedWatch, or None if polling - self.status: dict[str, Any] = {} # Current status for communicating status back to main thread + self.currentdir: Path | None = None # The actual logdir that we're monitoring + self.observer: Observer | PollingObserver | None = None # type: ignore + self.status: dict[str, Any] = {} # Current status for communicating back to main thread def start(self, root: tk.Tk, started: int) -> bool: """ Start monitoring of Journal directory. :param root: tkinter parent window. - :param started: unix epoch timestamp of LoadGame event. Ref: monitor.started. + :param started: unix epoch timestamp of LoadGame event. Ref: monitor.started. :return: Successful start. """ logger.debug('Starting...') self.root = root self.session_start = started - logdir = expanduser(config.get_str('journaldir', default=config.default_journal_dir)) - logdir = logdir or config.default_journal_dir - if not Path.is_dir(Path(logdir)): + logdir_str = config.get_str('journaldir', default=config.default_journal_dir) + logdir = Path(logdir_str).expanduser() if logdir_str else Path(config.default_journal_dir).expanduser() + if not logdir.is_dir(): logger.info(f"No logdir, or it isn't a directory: {logdir=}") self.stop() return False @@ -70,34 +60,26 @@ class Dashboard(FileSystemEventHandler): self.currentdir = logdir # Set up a watchdog observer. - # File system events are unreliable/non-existent over network drives on Linux. - # We can't easily tell whether a path points to a network drive, so assume - # any non-standard logdir might be on a network drive and poll instead. - if sys.platform == 'win32' and not self.observer: + # Native file system events are unreliable over network drives (CIFS/NFS). + # We use standard native Observer on Windows, and PollingObserver everywhere else + # to ensure seamless cross-platform compatibility without manual tkinter loops. + if not self.observer: logger.debug('Setting up observer...') - self.observer = Observer() + if sys.platform == 'win32': + self.observer = Observer() + else: + self.observer = PollingObserver(timeout=1.0) + self.observer.daemon = True + self.observer.schedule(self, path=str(self.currentdir), recursive=False) self.observer.start() logger.debug('Done') - elif (sys.platform != 'win32') and self.observer: - logger.debug('Using polling, stopping observer...') - self.observer.stop() - self.observer = None # type: ignore - logger.debug('Done') + logger.info(f'{"Monitoring" if sys.platform == "win32" else "Polling"} Dashboard "{self.currentdir}"') - if not self.observed and sys.platform == 'win32': - logger.debug('Starting observer...') - self.observed = cast(BaseObserver, self.observer).schedule(self, self.currentdir) # type: ignore - logger.debug('Done') - - logger.info(f'{(sys.platform != "win32") and "Polling" or "Monitoring"} Dashboard "{self.currentdir}"') - - # Even if we're not intending to poll, poll at least once to process pre-existing - # data and to check whether the watchdog thread has crashed due to events not - # being supported on this filesystem. - logger.debug('Polling once to process pre-existing data, and check whether watchdog thread crashed...') - self.root.after(int(self._POLL * 1000/2), self.poll, True) + # Process the initial file state immediately to catch pre-existing data + logger.debug('Processing initial state...') + self.process() logger.debug('Done.') return True @@ -105,13 +87,16 @@ class Dashboard(FileSystemEventHandler): def stop(self) -> None: """Stop monitoring dashboard.""" logger.debug('Stopping monitoring Dashboard') - self.currentdir = None # type: ignore + self.currentdir = None - if self.observed: - logger.debug('Was observed') - self.observed = None - logger.debug('Unscheduling all observer') - self.observer.unschedule_all() + if self.observer: + logger.debug('Stopping observer thread...') + try: + self.observer.stop() + self.observer.join(timeout=2.0) + except Exception: + logger.exception('Error tearing down observer') + self.observer = None logger.debug('Done.') self.status = {} @@ -121,74 +106,56 @@ class Dashboard(FileSystemEventHandler): """Close down dashboard.""" logger.debug('Calling self.stop()') self.stop() - - if self.observer: - logger.debug('Calling self.observer.stop()') - self.observer.stop() # type: ignore - logger.debug('Done') - - if self.observer: - logger.debug('Joining self.observer...') - self.observer.join() # type: ignore - logger.debug('Done') - self.observer = None # type: ignore - logger.debug('Done.') - def poll(self, first_time: bool = False) -> None: - """ - Poll Status.json via calling self.process() once a second. - - :param first_time: True if first call of this. - """ - if not self.currentdir: - # Stopped - self.status = {} - - else: - self.process() - - if first_time: - emitter = None - # Watchdog thread - if self.observed: - emitter = self.observer._emitter_for_watch[self.observed] # Note: Uses undocumented attribute - - if emitter and emitter.is_alive(): # type: ignore - return # Watchdog thread still running - stop polling - - self.root.after(self._POLL * 1000, self.poll) # keep polling - def on_modified(self, event) -> None: """ - Watchdog callback - FileModifiedEvent on Windows. + Watchdog callback - FileModifiedEvent. :param event: Watchdog event. """ + if event.is_directory: + self.process() + return + modpath = Path(event.src_path) - if event.is_directory or (modpath.is_file() and modpath.stat().st_size): - # Can get on_modified events when the file is emptied - self.process(event.src_path if not event.is_directory else None) + if modpath.name == 'Status.json' and modpath.stat().st_size > 0: + self.process() def process(self, logfile: str | None = None) -> None: """ Process the contents of current Status.json file. - Can be called either in watchdog thread or, if polling, in main thread. + Safely handles intermittent race conditions from concurrent game writes. """ - if config.shutting_down: + if config.shutting_down or not self.currentdir: + return + + status_json_path = self.currentdir / 'Status.json' + if not status_json_path.is_file(): return try: - status_json_path = expanduser(Path(self.currentdir) / 'Status.json') with open(status_json_path, 'rb') as h: data = h.read().strip() - if data: # Can be empty if polling while the file is being re-written - entry = json.loads(data) - # Status file is shared between beta and live. Filter out status not in this game session. - entry_timestamp = timegm(time.strptime(entry['timestamp'], '%Y-%m-%dT%H:%M:%SZ')) - if entry_timestamp >= self.session_start and self.status != entry: - self.status = entry + + if not data: + return # File is currently empty/being rewritten by the game + + entry = json.loads(data) + timestamp_str = entry.get('timestamp', '') + + if timestamp_str: + dt = datetime.fromisoformat(timestamp_str.replace('Z', '+00:00')) + entry_timestamp = int(dt.timestamp()) + + # Filter out status changes not relevant to the active game session + if entry_timestamp >= self.session_start and self.status != entry: + self.status = entry + if self.root: self.root.event_generate('<>', when="tail") + + except (json.JSONDecodeError, KeyError): + logger.debug('Status.json was caught in a partially written state. Skipping frame.') except Exception: logger.exception('Processing Status.json') diff --git a/tests/test_dashboard.py b/tests/test_dashboard.py index 49681c28..b3f2c430 100644 --- a/tests/test_dashboard.py +++ b/tests/test_dashboard.py @@ -37,65 +37,88 @@ class TestDashboard: with patch("sys.platform", "win32"), patch( "dashboard.Observer" ) as mock_obs_cls, patch("dashboard.config") as mock_config: + mock_config.shutting_down = False mock_config.get_str.return_value = str(temp_journal_dir) db = dashboard.Dashboard() - success = db.start(mock_root, started=0) - assert success is True - assert db.observer is not None - mock_obs_cls.return_value.start.assert_called_once() - # Ensure a poll is scheduled to catch pre-existing data - mock_root.after.assert_called() + try: + assert success is True + assert db.observer is not None + mock_obs_cls.return_value.start.assert_called_once() + assert db.status["event"] == "Status" + mock_root.event_generate.assert_called_with( + "<>", when="tail" + ) + finally: + db.stop() def test_start_logic_linux_polling(self, mock_root, temp_journal_dir): """Verify polling behavior on Linux/non-Windows platforms.""" - with patch("sys.platform", "linux"), patch("dashboard.config") as mock_config: + with patch("sys.platform", "linux"), patch( + "dashboard.PollingObserver" + ) as mock_poll_obs_cls, patch("dashboard.config") as mock_config: + mock_config.shutting_down = False mock_config.get_str.return_value = str(temp_journal_dir) db = dashboard.Dashboard() success = db.start(mock_root, started=0) - assert success is True - assert db.observer is None # Should be None on Linux - # Ensure the polling loop is started - mock_root.after.assert_called() + try: + assert success is True + assert db.observer is not None + mock_poll_obs_cls.return_value.start.assert_called_once() + assert db.status["event"] == "Status" + finally: + db.stop() def test_process_valid_json(self, mock_root, temp_journal_dir): """Verify Status.json content is parsed and triggers a UI event.""" - db = dashboard.Dashboard() - db.currentdir = str(temp_journal_dir) - db.root = mock_root - db.session_start = 0 # Ensure timestamp check passes + with patch("dashboard.config") as mock_config: + mock_config.shutting_down = False + db = dashboard.Dashboard() + db.currentdir = temp_journal_dir + db.root = mock_root + db.session_start = 0 - db.process() + db.process() - assert db.status["event"] == "Status" - mock_root.event_generate.assert_called_with("<>", when="tail") + assert db.status["event"] == "Status" + mock_root.event_generate.assert_called_with( + "<>", when="tail" + ) def test_process_stale_data_filter(self, mock_root, temp_journal_dir): """Verify that status updates from previous sessions are ignored.""" - db = dashboard.Dashboard() - db.currentdir = str(temp_journal_dir) - db.root = mock_root + with patch("dashboard.config") as mock_config: + mock_config.shutting_down = False + db = dashboard.Dashboard() + db.currentdir = temp_journal_dir + db.root = mock_root - # Set session start to a future date relative to the file timestamp - db.session_start = 2000000000 + # Set session start to a future date relative to the file timestamp + db.session_start = 2000000000 - db.process() + db.process() - # Status should remain empty because file timestamp < session_start - assert db.status == {} - mock_root.event_generate.assert_not_called() + # Status should remain empty because file timestamp < session_start + assert db.status == {} + mock_root.event_generate.assert_not_called() - def test_poll_recursion(self, mock_root): - """Verify that poll schedules itself for the next interval.""" - db = dashboard.Dashboard() - db.root = mock_root - db.currentdir = "/fake/dir" + def test_process_midpoint_flush_resilience(self, mock_root, temp_journal_dir): + """Verify resilience against json.JSONDecodeError when catching partial writes.""" + with patch("dashboard.config") as mock_config: + mock_config.shutting_down = False + db = dashboard.Dashboard() + db.currentdir = temp_journal_dir + db.root = mock_root + db.session_start = 0 + status_file = temp_journal_dir / "Status.json" + status_file.write_text( + '{ "timestamp": "2026-01-25T12:00:00Z", truncated_json...' + ) - with patch.object(db, "process"): - db.poll(first_time=False) + db.process() - # Should call after() to schedule the next poll in 1000ms - mock_root.after.assert_called_with(1000, db.poll) + assert db.status == {} + mock_root.event_generate.assert_not_called()