WIP - Price Check app

db_utils.py

@@ -4,6 +4,21 @@
 # All rights reserved
 #
 # THIS SOFTWARE IS PROVIDED AS IS WITHOUT WARRANTY
+from txn_view import (
+    get_transactions_statement,
+    get_session_transactions_statement,
+)
+from dateutil.parser import parse as parse_time
+import pandas as pd
+NON_IDENTIFIER_COLUMNS = [
+    'ts',
+    'store',
+    'quantity',
+    'unit',
+    'price',
+    'organic',
+]
+
 def cursor_as_dict(cur):
     _col_idx_map=dict(map(lambda col: (col[1].name, col[0]), enumerate(cur.description)))
     for row in map(lambda row, _map=_col_idx_map: dict([
@@ -11,3 +26,73 @@ def cursor_as_dict(cur):
     ]), cur.fetchall()):
         #print(row)
         yield row
+
+def get_transactions(cursor, statement, display):
+    cursor.execute(statement)
+    yield from  map(lambda x: dict([
+        (k, display(v, k)) for k,v in x.items()
+    ]), cursor_as_dict(cursor))
+
+def get_session_transactions(cursor, statement, date, store):
+    #print(cur.mogrify(statement).decode("utf-8"))
+    #input()
+    cursor.execute(statement)
+    df = pd.DataFrame(cursor_as_dict(cursor))
+    if df.empty:
+        return ''
+    return df.drop(labels=[
+        'id', 'ts', 'store', 'code',
+    ], axis=1).to_string(header=[
+        'Description', 'Volume', 'Unit', 'Price', '$/unit', 'Total',
+        'Group', 'Category', 'Product', 'Organic',
+    ], justify='justify-all', max_colwidth=60, index=False)
+
+
+def record_matches(record, strict=None, **kwargs):
+    strict = strict or []
+    for k,v in kwargs.items():
+        if not v:
+            continue
+        
+        if k in strict and v.lower() != record[k].lower():
+            return False
+        
+        if v.lower() not in record[k].lower():
+            return False
+        
+    return True
+
+def unique_suggestions(cur, statement, name, display, exclude=NON_IDENTIFIER_COLUMNS, **kwargs):
+    exclude = filter(
+        lambda x: x != name or name == 'ts',
+        exclude,
+    )
+    [ kwargs.pop(k) for k in exclude if k in kwargs]
+    items = suggestions(cur, statement, name, display, exclude=exclude, **kwargs)
+    return sorted(set(map(lambda x: x[name], items)))
+
+def suggestions(cur, statement, name, display, exclude=NON_IDENTIFIER_COLUMNS, **kwargs):
+    exclude = filter(
+        lambda x: x != name or name == 'ts',
+        exclude,
+    )
+    [ kwargs.pop(k) for k in exclude if k in kwargs]
+    yield from filter(lambda x: record_matches(
+        x, strict=[ k for k in kwargs if k != name ], **kwargs
+    ), get_transactions(cur, statement, display))
+
+class QueryManager(object):
+    
+    def __init__(self, activity_manager, cursor, display):
+        self.display = display
+        self.cursor = cursor
+        self.activity_manager = activity_manager
+    
+    def get_session_transactions(self, date, store):
+        return get_session_transactions(
+            self.cursor, get_session_transactions_statement(
+                parse_time(date), store, full_name=True, exact_time=True
+            ), date, store)
+
+    def unique_suggestions(self, name, **kwargs):
+        return unique_suggestions(self.cursor, get_transactions_statement(), name, self.display, **kwargs)

grocery_transactions.py

@@ -10,7 +10,7 @@ import itertools
 import sys
 import urwid
 import pandas as pd
-from dateutil.parser import parse as parse_time
+from db_utils import QueryManager
 from widgets import (
     NoTabCheckBox,
     AutoCompleteEdit,
@@ -31,11 +31,6 @@ except:
     password = ''
 
 try:
-    from txn_view import (
-        get_transactions_statement,
-        get_session_transactions_statement,
-    )
-    from db_utils import cursor_as_dict
     import psycopg2
     from psycopg2.sql import SQL
     import os
@@ -76,15 +71,6 @@ cols = [
     )
 ]
 
