diff --git a/.gitignore b/.gitignore index c9ddc9b..45db0fd 100644 --- a/.gitignore +++ b/.gitignore @@ -1,4 +1,5 @@ *.egg-info/ +*.log *.swo *.swp .eggs/ diff --git a/.pip.lock b/.pip.lock index cfd79ab..099a2da 100644 --- a/.pip.lock +++ b/.pip.lock @@ -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 diff --git a/requirements.txt b/requirements.txt index 5d5fd7c..01e8c97 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,7 +1,6 @@ --index-url https://1hejIc-6NBRw68BmAkuppZVHsgbOfkQPs0@pypi.fury.io/boomtechnology/ --extra-index-url https://pypi.org/simple -Mako==1.0.7 mamba==0.9.3 pycodestyle==2.3.1 pydocstyle==2.1.1 diff --git a/setup.py b/setup.py index e49d9c3..25912fc 100644 --- a/setup.py +++ b/setup.py @@ -11,7 +11,7 @@ setuptools.setup( author_email="andrewminer@mac.com", packages=[ - "pycurses", + "pycursesui", ], package_dir={"": "src"}, diff --git a/src/hooks/pre-commit b/src/hooks/pre-commit index 020ca8c..45f9f8a 100755 --- a/src/hooks/pre-commit +++ b/src/hooks/pre-commit @@ -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 diff --git a/src/pycurses/__init__.py b/src/pycurses/__init__.py deleted file mode 100644 index 2caff4b..0000000 --- a/src/pycurses/__init__.py +++ /dev/null @@ -1,3 +0,0 @@ -"""A python UI framework for command-line applications using curses.""" - -__all__ = [] diff --git a/src/pycursesui/__init__.py b/src/pycursesui/__init__.py new file mode 100644 index 0000000..5359564 --- /dev/null +++ b/src/pycursesui/__init__.py @@ -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", +] diff --git a/src/pycursesui/logger.py b/src/pycursesui/logger.py new file mode 100644 index 0000000..9ce7201 --- /dev/null +++ b/src/pycursesui/logger.py @@ -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 diff --git a/src/pycursesui/session.py b/src/pycursesui/session.py new file mode 100644 index 0000000..8f7a052 --- /dev/null +++ b/src/pycursesui/session.py @@ -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 diff --git a/src/pycursesui/time.py b/src/pycursesui/time.py new file mode 100644 index 0000000..5d54f25 --- /dev/null +++ b/src/pycursesui/time.py @@ -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) diff --git a/src/test/main.py b/src/test/main.py new file mode 100644 index 0000000..4a4a666 --- /dev/null +++ b/src/test/main.py @@ -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) diff --git a/src/test/session_spec.py b/src/test/session_spec.py new file mode 100644 index 0000000..7cfa16e --- /dev/null +++ b/src/test/session_spec.py @@ -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()