1
0
mirror of https://github.com/EDCD/EDMarketConnector.git synced 2025-04-14 08:17:13 +03:00

dashboard.py: Fix flake8 lints

This commit is contained in:
Athanasius 2021-04-07 13:24:35 +01:00
parent 6a403a5482
commit e67e0a03fa

View File

@ -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()