-NON_IDENTIFIER_COLUMNS = [
-    'ts',
-    'store',
-    'quantity',
-    'unit',
-    'price',
-    'organic',
-]
-
 display_map = {
     'ts': lambda x: f"{time.strftime('%Y-%m-%d %H:%M', (x.year, x.month, x.day, x.hour, x.minute, 0, 0, 0, 0))}",
     'price': lambda x: f"{x:.2f}",
@@ -93,60 +79,6 @@ display_map = {
 }
 display = lambda data, name: display_map[name](data) if name in display_map else data
 
-def get_session_transactions(cursor, statement, date, store):
-    #print(cur.mogrify(statement).decode("utf-8"))
-    #input()
-    cursor.execute(statement)
-    df = pd.DataFrame(cursor_as_dict(cursor))
-    if df.empty:
-        return ''
-    return df.drop(labels=[
-        'id', 'ts', 'store', 'code',
-    ], axis=1).to_string(header=[
-        'Description', 'Volume', 'Unit', 'Price', '$/unit', 'Total',
-        'Group', 'Category', 'Product', 'Organic',
-    ], justify='justify-all', max_colwidth=60, index=False)
-
-
-def get_transactions(cursor, statement):
-    cursor.execute(statement)
-    yield from  map(lambda x: dict([
-        (k, display(v, k)) for k,v in x.items()
-    ]), cursor_as_dict(cursor))
-
-def record_matches(record, strict=None, **kwargs):
-    strict = strict or []
-    for k,v in kwargs.items():
-        if not v:
-            continue
-        
-        if k in strict and v.lower() != record[k].lower():
-            return False
-        
-        if v.lower() not in record[k].lower():
-            return False
-        
-    return True
-
-def unique_suggestions(cur, statement, name, exclude=NON_IDENTIFIER_COLUMNS, **kwargs):
-    exclude = filter(
-        lambda x: x != name or name == 'ts',
-        exclude,
-    )
-    [ kwargs.pop(k) for k in exclude if k in kwargs]
-    items = suggestions(cur, statement, name, exclude=exclude, **kwargs)
-    return sorted(set(map(lambda x: x[name], items)))
-
-def suggestions(cur, statement, name, exclude=NON_IDENTIFIER_COLUMNS, **kwargs):
-    exclude = filter(
-        lambda x: x != name or name == 'ts',
-        exclude,
-    )
-    [ kwargs.pop(k) for k in exclude if k in kwargs]
-    yield from filter(lambda x: record_matches(
-        x, strict=[ k for k in kwargs if k != name ], **kwargs
-    ), get_transactions(cur, statement))
-
 def show_or_exit(key):
     if isinstance(key, tuple):
         return
@@ -154,11 +86,6 @@ def show_or_exit(key):
     if key in ('esc',):
         raise urwid.ExitMainLoop()
 
-def interleave(_list, div):
-    for element in _list:
-        yield element
-        yield div
-
 def _apply_choice_callback(activity_manager, name, widget, value):
     txn = activity_manager.get('transaction')
     txn.apply_choice(name)(widget, value)
@@ -173,35 +100,16 @@ def _show_suggestions_callback(activity_manager, name, options):
             lambda: activity_manager.show(txn.update())
         ))
 
-def _save_and_clear_cb(activity_manager):
+def _autocomplete_callback(activity_manager, query_manager, name, data):
+    options = query_manager.unique_suggestions(name, **data)
+    if len(options) > 0:
+        _show_suggestions_callback(activity_manager, name, options)
+
+def _save_and_clear_callback(activity_manager):
     txn = activity_manager.get('transaction')
     activity_manager.app.save(txn.data)
     txn.clear()
     activity_manager.show(txn.update())
-    
-
-class QueryManager(object):
-    
-    def __init__(self, activity_manager):
-        self.activity_manager = activity_manager
-    
-    def get_session_transactions(self, date, store):
-        return get_session_transactions(
-            cur, get_session_transactions_statement(
-                parse_time(date), store, full_name=True, exact_time=True
-            ), date, store)
-    
-    def autocomplete(self, name, data):
-        options = self.unique_suggestions(name, **data)
-        if len(options) > 0:
-            _show_suggestions_callback(self.activity_manager, name, options)
-
-    def unique_suggestions(self, name, **kwargs):
-        return unique_suggestions(cur, get_transactions_statement(), name, **kwargs)
-
-#screen = urwid.raw_display.Screen()
-#screen.set_terminal_properties(colors=256, has_underline=True)
-#screen.register_palette(palette)
 
 args = sys.argv
 log = args[1]
@@ -209,11 +117,12 @@ log = args[1]
 cur.execute("BEGIN")
 
 activity_manager = ActivityManager()
