From 9a482cb04bcb46e345f9248f7d51a488fba39411 Mon Sep 17 00:00:00 2001
From: A_D <aunderscored@gmail.com>
Date: Wed, 22 Jul 2020 14:40:08 +0200
Subject: [PATCH] Added type annotations where needed
Not everywhere because they can be inferred in a lot of places. But I
added them to a lot of the self.* variables
---
monitor.py | 93 +++++++++++++++++++++++++++++-------------------------
1 file changed, 50 insertions(+), 43 deletions(-)
diff --git a/monitor.py b/monitor.py
index ac04abf0..c5a9a41d 100644
--- a/monitor.py
+++ b/monitor.py
@@ -8,7 +8,10 @@ from os.path import basename, expanduser, isdir, join
from sys import platform
from time import gmtime, localtime, sleep, strftime, strptime, time
from calendar import timegm
-from typing import Any, Dict
+from typing import Any, Dict, List, Optional, OrderedDict as OrderedDictT, Tuple, TYPE_CHECKING
+
+if TYPE_CHECKING:
+ import tkinter
if __debug__:
from traceback import print_exc
@@ -47,7 +50,8 @@ else:
# Journal handler
-class EDLogs(FileSystemEventHandler):
+class EDLogs(FileSystemEventHandler): # type: ignore # See below
+ # Magic with FileSystemEventHandler can confuse type checkers when they do not have access to every import
_POLL = 1 # Polling is cheap, so do it often
_RE_CANONICALISE = re.compile(r'\$(.+)_name;')
@@ -55,13 +59,14 @@ class EDLogs(FileSystemEventHandler):
_RE_LOGFILE = re.compile(r'^Journal(Beta)?\.[0-9]{12}\.[0-9]{2}\.log$')
def __init__(self):
+ # TODO(A_D): A bunch of these should be switched to default values (eg '' for strings) and no longer be Optional
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.logfile = None
+ self.currentdir: Optional[str] = None # The actual logdir that we're monitoring
+ self.logfile: Optional[str] = None
self.observer = None
self.observed = None # a watchdog ObservedWatch, or None if polling
- self.thread = None
+ self.thread: Optional[threading.Thread] = None
self.event_queue = [] # For communicating journal entries back to main thread
# On startup we might be:
@@ -75,19 +80,19 @@ class EDLogs(FileSystemEventHandler):
self.game_was_running = False # For generation the "ShutDown" event
# Context for journal handling
- self.version = None
+ self.version: Optional[str] = None
self.is_beta = False
- self.mode = None
- self.group = None
- self.cmdr = None
- self.planet = None
- self.system = None
- self.station = None
- self.station_marketid = None
- self.stationtype = None
- self.coordinates = None
- self.systemaddress = None
- self.started = None # Timestamp of the LoadGame event
+ self.mode: Optional[str] = None
+ self.group: Optional[str] = None
+ self.cmdr: Optional[str] = None
+ self.planet: Optional[str] = None
+ self.system: Optional[str] = None
+ self.station: Optional[str] = None
+ self.station_marketid: Optional[int] = None
+ self.stationtype: Optional[str] = None
+ self.coordinates: Optional[Tuple[int, int, int]] = None
+ self.systemaddress: Optional[int] = None
+ self.started: Optional[int] = None # Timestamp of the LoadGame event
# Cmdr state shared with EDSM and plugins
# If you change anything here update PLUGINS.md documentation!
@@ -117,10 +122,9 @@ class EDLogs(FileSystemEventHandler):
'Modules': None,
}
- def start(self, root):
+ def start(self, root: 'tkinter.Tk'):
self.root = root
- logdir = expanduser(config.get('journaldir') or config.default_journal_dir) # type: ignore # config is weird
-
+ logdir: str = config.get('journaldir') or config.default_journal_dir # type: ignore # config does weird things
if not logdir or not isdir(logdir): # type: ignore # config does weird things in its get
self.stop()
return False
@@ -134,11 +138,11 @@ class EDLogs(FileSystemEventHandler):
# Do this before setting up the observer in case the journal directory has gone away
try: # TODO: This should be replaced with something specific ONLY wrapping listdir
logfiles = sorted(
- (x for x in listdir(self.currentdir) if self._RE_LOGFILE.search(x)),
+ (x for x in listdir(self.currentdir) if self._RE_LOGFILE.search(x)), # type: ignore # config is weird
key=lambda x: x.split('.')[1:]
)
- self.logfile = join(self.currentdir, logfiles[-1]) if logfiles else None
+ self.logfile = join(self.currentdir, logfiles[-1]) if logfiles else None # type: ignore # config is weird
except Exception:
self.logfile = None
@@ -244,7 +248,7 @@ class EDLogs(FileSystemEventHandler):
if self.live:
if self.game_was_running:
# Game is running locally
- entry = OrderedDict([
+ entry: OrderedDictT[str, Any] = OrderedDict([
('timestamp', strftime('%Y-%m-%dT%H:%M:%SZ', gmtime())),
('event', 'StartUp'),
('StarSystem', self.system),
@@ -286,7 +290,7 @@ class EDLogs(FileSystemEventHandler):
key=lambda x: x.split('.')[1:]
)
- newlogfile = join(self.currentdir, logfiles[-1]) if logfiles else None
+ newlogfile = join(self.currentdir, logfiles[-1]) if logfiles else None # type: ignore
except Exception:
if __debug__:
@@ -339,11 +343,13 @@ class EDLogs(FileSystemEventHandler):
self.game_was_running = self.game_running()
def parse_entry(self, line):
+ # TODO(A_D): a bunch of these can be simplified to use if itertools.product and filters
if line is None:
return {'event': None} # Fake startup event
try:
- entry: Dict[str, Any] = json.loads(line, object_pairs_hook=OrderedDict) # Preserve property order because why not?
+ # Preserve property order because why not?
+ entry: OrderedDictT[str, Any] = json.loads(line, object_pairs_hook=OrderedDict)
entry['timestamp'] # we expect this to exist
event_type = entry['event']
@@ -355,11 +361,11 @@ class EDLogs(FileSystemEventHandler):
self.mode = None
self.group = None
self.planet = None
- self.system = None
- self.station = None
- self.station_marketid = None
- self.stationtype = None
- self.stationservices = None
+ self.system: Optional[str] = None
+ self.station: Optional[str] = None
+ self.station_marketid: Optional[int] = None
+ self.stationtype: Optional[str] = None
+ self.stationservices: Optional[List[str]] = None
self.coordinates = None
self.systemaddress = None
self.started = None
@@ -408,16 +414,16 @@ class EDLogs(FileSystemEventHandler):
self.started = timegm(strptime(entry['timestamp'], '%Y-%m-%dT%H:%M:%SZ'))
# Don't set Ship, ShipID etc since this will reflect Fighter or SRV if starting in those
self.state.update({
- 'Captain': None,
- 'Credits': entry['Credits'],
- 'FID': entry.get('FID'), # From 3.3
- 'Horizons': entry['Horizons'], # From 3.0
- 'Loan': entry['Loan'],
- 'Engineers': {},
- 'Rank': {},
+ 'Captain': None,
+ 'Credits': entry['Credits'],
+ 'FID': entry.get('FID'), # From 3.3
+ 'Horizons': entry['Horizons'], # From 3.0
+ 'Loan': entry['Loan'],
+ 'Engineers': {},
+ 'Rank': {},
'Reputation': {},
'Statistics': {},
- 'Role': None,
+ 'Role': None,
})
elif event_type == 'NewCommander':
@@ -583,7 +589,7 @@ class EDLogs(FileSystemEventHandler):
self.state['Cargo'] = defaultdict(int)
# From 3.3 full Cargo event (after the first one) is written to a separate file
if 'Inventory' not in entry:
- with open(join(self.currentdir, 'Cargo.json'), 'rb') as h:
+ with open(join(self.currentdir, 'Cargo.json'), 'rb') as h: # type: ignore
entry = json.load(h, object_pairs_hook=OrderedDict) # Preserve property order because why not?
self.state['Cargo'].update({self.canonicalise(x['Name']): x['Count'] for x in entry['Inventory']})
@@ -773,7 +779,7 @@ class EDLogs(FileSystemEventHandler):
# and "hnshockmount", "$int_cargorack_size6_class1_name;" and "Int_CargoRack_Size6_Class1",
# "python" and "Python", etc.
# This returns a simple lowercased name e.g. 'hnshockmount', 'int_cargorack_size6_class1', 'python', etc
- def canonicalise(self, item: str):
+ def canonicalise(self, item: Optional[str]):
if not item:
return ''
@@ -909,6 +915,7 @@ class EDLogs(FileSystemEventHandler):
# Export ship loadout as a Loadout event
def export_ship(self, filename=None):
+ # TODO(A_D): Some type checking has been disabled in here due to config.get getting weird outputs
string = json.dumps(self.ship(False), ensure_ascii=False, indent=2, separators=(',', ': ')) # pretty print
if filename:
with open(filename, 'wt') as h:
@@ -918,14 +925,14 @@ class EDLogs(FileSystemEventHandler):
ship = ship_file_name(self.state['ShipName'], self.state['ShipType'])
regexp = re.compile(re.escape(ship) + r'\.\d{4}\-\d\d\-\d\dT\d\d\.\d\d\.\d\d\.txt')
- oldfiles = sorted((x for x in listdir(config.get('outdir')) if regexp.match(x)))
+ oldfiles = sorted((x for x in listdir(config.get('outdir')) if regexp.match(x))) # type: ignore
if oldfiles:
- with open(join(config.get('outdir'), oldfiles[-1]), 'rU') as h:
+ with open(join(config.get('outdir'), oldfiles[-1]), 'rU') as h: # type: ignore
if h.read() == string:
return # same as last time - don't write
# Write
- filename = join(
+ filename = join( # type: ignore
config.get('outdir'), '{}.{}.txt'.format(ship, strftime('%Y-%m-%dT%H.%M.%S', localtime(time())))
)