322 lines
12 KiB
Python
322 lines
12 KiB
Python
# config.py - deals with loading configuration information.
|
|
|
|
# Loads config data from a .cfg file. Also caches the compiled
|
|
# data back into a .cfc file.
|
|
|
|
# If you are wondering how to avoid needing .cfg files (eg,
|
|
# if you are freezing Pythonwin etc) I suggest you create a
|
|
# .py file, and put the config info in a docstring. Then
|
|
# pass a CStringIO file (rather than a filename) to the
|
|
# config manager.
|
|
import sys
|
|
import string
|
|
from . import keycodes
|
|
import marshal
|
|
import stat
|
|
import os
|
|
import types
|
|
import traceback
|
|
import pywin
|
|
import glob
|
|
import imp
|
|
|
|
import win32api
|
|
|
|
debugging = 0
|
|
if debugging:
|
|
import win32traceutil # Some trace statements fire before the interactive window is open.
|
|
def trace(*args):
|
|
sys.stderr.write(" ".join(map(str, args)) + "\n")
|
|
else:
|
|
trace = lambda *args: None
|
|
|
|
compiled_config_version = 3
|
|
|
|
def split_line(line, lineno):
|
|
comment_pos = line.find("#")
|
|
if comment_pos>=0: line = line[:comment_pos]
|
|
sep_pos = line.rfind("=")
|
|
if sep_pos == -1:
|
|
if line.strip():
|
|
print("Warning: Line %d: %s is an invalid entry" % (lineno, repr(line)))
|
|
return None, None
|
|
return "", ""
|
|
return line[:sep_pos].strip(), line[sep_pos+1:].strip()
|
|
|
|
def get_section_header(line):
|
|
# Returns the section if the line is a section header, else None
|
|
if line[0] == "[":
|
|
end = line.find("]")
|
|
if end==-1: end=len(line)
|
|
rc = line[1:end].lower()
|
|
try:
|
|
i = rc.index(":")
|
|
return rc[:i], rc[i+1:]
|
|
except ValueError:
|
|
return rc, ""
|
|
return None, None
|
|
|
|
def find_config_file(f):
|
|
return os.path.join(pywin.__path__[0], f + ".cfg")
|
|
|
|
def find_config_files():
|
|
return [os.path.split(x)[1]
|
|
for x in [os.path.splitext(x)[0] for x in glob.glob(os.path.join(pywin.__path__[0], "*.cfg"))]
|
|
]
|
|
|
|
class ConfigManager:
|
|
def __init__(self, f):
|
|
self.filename = "unknown"
|
|
self.last_error = None
|
|
self.key_to_events = {}
|
|
if hasattr(f, "readline"):
|
|
fp = f
|
|
self.filename = "<config string>"
|
|
compiled_name = None
|
|
else:
|
|
try:
|
|
f = find_config_file(f)
|
|
src_stat = os.stat(f)
|
|
except os.error:
|
|
self.report_error("Config file '%s' not found" % f)
|
|
return
|
|
self.filename = f
|
|
self.basename = os.path.basename(f)
|
|
trace("Loading configuration", self.basename)
|
|
compiled_name = os.path.splitext(f)[0] + ".cfc"
|
|
try:
|
|
cf = open(compiled_name, "rb")
|
|
try:
|
|
ver = marshal.load(cf)
|
|
ok = compiled_config_version == ver
|
|
if ok:
|
|
kblayoutname = marshal.load(cf)
|
|
magic = marshal.load(cf)
|
|
size = marshal.load(cf)
|
|
mtime = marshal.load(cf)
|
|
if magic == imp.get_magic() and \
|
|
win32api.GetKeyboardLayoutName() == kblayoutname and \
|
|
src_stat[stat.ST_MTIME] == mtime and \
|
|
src_stat[stat.ST_SIZE] == size:
|
|
self.cache = marshal.load(cf)
|
|
trace("Configuration loaded cached", compiled_name)
|
|
return # We are ready to roll!
|
|
finally:
|
|
cf.close()
|
|
except (os.error, IOError, EOFError):
|
|
pass
|
|
fp = open(f)
|
|
self.cache = {}
|
|
lineno = 1
|
|
line = fp.readline()
|
|
while line:
|
|
# Skip to the next section (maybe already there!)
|
|
section, subsection = get_section_header(line)
|
|
while line and section is None:
|
|
line = fp.readline()
|
|
if not line: break
|
|
lineno = lineno + 1
|
|
section, subsection = get_section_header(line)
|
|
if not line: break
|
|
|
|
if section=="keys":
|
|
line, lineno = self._load_keys(subsection, fp, lineno)
|
|
elif section == "extensions":
|
|
line, lineno = self._load_extensions(subsection, fp, lineno)
|
|
elif section == "idle extensions":
|
|
line, lineno = self._load_idle_extensions(subsection, fp, lineno)
|
|
elif section == "general":
|
|
line, lineno = self._load_general(subsection, fp, lineno)
|
|
else:
|
|
self.report_error("Unrecognised section header '%s:%s'" % (section,subsection))
|
|
line = fp.readline()
|
|
lineno = lineno + 1
|
|
# Check critical data.
|
|
if not self.cache.get("keys"):
|
|
self.report_error("No keyboard definitions were loaded")
|
|
if not self.last_error and compiled_name:
|
|
try:
|
|
cf = open(compiled_name, "wb")
|
|
marshal.dump(compiled_config_version, cf)
|
|
marshal.dump(win32api.GetKeyboardLayoutName(), cf)
|
|
marshal.dump(imp.get_magic(), cf)
|
|
marshal.dump(src_stat[stat.ST_SIZE], cf)
|
|
marshal.dump(src_stat[stat.ST_MTIME], cf)
|
|
marshal.dump(self.cache, cf)
|
|
cf.close()
|
|
except (IOError, EOFError):
|
|
pass # Ignore errors - may be read only.
|
|
|
|
def configure(self, editor, subsections = None):
|
|
# Execute the extension code, and find any events.
|
|
# First, we "recursively" connect any we are based on.
|
|
if subsections is None: subsections = []
|
|
subsections = [''] + subsections
|
|
general = self.get_data("general")
|
|
if general:
|
|
parents = general.get("based on", [])
|
|
for parent in parents:
|
|
trace("Configuration based on", parent, "- loading.")
|
|
parent = self.__class__(parent)
|
|
parent.configure(editor, subsections)
|
|
if parent.last_error:
|
|
self.report_error(parent.last_error)
|
|
|
|
bindings = editor.bindings
|
|
codeob = self.get_data("extension code")
|
|
if codeob is not None:
|
|
ns = {}
|
|
try:
|
|
exec(codeob, ns)
|
|
except:
|
|
traceback.print_exc()
|
|
self.report_error("Executing extension code failed")
|
|
ns = None
|
|
if ns:
|
|
num = 0
|
|
for name, func in list(ns.items()):
|
|
if type(func)==types.FunctionType and name[:1] != '_':
|
|
bindings.bind(name, func)
|
|
num = num + 1
|
|
trace("Configuration Extension code loaded", num, "events")
|
|
# Load the idle extensions
|
|
for subsection in subsections:
|
|
for ext in self.get_data("idle extensions", {}).get(subsection, []):
|
|
try:
|
|
editor.idle.IDLEExtension(ext)
|
|
trace("Loaded IDLE extension", ext)
|
|
except:
|
|
self.report_error("Can not load the IDLE extension '%s'" % ext)
|
|
|
|
# Now bind up the key-map (remembering a reverse map
|
|
subsection_keymap = self.get_data("keys")
|
|
num_bound = 0
|
|
for subsection in subsections:
|
|
keymap = subsection_keymap.get(subsection, {})
|
|
bindings.update_keymap(keymap)
|
|
num_bound = num_bound + len(keymap)
|
|
trace("Configuration bound", num_bound, "keys")
|
|
|
|
def get_key_binding(self, event, subsections = None):
|
|
if subsections is None: subsections = []
|
|
subsections = [''] + subsections
|
|
|
|
subsection_keymap = self.get_data("keys")
|
|
for subsection in subsections:
|
|
map = self.key_to_events.get(subsection)
|
|
if map is None: # Build it
|
|
map = {}
|
|
keymap = subsection_keymap.get(subsection, {})
|
|
for key_info, map_event in list(keymap.items()):
|
|
map[map_event] = key_info
|
|
self.key_to_events[subsection] = map
|
|
|
|
info = map.get(event)
|
|
if info is not None:
|
|
return keycodes.make_key_name( info[0], info[1] )
|
|
return None
|
|
|
|
def report_error(self, msg):
|
|
self.last_error = msg
|
|
print("Error in %s: %s" % (self.filename, msg))
|
|
def report_warning(self, msg):
|
|
print("Warning in %s: %s" % (self.filename, msg))
|
|
|
|
def _readline(self, fp, lineno, bStripComments = 1):
|
|
line = fp.readline()
|
|
lineno = lineno + 1
|
|
if line:
|
|
bBreak = get_section_header(line)[0] is not None # A new section is starting
|
|
if bStripComments and not bBreak:
|
|
pos = line.find("#")
|
|
if pos>=0: line=line[:pos]+"\n"
|
|
else:
|
|
bBreak=1
|
|
return line, lineno, bBreak
|
|
|
|
def get_data(self, name, default=None):
|
|
return self.cache.get(name, default)
|
|
|
|
def _save_data(self, name, data):
|
|
self.cache[name] = data
|
|
return data
|
|
|
|
def _load_general(self, sub_section, fp, lineno):
|
|
map = {}
|
|
while 1:
|
|
line, lineno, bBreak = self._readline(fp, lineno)
|
|
if bBreak: break
|
|
|
|
key, val = split_line(line, lineno)
|
|
if not key: continue
|
|
key = key.lower()
|
|
l = map.get(key, [])
|
|
l.append(val)
|
|
map[key]=l
|
|
self._save_data("general", map)
|
|
return line, lineno
|
|
|
|
def _load_keys(self, sub_section, fp, lineno):
|
|
# Builds a nested dictionary of
|
|
# (scancode, flags) = event_name
|
|
main_map = self.get_data("keys", {})
|
|
map = main_map.get(sub_section, {})
|
|
while 1:
|
|
line, lineno, bBreak = self._readline(fp, lineno)
|
|
if bBreak: break
|
|
|
|
key, event = split_line(line, lineno)
|
|
if not event: continue
|
|
sc, flag = keycodes.parse_key_name(key)
|
|
if sc is None:
|
|
self.report_warning("Line %d: Invalid key name '%s'" % (lineno, key))
|
|
else:
|
|
map[sc, flag] = event
|
|
main_map[sub_section] = map
|
|
self._save_data("keys", main_map)
|
|
return line, lineno
|
|
|
|
def _load_extensions(self, sub_section, fp, lineno):
|
|
start_lineno = lineno
|
|
lines = []
|
|
while 1:
|
|
line, lineno, bBreak = self._readline(fp, lineno, 0)
|
|
if bBreak: break
|
|
lines.append(line)
|
|
try:
|
|
c = compile(
|
|
"\n" * start_lineno + # produces correct tracebacks
|
|
"".join(lines), self.filename, "exec")
|
|
self._save_data("extension code", c)
|
|
except SyntaxError as details:
|
|
errlineno = details.lineno + start_lineno
|
|
# Should handle syntax errors better here, and offset the lineno.
|
|
self.report_error("Compiling extension code failed:\r\nFile: %s\r\nLine %d\r\n%s" \
|
|
% (details.filename, errlineno, details.msg))
|
|
return line, lineno
|
|
|
|
def _load_idle_extensions(self, sub_section, fp, lineno):
|
|
extension_map = self.get_data("idle extensions")
|
|
if extension_map is None: extension_map = {}
|
|
extensions = []
|
|
while 1:
|
|
line, lineno, bBreak = self._readline(fp, lineno)
|
|
if bBreak: break
|
|
line = line.strip()
|
|
if line:
|
|
extensions.append(line)
|
|
extension_map[sub_section] = extensions
|
|
self._save_data("idle extensions", extension_map)
|
|
return line, lineno
|
|
|
|
def test():
|
|
import time
|
|
start = time.clock()
|
|
f="default"
|
|
cm = ConfigManager(f)
|
|
map = cm.get_data("keys")
|
|
took = time.clock()-start
|
|
print("Loaded %s items in %.4f secs" % (len(map), took))
|
|
|
|
if __name__=='__main__':
|
|
test()
|