WIP interactive shell

groove/shell/__init__.py

@@ -0,0 +1,7 @@
+from .base import BasePrompt
+from .quit import quit
+from .help import help
+from .browse import browse
+from .stats import stats
+from .playlist import _playlist
+from .create import create

groove/shell/base.py

@@ -0,0 +1,74 @@
+from prompt_toolkit import prompt
+from prompt_toolkit.completion import Completer, Completion
+
+
+class BasePrompt(Completer):
+
+    def __init__(self, manager=None, parent=None):
+        super(BasePrompt, self).__init__()
+
+        if (not manager and not parent):
+            raise RuntimeError("Must define either a database manager or a parent object.")
+
+        self._prompt = ''
+        self._values = []
+        self._parent = parent
+        self._manager = manager
+
+    @property
+    def usage(self):
+        return self.__class__.__name__
+
+    @property
+    def help_text(self):
+        return self.__doc__
+
+    @property
+    def manager(self):
+        if self._manager:
+            return self._manager
+        elif self._parent:
+            return self._parent.manager
+
+    @property
+    def parent(self):
+        return self._parent
+
+    @property
+    def prompt(self):
+        return self._prompt
+
+    @property
+    def values(self):
+        return self._values
+
+    def get_completions(self, document, complete_event):
+        word = document.get_word_before_cursor()
+        found = False
+        for value in self.values:
+            if word in value:
+                found = True
+                yield Completion(value, start_position=-len(word))
+        if not found:
+            try:
+                for result in self.default_completer(document, complete_event):
+                    yield result
+            except NotImplementedError:
+                pass
+
+    def start(self, cmd=''):
+        while True:
+            if not cmd:
+                cmd = prompt(f'{self.prompt} ', completer=self)
+            if not cmd:
+                return
+            cmd, *parts = cmd.split()
+            if not self.process(cmd, *parts):
+                return
+            cmd = ''
+
+    def default_completer(self, document, complete_event):
+        raise NotImplementedError()
+
+    def process(self, cmd, *parts):
+        raise NotImplementedError()

groove/shell/browse.py

@@ -0,0 +1,26 @@
+from .base import BasePrompt
+
+from rich.table import Table, Column
+from rich import print
+
+from sqlalchemy import func
+from groove import db
+from groove.playlist import Playlist
+
+
+class browse(BasePrompt):
+    """Browse the playlists."""
+
+    def process(self, cmd, *parts):
+        count = self.parent.manager.session.query(func.count(db.playlist.c.id)).scalar()
+        print(f"Displaying {count} playlists:")
+        query = self.parent.manager.session.query(db.playlist)
+        table = Table(
+            *[Column(k.name.title()) for k in db.playlist.columns]
+        )
+        for row in db.windowed_query(query, db.playlist.c.id, 1000):
+            columns = tuple(Playlist.from_row(row, self.manager.session).record)[0:-1]
+            table.add_row(*[str(col) for col in columns])
+        print()
+        print(table)
+        print()

groove/shell/create.py

@@ -0,0 +1,27 @@
+from .base import BasePrompt
+
+from slugify import slugify
+
+from groove.playlist import Playlist
+
+
+class create(BasePrompt):
+    """Create a new playlist."""
+
+    @property
+    def usage(self):
+        return "create PLAYLIST_NAME"
+
+    def process(self, cmd, *parts):
+        name = ' '.join(parts)
+        if not name:
+            print(f"Usage: {self.usage}")
+            return
+        slug = slugify(name)
+        self.parent._playlist = Playlist(
+            slug=slug,
+            name=name,
+            session=self.manager.session,
+            create_if_not_exists=True
+        )
+        return self.parent.commands['_playlist'].start()

groove/shell/help.py

@@ -0,0 +1,26 @@
+from .base import BasePrompt
+
+from rich import print
+import rich.table
+
+
+class help(BasePrompt):
+    """Display help documentation."""
+
+    @property
+    def usage(self):
+        return "help [COMMAND]"
+
+    def process(self, cmd, *parts):
+        if not parts:
+            print("Available Commands:")
+            table = rich.table.Table()
+            table.add_column("Command", style="yellow", no_wrap=True)
+            table.add_column("Description")
+            for name, obj in self.parent.commands.items():
+                if name.startswith('_'):
+                    continue
+                table.add_row(getattr(obj, 'usage', name), obj.__doc__)
+            print(table)
+        else:
+            print(f"Help for {parts}:")

groove/shell/interactive_shell.py

