From e67e0a03fa95ad153f50cfa975a97b625aec6313 Mon Sep 17 00:00:00 2001
From: Athanasius <github@miggy.org>
Date: Wed, 7 Apr 2021 13:24:35 +0100
Subject: [PATCH] dashboard.py: Fix flake8 lints
---
dashboard.py | 81 ++++++++++++++++++++++++++++++++++------------------
1 file changed, 54 insertions(+), 27 deletions(-)
diff --git a/dashboard.py b/dashboard.py
index 71b16f4c..e98af412 100644
--- a/dashboard.py
+++ b/dashboard.py
@@ -1,42 +1,51 @@
+"""Handle the game Status.json file."""
+
import json
-from calendar import timegm
-from os.path import isdir, isfile, join, getsize
-from sys import platform
import time
+import tkinter as tk
+from calendar import timegm
+from os.path import getsize, isdir, isfile, join
+from sys import platform
from config import config
from EDMCLogging import get_main_logger
logger = get_main_logger()
-if platform=='darwin':
- from watchdog.observers import Observer
+if platform == 'darwin':
from watchdog.events import FileSystemEventHandler
+ from watchdog.observers import Observer
-elif platform=='win32':
- from watchdog.observers import Observer
+elif 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
- FileSystemEventHandler = object # dummy
+ FileSystemEventHandler = object # dummy
-# Status.json handler
class Dashboard(FileSystemEventHandler):
+ """Status.json handler."""
_POLL = 1 # Fallback polling interval
- def __init__(self):
- FileSystemEventHandler.__init__(self) # futureproofing - not need for current version of watchdog
+ def __init__(self) -> None:
+ FileSystemEventHandler.__init__(self) # futureproofing - not need for current version of watchdog
self.root = None
- self.currentdir = None # The actual logdir that we're monitoring
+ self.currentdir = None # The actual logdir that we're monitoring
self.observer = None
- self.observed = None # a watchdog ObservedWatch, or None if polling
- self.status = {} # Current status for communicating status back to main thread
+ self.observed = None # a watchdog ObservedWatch, or None if polling
+ self.status = {} # Current status for communicating status back to main thread
- def start(self, root, started):
- """Start monitoring of Journal directory."""
+ 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.
+ :return: Successful start.
+ """
logger.debug('Starting...')
self.root = root
self.session_start = started
@@ -90,7 +99,7 @@ class Dashboard(FileSystemEventHandler):
return True
- def stop(self):
+ def stop(self) -> None:
"""Stop monitoring dashboard."""
logger.debug('Stopping monitoring Dashboard')
self.currentdir = None
@@ -105,7 +114,7 @@ class Dashboard(FileSystemEventHandler):
self.status = {}
logger.debug('Done.')
- def close(self):
+ def close(self) -> None:
"""Close down dashboard."""
logger.debug('Calling self.stop()')
self.stop()
@@ -123,28 +132,45 @@ class Dashboard(FileSystemEventHandler):
logger.debug('Done.')
- def poll(self, first_time=False):
+ 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:
# Watchdog thread
- emitter = self.observed and self.observer._emitter_for_watch[self.observed] # Note: Uses undocumented attribute
+ if self.observed:
+ emitter = self.observer._emitter_for_watch[self.observed] # Note: Uses undocumented attribute
+
if emitter and emitter.is_alive():
- return # Watchdog thread still running - stop polling
+ return # Watchdog thread still running - stop polling
- self.root.after(self._POLL * 1000, self.poll) # keep polling
+ self.root.after(self._POLL * 1000, self.poll) # keep polling
- def on_modified(self, event):
- # watchdog callback - DirModifiedEvent on macOS, FileModifiedEvent on Windows
- if event.is_directory or (isfile(event.src_path) and getsize(event.src_path)): # Can get on_modified events when the file is emptied
+ def on_modified(self, event) -> None:
+ """
+ Watchdog callback - DirModifiedEvent on macOS, FileModifiedEvent on Windows.
+
+ :param event: Watchdog event.
+ """
+ if event.is_directory or (isfile(event.src_path) and getsize(event.src_path)):
+ # Can get on_modified events when the file is emptied
self.process(event.src_path if not event.is_directory else None)
- # Can be called either in watchdog thread or, if polling, in main thread.
- def process(self, logfile=None):
+ def process(self) -> None:
+ """
+ Process the contents of current Status.json file.
+
+ Can be called either in watchdog thread or, if polling, in main thread.
+ """
if config.shutting_down:
return
@@ -166,5 +192,6 @@ class Dashboard(FileSystemEventHandler):
except Exception:
logger.exception('Processing Status.json')
+
# singleton
dashboard = Dashboard()