-query_manager = QueryManager(activity_manager)
+query_manager = QueryManager(activity_manager, cur, display)
 
 activity_manager.create(TransactionEditor, 'transaction',
     query_manager, cols, grid_layout, side_pane, bottom_pane,
-    lambda: _save_and_clear_cb(activity_manager))
+    lambda: _save_and_clear_callback(activity_manager),
+    lambda name, data: _autocomplete_callback(activity_manager, query_manager, name, data))
 
 app = GroceryTransactionEditor(activity_manager, cur, log)
 

price_check.py

@@ -0,0 +1,344 @@
+#!/usr/bin/python3
+#
+# Copyright (c) Daniel Sheffield 2021
+#
+# All rights reserved
+#
+# THIS SOFTWARE IS PROVIDED AS IS WITHOUT WARRANTY
+import time
+import itertools
+import sys
+import urwid
+import pandas as pd
+from widgets import COPYRIGHT
+from db_utils import QueryManager
+from dateutil.parser import parse as parse_time
+from widgets import (
+    NoTabCheckBox,
+    AutoCompleteEdit,
+    AutoCompleteFloatEdit,
+    SuggestionPopup,
+    ActivityManager,
+    #_set_focus_path,
+)
+
+from collections import (
+    OrderedDict,
+)
+
+try:
+    from db_credentials import HOST, PASSWORD
+    host = f'host={HOST}'
+    password = f'password={PASSWORD}'
+except:
+    host = ''
+    password = ''
+
+try:
+    import psycopg2
+    from psycopg2.sql import SQL
+    import os
+    user = os.getenv('USER')
+    conn = psycopg2.connect(f"{host} dbname=das user={user} {password}")
+    cur = conn.cursor()
+except:
+    print('Failed to set up db connection. Entering Mock mode')
+    exit(1)
+    #from mock import *
+
+palette = [
+    ('banner', 'light gray', 'dark red'),
+    ('streak', 'light red', 'dark gray'),
+    ('bg', 'light red', 'black'),
+]
+
+top_pane = [
+    'clear',
+    'exit',
+    'sort',
+]
+
+left_pane = [
+    'product',
+    'organic',
+    None,
+]
+right_pane = [
+    'unit',
+    'quantity',
+    'price',
+]
+bottom_pane = [
+    #'rating',
+    'dbview',
+]
+
+inputs = filter(
+    lambda x: x is not None,
+    itertools.chain(
+        left_pane, right_pane
+    )
+)
+inputs_layout = [
+    ['product', 'unit', ],
+    ['organic', 'quantity', ],
+    ['price', None, ],
+]
+
+display_map = {
+    'price': lambda x: f"{x:.4f}",
+    'quantity': lambda x: f"{x:.2f}",
+    'organic': lambda x: "true" if x else "false",
+}
+display = lambda data, name: display_map[name](data) if name in display_map else data
+
+def show_or_exit(key):
+    if isinstance(key, tuple):
+        return
+    
+    if key in ('esc',):
+        raise urwid.ExitMainLoop()
+
+def _apply_choice_callback(activity_manager, name, widget, value):
+    txn = activity_manager.get('price_check')
+    txn.apply_choice(name)(widget, value)
+    activity_manager.show(txn.update())
+
+def _show_suggestions_callback(activity_manager, name, options):
+    txn = activity_manager.get('price_check')
+    activity_manager.show(
+        activity_manager.create(SuggestionPopup, None,
+            txn.original_widget, name, options,
+            lambda w,x: _apply_choice_callback(activity_manager, name, w, x),
+            lambda: activity_manager.show(txn.update())
+        ))
+
+def _autocomplete_callback(activity_manager, query_manager, name, data):
+    options = query_manager.unique_suggestions(name, **data)
+    if len(options) > 0:
+        _show_suggestions_callback(activity_manager, name, options)
+
+class GroceryPriceCheck(urwid.WidgetPlaceholder):
+    def __init__(self, activity_manager):
+        super().__init__(urwid.SolidFill(u'/'))
+        self.activity_manager = activity_manager
+        price_check = self.activity_manager.get('price_check')
+
+        self.activity_manager.show(self)
+        self.activity_manager.show(price_check)
+
+class PriceCheck(urwid.WidgetPlaceholder):
+
+    def iter_focus_paths(self):
+        initial = [2,0,0,0]
+        container = self.original_widget.original_widget.original_widget
+        _set_focus_path(container, initial)
+        while True:
+            path = container.get_focus_path()
+            yield path
+            self.advance_focus()
+            path = container.get_focus_path()
+            if path == initial:
+                self.advance_focus()
+                break
+
+    def advance_focus(self, reverse=False):
+        container = self.original_widget.original_widget.original_widget
+
+        path = container.get_focus_path()
+        if reverse:
+            paths = [ i for i in self.iter_focus_paths() ]
+            zipped_paths = zip(paths, [
+                *paths[1:], paths[0]
+            ])
+            prev_path = map(lambda x: x[0], filter(
+                lambda x: x[1] == path,
+                zipped_paths
+            ))
+            _set_focus_path(container, next(prev_path))
+            return
+        
+        _iter = [ i for i in enumerate(path) ][::-1]
+
+        for idx, part in _iter:
+            p = [ i for i in path ]
+            if reverse:
+                p[idx] -= 1
+            else:
+                p[idx] += 1
+            
+            try:
+                _set_focus_path(container, p)
+                if path == [3]:
+                    self.advance_focus(reverse=reverse)
+                return
+            except IndexError:
+                path[idx] = 0
+        
+        container.set_focus_path([2,0,0,0])
+
+    def keypress(self, size, key):
+        if isinstance(key, tuple):
+            return
+        
+        if getattr(self.original_widget.original_widget, 'original_widget', None) is None:
+            return super().keypress(size, key)
+        
+        if key == 'tab':
+            self.advance_focus()
+        elif key == 'shift tab':
+            self.advance_focus(reverse=True)
+        else:
+            return super().keypress(size, key)
+
+    def _init_data(self, fields):
+        self.data = OrderedDict([
+            (k, '') for k in fields
+        ])
+        self.clear()
+
+    def _apply_choice(self, name, value):
+        self._apply_changes(name, value)
+        for k,v in self.data.items():
+            if k == name or v:
+                continue
+            options = self.query_manager.unique_suggestions(k, **self.data)
+            if len(options) == 1 and k != 'ts':
+                self._apply_changes(k, list(options)[0])
+    
+    def apply_choice(self, name):
+        return  lambda w,x: self._apply_choice(name, x)
+
+    def _apply_changes(self, name, value):
+        self.data.update({
+            name: value,
+        })
+    
+    def apply_changes(self, name):
+        return lambda w,x: self._apply_changes(name, x)
+
+    def apply_organic_state(self, w, state):
+        self.data['organic'] = repr(state).lower()
+
+    def clear(self):
+        for k in self.data:
+            if k in ('ts', 'store',):
+                continue
+            self.data[k] = ''
+        self.organic_checkbox.set_state(False)
+    
+    def update(self):
+        for k in self.edit_fields:
+            self.edit_fields[k].set_edit_text(self.data[k])
+        date, store = self.data['ts'], self.data['store']
+        self.text_fields['dbview'].set_text(
+            self.query_manager.get_session_transactions(date, store) if None not in (
+                date or None, store or None
+            ) else ''
+        )
+        self.organic_checkbox.set_state(True if self.data['organic'] == 'true' else False)
+        return self
+
+    def __init__(self, query_manager, fields,
+        top_pane, left_pane, right_pane, bottom_pane,
+        autocomplete_cb):
+        super().__init__(urwid.SolidFill(u'/'))
+        self.organic_checkbox = NoTabCheckBox(
+            u"Organic",
+            on_state_change=self.apply_organic_state
+        )
+        self.query_manager = query_manager
+        self._init_data(fields)
+        self.top_pane = top_pane
+        self.left_pane = left_pane
+        self.right_pane = right_pane
+        self.bottom_pane = bottom_pane
+        self.edit_fields = OrderedDict()
+        self.text_fields = OrderedDict()
+        for k in self.data:
+            if k in self.right_pane and k != 'unit':
+                ef = AutoCompleteFloatEdit(('bg', k), apply_change_func=lambda name: autocomplete_cb(name, self.data))
+            elif k != 'organic':
+                ef = AutoCompleteEdit(('bg', k), apply_change_func=lambda name: autocomplete_cb(name, self.data))
+            else:
+                continue
+            
+            ef.set_edit_text(self.data[k])
+            urwid.connect_signal(ef, 'change', self.apply_changes(k))
+            self.edit_fields[k] = ef
+    
+        header = urwid.Text(u'Price Check', 'center')
+        _copyright = urwid.Text(COPYRIGHT, 'center')
+        self.buttons = OrderedDict()
+        group = []
+        for k in top_pane:
+            if k != 'sort':
+                self.buttons.update({ k:  urwid.Button(('streak', f'{k}')) })
+                continue
+            self.buttons.update({ f'{k}_price':  urwid.RadioButton(group, ('streak', u'Best Price'), state="first True",
+                on_state_change=None, user_data=[ self.buttons[b] for b in self.buttons if 'sort' in b ] )})
+            self.buttons.update({ f'{k}_date':  urwid.RadioButton(group, ('streak', u'Last Price'), state="first True",
+                on_state_change=None, user_data=[ self.buttons[b] for b in self.buttons if 'sort' in b ] )})
+        
+        urwid.connect_signal(self.buttons['clear'], 'click', lambda: None)
+        urwid.connect_signal(self.buttons['exit'], 'click', lambda: None)
+        
+        banner = urwid.Pile([
+            urwid.Padding(header, 'center', width=('relative', 100)),
+            urwid.Padding(_copyright, 'center', width=('relative', 100)),
+        ])
+        banner = urwid.AttrMap(banner, 'banner')
+        fields = dict([
+            (k, urwid.LineBox(urwid.AttrMap(self.edit_fields[k], 'streak'), title=k.title(), title_align='left')) for k in self.edit_fields
+        ])
+        fields['organic'] = urwid.LineBox(urwid.AttrMap(self.organic_checkbox, 'bg'))
+        dbview = urwid.Text('')
+        self.text_fields.update({'dbview': dbview})
+        fields.update({
+            'dbview': urwid.LineBox(
+                urwid.AttrMap(dbview, 'streak'),
+                title="Historic Prices",
+                title_align='center',
+            )
+        })
+        
+        right_pane_widget = (12, urwid.Pile(map(
+            lambda x: fields[x] if x is not None else urwid.Divider(),
+            self.right_pane
+        )))
+        left_pane_widget = urwid.Pile(map(
+            lambda x: fields[x] if x is not None else urwid.Divider(),
+            self.left_pane
+        ))
+        
+        widget = urwid.Pile([
+            banner,
+            urwid.Divider(),
+            urwid.Columns([v for v in self.buttons.values()],
+                dividechars=2,
+            ),
+            urwid.Columns((left_pane_widget, right_pane_widget),
+                dividechars=2,
+            ),
+            *[ fields[c] if c is not None else urwid.Divider() for c in self.bottom_pane ],
+        ])
+        widget = urwid.Filler(widget, 'top') 
+        widget = urwid.AttrMap(widget, 'bg')
+        self.original_widget = widget
+
+cur.execute("BEGIN")
+
+activity_manager = ActivityManager()
+query_manager = QueryManager(activity_manager, cur, display)
+
+activity_manager.create(PriceCheck, 'price_check',
+    query_manager, inputs, top_pane, left_pane, right_pane, bottom_pane,
+    lambda name, data: _autocomplete_callback(activity_manager, query_manager, name, data))
+
+app = GroceryPriceCheck(activity_manager)
+
+loop = urwid.MainLoop(app, palette, unhandled_input=show_or_exit)
+loop.run()
+
+cur.close()
+conn.close()