@@ -0,0 +1,68 @@
+from rich import print
+from slugify import slugify
+
+from groove.db.manager import database_manager
+from groove.shell.base import BasePrompt
+from groove import db
+from groove.playlist import Playlist
+
+
+class CommandPrompt(BasePrompt):
+
+    def __init__(self, manager):
+        super().__init__(manager=manager)
+        self._playlist = None
+        self._prompt = "Groove on Demand interactive shell. Try 'help' for help.\ngroove>"
+        self._completer = None
+        self._commands = None
+
+    @property
+    def playlist(self):
+        return self._playlist
+
+    @property
+    def commands(self):
+        if not self._commands:
+            self._commands = {}
+            for cmd in BasePrompt.__subclasses__():
+                if cmd.__name__ == self.__class__.__name__:
+                    continue
+                self._commands[cmd.__name__] = cmd(manager=self.manager, parent=self)
+        return self._commands
+
+    @property
+    def values(self):
+        return [k for k in self.commands.keys() if not k.startswith('_')]
+
+    def default_completer(self, document, complete_event):
+        def _formatter(row):
+            self._playlist = Playlist.from_row(row, self.manager)
+            return self.playlist.record.name
+        return self.manager.fuzzy_table_completer(
+            db.playlist,
+            db.playlist.c.name,
+            _formatter
+         ).get_completions(document, complete_event)
+
+    def process(self, cmd, *parts):
+        name = cmd + ' ' + ' '.join(parts)
+        if cmd in self.commands:
+            self.commands[cmd].start(name)
+        elif not parts:
+            print(f"Command not understood: {cmd}")
+        else:
+            slug = slugify(name)
+            self._playlist = Playlist(
+                slug=slug,
+                name=name,
+                session=self.manager.session,
+                create_if_not_exists=False
+            )
+            self.commands['_playlist'].start()
+            self._playlist = None
+        return True
+
+
+def start():
+    with database_manager() as manager:
+        CommandPrompt(manager).start()

groove/shell/playlist.py

@@ -0,0 +1,86 @@
+from .base import BasePrompt
+
+from prompt_toolkit import prompt
+from rich import print
+from sqlalchemy.exc import NoResultFound
+
+from groove import db
+
+
+class _playlist(BasePrompt):
+
+    def __init__(self, parent, manager=None):
+        super().__init__(manager=manager, parent=parent)
+        self._parent = parent
+        self._prompt = ''
+        self._commands = None
+
+    @property
+    def prompt(self):
+        return f"{self.parent.playlist}\n{self.parent.playlist.slug}> "
+
+    @property
+    def values(self):
+        return self.commands.keys()
+
+    @property
+    def commands(self):
+        if not self._commands:
+            self._commands = {
+                'show':  self.show,
+                'delete': self.delete,
+                'add': self.add,
+            }
+        return self._commands
+
+    def process(self, cmd, *parts):
+        res = True
+        if cmd in self.commands:
+            res = self.commands[cmd](parts)
+        else:
+            print(f"Command not understood: {cmd}")
+        return res is True
+
+    def show(self, parts):
+        print(self.parent.playlist)
+        return True
+
+    def add(self, parts):
+        print("Add tracks one at a time by title. ENTER to finish.")
+        while True:
+            text = prompt(
+                '  ? ',
+                completer=self.manager.fuzzy_table_completer(
+                    db.track,
+                    db.track.c.relpath,
+                    lambda row: row.relpath
+                ),
+                complete_in_thread=True, complete_while_typing=True
+            )
+            if not text:
+                return True
+            self._add_track(text)
+
+    def _add_track(self, text):
+        sess = self.parent.manager.session
+        try:
+            track = sess.query(db.track).filter(db.track.c.relpath == text).one()
+            self.parent.playlist.create_entries([track])
+        except NoResultFound:
+            print("No match for '{text}'")
+            return
+        return text
+
+    def delete(self, parts):
+        res = prompt(
+            'Type DELETE to permanently delete the playlist '
+            f'"{self.parent.playlist.record.name}".\nDELETE {self.prompt}'
+        )
+        if res != 'DELETE':
+            print("Delete aborted. No changes have been made.")
+            return True
+
+        self.parent.playlist.delete()
+        print("Deleted the playlist.")
+        self.parent._playlist = None
+        return False

groove/shell/quit.py

@@ -0,0 +1,8 @@
+from .base import BasePrompt
+
+
+class quit(BasePrompt):
+    """Exit the interactive shell."""
+
+    def process(self, cmd, *parts):
+        raise SystemExit()

groove/shell/stats.py

@@ -0,0 +1,16 @@
+from .base import BasePrompt
+
+from sqlalchemy import func
+from rich import print
+
+from groove import db
+
+
+class stats(BasePrompt):
+
+    def process(self, cmd, *parts):
+        sess = self.parent.manager.session
+        playlists = sess.query(func.count(db.playlist.c.id)).scalar()
+        entries = sess.query(func.count(db.entry.c.track)).scalar()
+        tracks = sess.query(func.count(db.track.c.relpath)).scalar()
+        print(f"Database contains {playlists} playlists with a total of {entries} entries, from {tracks} known tracks.")