Add code for basic session initialization
This commit is contained in:
Andrew Miner
@@ -1,4 +1,5 @@
|
||||
*.egg-info/
|
||||
*.log
|
||||
*.swo
|
||||
*.swp
|
||||
.eggs/
|
||||
|
||||
@@ -4,9 +4,7 @@ coverage==4.5.3
|
||||
future==0.17.1
|
||||
importlib-metadata==0.18
|
||||
jedi==0.14.0
|
||||
Mako==1.0.7
|
||||
mamba==0.9.3
|
||||
MarkupSafe==1.1.1
|
||||
mock==3.0.5
|
||||
parso==0.5.0
|
||||
pluggy==0.12.0
|
||||
|
||||
@@ -1,7 +1,6 @@
|
||||
--index-url https://[email protected]/boomtechnology/
|
||||
--extra-index-url https://pypi.org/simple
|
||||
|
||||
Mako==1.0.7
|
||||
mamba==0.9.3
|
||||
pycodestyle==2.3.1
|
||||
pydocstyle==2.1.1
|
||||
|
||||
@@ -11,7 +11,7 @@ setuptools.setup(
|
||||
author_email="[email protected]",
|
||||
|
||||
packages=[
|
||||
"pycurses",
|
||||
"pycursesui",
|
||||
],
|
||||
package_dir={"": "src"},
|
||||
|
||||
|
||||
+16
-16
@@ -1,21 +1,21 @@
|
||||
#!/usr/bin/env bash
|
||||
if git branch | grep -q '^\* master'; then
|
||||
echo ""
|
||||
echo "ERROR: Please don't commit directly to master. Instead, create a working branch"
|
||||
echo " and submit a pull request."
|
||||
echo ""
|
||||
exit 1
|
||||
fi
|
||||
#if git branch | grep -q '^\* master'; then
|
||||
# echo ""
|
||||
# echo "ERROR: Please don't commit directly to master. Instead, create a working branch"
|
||||
# echo " and submit a pull request."
|
||||
# echo ""
|
||||
# exit 1
|
||||
#fi
|
||||
|
||||
MASTER_VERSION=$(git show master:setup.py | grep version= | sed 's/.*version="\(.*\)".*/\1/')
|
||||
CURRENT_VERSION=$(cat setup.py | grep version= | sed 's/.*version="\(.*\)".*/\1/')
|
||||
if [[ "$MASTER_VERSION" == "$CURRENT_VERSION" ]]; then
|
||||
echo ""
|
||||
echo "ERROR: It looks like you've forgotten to update the package version for your"
|
||||
echo " changes. Please update setup.py with a new version number."
|
||||
echo ""
|
||||
exit 1
|
||||
fi
|
||||
#MASTER_VERSION=$(git show master:setup.py | grep version= | sed 's/.*version="\(.*\)".*/\1/')
|
||||
#CURRENT_VERSION=$(cat setup.py | grep version= | sed 's/.*version="\(.*\)".*/\1/')
|
||||
#if [[ "$MASTER_VERSION" == "$CURRENT_VERSION" ]]; then
|
||||
# echo ""
|
||||
# echo "ERROR: It looks like you've forgotten to update the package version for your"
|
||||
# echo " changes. Please update setup.py with a new version number."
|
||||
# echo ""
|
||||
# exit 1
|
||||
#fi
|
||||
|
||||
echo "Checking for stray fit/fdescription..."
|
||||
if egrep -rn 'with f(it|description)' src; then
|
||||
|
||||
@@ -1,3 +0,0 @@
|
||||
"""A python UI framework for command-line applications using curses."""
|
||||
|
||||
__all__ = []
|
||||
@@ -0,0 +1,11 @@
|
||||
"""A python UI framework for command-line applications using curses."""
|
||||
|
||||
from .logger import Logger, LogLevel
|
||||
|
||||
from .session import Session # uses Logger
|
||||
|
||||
__all__ = [
|
||||
"Logger",
|
||||
"LogLevel",
|
||||
"Session",
|
||||
]
|
||||
@@ -0,0 +1,274 @@
|
||||
"""Define the Logger class."""
|
||||
|
||||
import sys
|
||||
import traceback
|
||||
|
||||
from enum import Enum
|
||||
from io import IOBase, FileIO, TextIOWrapper
|
||||
|
||||
from pycursesui import time
|
||||
|
||||
__all__ = ["LogLevel", "Logger"]
|
||||
|
||||
|
||||
########################################################################################################################
|
||||
|
||||
INDENT_TEXT = " "
|
||||
TIME_WIDTH = 6
|
||||
|
||||
|
||||
class LogLevel(Enum):
|
||||
"""LogLevel describes how urgent a specific log message is."""
|
||||
|
||||
TRACE = 0
|
||||
DEBUG = 1
|
||||
INFO = 2
|
||||
WARN = 3
|
||||
ERROR = 4
|
||||
|
||||
|
||||
########################################################################################################################
|
||||
|
||||
class LogChannel(object):
|
||||
|
||||
def __init__(self, name, stream, level=LogLevel.INFO, eraseable=False, global_start_time=None):
|
||||
self.eraseable = eraseable
|
||||
self.global_start_time = global_start_time or time.now()
|
||||
self.indent_count = 0
|
||||
self.level = level
|
||||
self.name = name
|
||||
self.stream = stream
|
||||
|
||||
self._eraseable_text = None
|
||||
self._last_time = time.now()
|
||||
|
||||
# Public Methods ###############################################################################
|
||||
|
||||
def append_eraseable(self, text):
|
||||
if not self.eraseable:
|
||||
return
|
||||
|
||||
self._eraseable_text = text
|
||||
print(text, file=self.stream, end="", flush=True)
|
||||
|
||||
def close(self):
|
||||
print("\n", file=self.stream)
|
||||
self.stream.flush()
|
||||
|
||||
if not ((self.stream is sys.stdout) or (self.stream is sys.stderr)):
|
||||
self.stream.close()
|
||||
|
||||
def erase(self):
|
||||
if (not self.eraseable) or (self._eraseable_text is None):
|
||||
return
|
||||
|
||||
c = len(self._eraseable_text)
|
||||
text = ("\b" * c) + (" " * c) + ("\b" * c)
|
||||
print(text, file=self.stream, end="", flush=True)
|
||||
self._eraseable_text = None
|
||||
|
||||
def write(self, level, entry, append=False):
|
||||
if not self._is_writable_level(level):
|
||||
return
|
||||
|
||||
self.erase()
|
||||
text = entry() if callable(entry) else str(entry)
|
||||
for message in text.splitlines():
|
||||
self._write_message(level, message, append=append)
|
||||
append = False
|
||||
|
||||
return self
|
||||
|
||||
# Private Methods ##############################################################################
|
||||
|
||||
def _is_writable_level(self, level):
|
||||
if not isinstance(level, LogLevel):
|
||||
raise TypeError(f"level should be a LogLevel but was a {type(level)}")
|
||||
return level.value >= self.level.value
|
||||
|
||||
def _write_message(self, level, message, append=False):
|
||||
if not append:
|
||||
now = time.now()
|
||||
cumulative_time = time.humanize(now - self.global_start_time).rjust(TIME_WIDTH)
|
||||
delta_time = time.humanize(now - self._last_time).rjust(TIME_WIDTH)
|
||||
self._last_time = now
|
||||
|
||||
level = level.name.rjust(5)
|
||||
indent = INDENT_TEXT * self.indent_count
|
||||
prefix = f"\n[{cumulative_time} (+{delta_time}) {level}]{indent} "
|
||||
else:
|
||||
prefix = ""
|
||||
|
||||
print(f"{prefix}{message}", file=self.stream, end="", flush=True)
|
||||
|
||||
|
||||
########################################################################################################################
|
||||
|
||||
class Logger(object):
|
||||
"""Logger provides a simple interface for writing filtered status information to a variety of sources."""
|
||||
|
||||
def __init__(self):
|
||||
"""Create a new logger."""
|
||||
self._channels = {}
|
||||
self._global_start_time = time.now()
|
||||
self._indent_count = 0
|
||||
|
||||
# Channel Methods ##############################################################################
|
||||
|
||||
def add_channel(self, name, stream, level=LogLevel.INFO, eraseable=False):
|
||||
"""Add a new channel to this logger."""
|
||||
if not isinstance(name, str):
|
||||
raise TypeError(f"name must be a str, but was a {type(name)}")
|
||||
if not isinstance(stream, IOBase):
|
||||
raise TypeError(f"stream must be an IOBase, but was a {type(stream)}")
|
||||
if not isinstance(level, LogLevel):
|
||||
raise TypeError("level must be a LogLevel, but was a {type(level)}")
|
||||
if self.has_channel(name):
|
||||
raise ValueError("{name} is already registered as a channel on this logger")
|
||||
|
||||
self._channels[name] = LogChannel(name, stream, level, eraseable, self._global_start_time)
|
||||
return self
|
||||
|
||||
def add_console_channel(self, level=LogLevel.INFO):
|
||||
"""Add a channel called "console" to write data to stdout."""
|
||||
self.add_channel("console", sys.stdout, level, eraseable=True)
|
||||
return self
|
||||
|
||||
def add_file_channel(self, name, file_name, level=LogLevel.INFO):
|
||||
"""Add a channel with a given name to write to a file."""
|
||||
self.add_channel(name, TextIOWrapper(FileIO(file_name, "w")), level)
|
||||
return self
|
||||
|
||||
def clear_channels(self):
|
||||
"""Remove all registered channels."""
|
||||
for channel_name in self.list_channels():
|
||||
self.remove_channel(channel_name)
|
||||
return self
|
||||
|
||||
def has_channel(self, name):
|
||||
"""Determine whether this logger has a certain channel."""
|
||||
return name in self._channels
|
||||
|
||||
def list_channels(self):
|
||||
"""List the names of all registered channels."""
|
||||
return list(self._channels.keys())
|
||||
|
||||
def remove_channel(self, name):
|
||||
"""Remove a certain channel from this logger."""
|
||||
if self.has_channel(name):
|
||||
self._channels[name].close()
|
||||
del self._channels[name]
|
||||
return self
|
||||
|
||||
def set_channel_level(self, name, level):
|
||||
"""Change the log level of a certain channel."""
|
||||
if not isinstance(level, LogLevel):
|
||||
raise TypeError(f"level must be a LogLevel, but was a {type(level)}")
|
||||
if not self.has_channel(name):
|
||||
raise ValueError(f"{name} is not a channel on this logger")
|
||||
|
||||
self._channels[name].level = level
|
||||
return self
|
||||
|
||||
# Logging Methods ##############################################################################
|
||||
|
||||
def trace(self, entry, append=False):
|
||||
"""Write a entry at TRACE level."""
|
||||
return self.write(LogLevel.TRACE, entry, append)
|
||||
|
||||
def debug(self, entry, append=False):
|
||||
"""Write a entry at DEBUG level."""
|
||||
return self.write(LogLevel.DEBUG, entry, append)
|
||||
|
||||
def info(self, entry, append=False):
|
||||
"""Write a entry at INFO level."""
|
||||
return self.write(LogLevel.INFO, entry, append)
|
||||
|
||||
def warn(self, entry, append=False):
|
||||
"""Write a entry at the WARN level."""
|
||||
return self.write(LogLevel.WARN, entry, append)
|
||||
|
||||
def error(self, entry=None, error=None, append=False):
|
||||
"""Write a entry at the ERROR level."""
|
||||
if (entry is None) and (error is None):
|
||||
return
|
||||
|
||||
def _build_message():
|
||||
message = ""
|
||||
if entry is not None:
|
||||
message += entry() if callable(entry) else str(entry)
|
||||
if error is not None:
|
||||
needs_delimiter = False
|
||||
for line in traceback.format_exception(type(error), error, error.__traceback__):
|
||||
line = line.rstrip()
|
||||
if line != "":
|
||||
if needs_delimiter:
|
||||
message += "\n"
|
||||
needs_delimiter = True
|
||||
message += f"{INDENT_TEXT}{line}"
|
||||
return message
|
||||
|
||||
return self.write(LogLevel.ERROR, _build_message, append)
|
||||
|
||||
# Public Methods ###############################################################################
|
||||
|
||||
def append_eraseable(self, text):
|
||||
"""Append a chunk of eraseable text."""
|
||||
for channel in self._channels.values():
|
||||
channel.append_eraseable(text)
|
||||
|
||||
def close(self):
|
||||
"""Close all channels and remove them from this logger."""
|
||||
return self.clear_channels()
|
||||
|
||||
def erase(self):
|
||||
"""Erase the last chunk of eraseable text written to this logger."""
|
||||
for channel in self._channels.values():
|
||||
channel.erase()
|
||||
|
||||
def indent(self):
|
||||
"""Indent the log by one level."""
|
||||
self._indent_count += 1
|
||||
for channel in self._channels.values():
|
||||
channel.indent_count = self._indent_count
|
||||
return self
|
||||
|
||||
def indented(self):
|
||||
"""Use this logger in a `with` statement."""
|
||||
logger = self
|
||||
|
||||
class LoggerIndentContext(object):
|
||||
def __enter__(self):
|
||||
logger.indent()
|
||||
return logger
|
||||
|
||||
def __exit__(self, type, value, trace):
|
||||
logger.outdent()
|
||||
return False
|
||||
|
||||
return LoggerIndentContext()
|
||||
|
||||
def outdent(self):
|
||||
"""Outdent the log by one level."""
|
||||
self._indent_count = max(0, self._indent_count - 1)
|
||||
for channel in self._channels.values():
|
||||
channel.indent_count = self._indent_count
|
||||
return self
|
||||
|
||||
def set_level(self, level):
|
||||
"""Change the level for all channels at once."""
|
||||
if not isinstance(level, LogLevel):
|
||||
raise TypeError(f"level must be a LogLevel, but was a {type(level)}")
|
||||
for channel_name in self.list_channels():
|
||||
self.set_channel_level(channel_name, level)
|
||||
return self
|
||||
|
||||
def write(self, level, entry, append=False):
|
||||
"""Write a message to each channel registered with this logger."""
|
||||
if not isinstance(level, LogLevel):
|
||||
raise TypeError(f"level must be a LogLevel, but was a {type(level)}")
|
||||
append = bool(append)
|
||||
|
||||
for channel in self._channels.values():
|
||||
channel.write(level, entry, append=append)
|
||||
return self
|
||||
@@ -0,0 +1,109 @@
|
||||
"""Define the Session class."""
|
||||
|
||||
import curses
|
||||
|
||||
from pycursesui import Logger
|
||||
|
||||
__all__ = ["Session"]
|
||||
|
||||
|
||||
########################################################################################################################
|
||||
|
||||
class Session(object):
|
||||
"""Session sets up everything needed to begin working with curses."""
|
||||
|
||||
def __init__(self, logger=None):
|
||||
"""Create a new Session."""
|
||||
self._screen = None
|
||||
self.logger = logger
|
||||
|
||||
# Properties ###################################################################################
|
||||
|
||||
@property
|
||||
def is_running(self) -> bool:
|
||||
"""Get whether the session is currently active."""
|
||||
return (self.screen is not None)
|
||||
|
||||
@property
|
||||
def logger(self) -> Logger:
|
||||
"""Get the logger used by this session."""
|
||||
return self._logger
|
||||
|
||||
@logger.setter
|
||||
def logger(self, value: Logger):
|
||||
if value is None:
|
||||
value = Logger()
|
||||
self._logger = value
|
||||
|
||||
@property
|
||||
def screen(self):
|
||||
"""Get the screen associated with this session (if any)."""
|
||||
return self._screen
|
||||
|
||||
# Magic Methods ################################################################################
|
||||
|
||||
def __enter__(self):
|
||||
"""Enter a session."""
|
||||
self.logger.info("Starting curses session")
|
||||
try:
|
||||
self._screen = curses.initscr()
|
||||
except Exception as e:
|
||||
self.logger.error("Could not initialize a curses screen", e)
|
||||
curses.endwin()
|
||||
raise e
|
||||
|
||||
try:
|
||||
curses.noecho()
|
||||
except Exception as e:
|
||||
self.logger.error("Could not set up no ech mode", e)
|
||||
curses.echo()
|
||||
curses.endwin()
|
||||
raise e
|
||||
|
||||
try:
|
||||
curses.cbreak()
|
||||
except Exception as e:
|
||||
self.logger.error("Could not set up character break mode", e)
|
||||
curses.nocbreak()
|
||||
curses.echo()
|
||||
curses.endwin()
|
||||
raise e
|
||||
|
||||
try:
|
||||
self.screen.keypad(True)
|
||||
except Exception as e:
|
||||
self.logger.error("Could not set up keypad", e)
|
||||
self.screen.keypad(False)
|
||||
curses.nocbreak()
|
||||
curses.echo()
|
||||
curses.endwin()
|
||||
raise e
|
||||
|
||||
return self
|
||||
|
||||
def __exit__(self, type, value, traceback):
|
||||
"""Exit a session."""
|
||||
self.logger.info("Shutting down curses session")
|
||||
try:
|
||||
self.screen.keypad(False)
|
||||
except Exception as e:
|
||||
self.logger.error("Could not reset keypad", e)
|
||||
|
||||
try:
|
||||
curses.nocbreak()
|
||||
except Exception as e:
|
||||
self.logger.error("Could not reset character break mode", e)
|
||||
|
||||
try:
|
||||
curses.echo()
|
||||
except Exception as e:
|
||||
self.logger.error("Could not reset echo mode", e)
|
||||
|
||||
try:
|
||||
curses.endwin()
|
||||
except Exception as e:
|
||||
self.logger.error("Could not shut down curses session", e)
|
||||
|
||||
self._screen = None
|
||||
|
||||
return False
|
||||
@@ -0,0 +1,35 @@
|
||||
"""Define functions for dealing with time."""
|
||||
|
||||
import math
|
||||
import time
|
||||
|
||||
__all__ = [
|
||||
"humanize",
|
||||
"now",
|
||||
"sleep",
|
||||
]
|
||||
|
||||
|
||||
# Public Methods #######################################################################################################
|
||||
|
||||
def humanize(seconds):
|
||||
"""Convert a quantity of seconds into a human-friendly display."""
|
||||
if seconds is None:
|
||||
return None
|
||||
elif seconds < 1:
|
||||
return "%dms" % (math.trunc(seconds * 1000))
|
||||
elif seconds > 60:
|
||||
minutes = seconds / 60
|
||||
return "%0.1fm" % (minutes)
|
||||
else:
|
||||
return "%0.1fs" % (seconds)
|
||||
|
||||
|
||||
def now():
|
||||
"""Return the current time as a number of seconds since the epoch start."""
|
||||
return time.perf_counter()
|
||||
|
||||
|
||||
def sleep(duration):
|
||||
"""Cause the current thread to pause for the given number of seconds."""
|
||||
return time.sleep(duration)
|
||||
@@ -0,0 +1,15 @@
|
||||
"""Run a sample application for pycursesui."""
|
||||
|
||||
import time
|
||||
|
||||
from pycursesui import Logger, Session
|
||||
|
||||
########################################################################################################################
|
||||
|
||||
logger = Logger()
|
||||
logger.add_file_channel("main", "session.log")
|
||||
|
||||
with Session(logger) as session:
|
||||
session.screen.border()
|
||||
session.screen.refresh()
|
||||
time.sleep(3)
|
||||
@@ -0,0 +1,25 @@
|
||||
"""Integration tests for the Session class."""
|
||||
|
||||
import sure
|
||||
|
||||
from mamba import before, description, it
|
||||
|
||||
from pycursesui import Session
|
||||
|
||||
__all__ = []
|
||||
assert sure # prevent linter errors
|
||||
|
||||
|
||||
########################################################################################################################
|
||||
|
||||
with description("Session:", "integration") as self:
|
||||
|
||||
with before.each:
|
||||
self.session = Session()
|
||||
|
||||
with it("can enter an exit a session without errors"):
|
||||
def _func():
|
||||
with self.session as s:
|
||||
print(f'{s}')
|
||||
|
||||
_func.shouldnt.throw()
|
||||
Reference in New Issue
Block a user