widgets.py

@@ -171,7 +171,6 @@ class GroceryTransactionEditor(urwid.WidgetPlaceholder):
         self.activity_manager.show(txn.update())
 
         self.log = self.open(log)
-        
     
     def _open(self, log):
         with open(log, 'a') as f:
@@ -316,7 +315,7 @@ class TransactionEditor(urwid.WidgetPlaceholder):
 
     def __init__(self, query_manager, fields,
         layout, side_pane, bottom_pane,
-        save_and_clear_cb):
+        save_and_clear_cb, autocomplete_cb):
         super().__init__(urwid.SolidFill(u'/'))
         self.organic_checkbox = NoTabCheckBox(
             u"Organic",
@@ -331,9 +330,9 @@ class TransactionEditor(urwid.WidgetPlaceholder):
         self.text_fields = OrderedDict()
         for k in self.data:
             if k in self.side_pane and k != 'unit':
-                ef = AutoCompleteFloatEdit(('bg', k), apply_change_func=lambda name: query_manager.autocomplete(name, self.data))
+                ef = AutoCompleteFloatEdit(('bg', k), apply_change_func=lambda name: autocomplete_cb(name, self.data))
             elif k != 'organic':
-                ef = AutoCompleteEdit(('bg', k), apply_change_func=lambda name: query_manager.autocomplete(name, self.data))
+                ef = AutoCompleteEdit(('bg', k), apply_change_func=lambda name: autocomplete_cb(name, self.data))
             else:
                 continue
             ef.set_edit_text(self.data[k])