add current price vs historic price badge

app/activities/PriceCheck.py

@@ -4,11 +4,9 @@
 # All rights reserved
 #
 # THIS SOFTWARE IS PROVIDED AS IS WITHOUT WARRANTY
-from collections import OrderedDict
-from decimal import Decimal, InvalidOperation
-from typing import Callable, List, Union
 import itertools
-import numpy as np
+from decimal import Decimal, InvalidOperation
+from typing import Callable, Union
 from urwid import (
     connect_signal,
     AttrMap,
@@ -22,7 +20,6 @@ from urwid import (
     RadioButton,
     Text,
 )
-from app.db_utils import QueryManager
 
 from .. import COPYRIGHT
 from ..widgets import (
@@ -32,7 +29,9 @@ from ..widgets import (
     AutoCompletePopUp,
     NoTabCheckBox
 )
+from ..db_utils import QueryManager
 from . import ActivityManager, show_or_exit
+from .Rating import Rating
 
 class PriceCheck(FocusWidget):
 
@@ -98,45 +97,6 @@ class PriceCheck(FocusWidget):
         self.update_historic_prices(self.data)
         return self
 
-    def update_rating(self, _avg, _min, _max, unit, price=None, quantity=None):
-        if None in (_avg, _min, _max):
-            return
-        current = None if None in (price, quantity or None) else float(price/quantity)
-        size = 14
-        chars = ['|', *['-']*(size - 2), '|' ]
-        rating = [' ']*len(chars)
-        _min, _max = min(_min, current or _min), max(_max, current or _max)
-        ls = np.linspace(_min, _max, len(chars))
-        if current is not None:
-            if current <= _avg:
-                p = 'badge_good'
-            else:
-                p = 'badge_bad'
-        else:
-            p = 'badge_neutral'
-        for idx, (e, a) in enumerate(zip(ls, ls[1:])):
-            if e <= _avg <= a:
-                for c, (_idx,_) in zip(''.join(list(f'{_avg:>5.2f}')[:idx+2]), filter(lambda x: idx-2 < x[0] and x[0]>0, enumerate(chars))):
-                    chars[_idx] = (p, c)
-                chars[0] = '|'
-                chars[-1] = '|'
-
-            if current is not None and e <= current < a:
-                rating[idx] = '^'
-
-        if current is not None:
-            self.text_fields['rating'].set_text(f"{current:>5.2f}/{unit} {current/_avg - 1:>4.0%}")
-
-        if _min != _max:
-            if current == _max:
-                rating[-1] = '^'
-            self.text_fields['spread'].set_text([f"{_min:>5.2f}", ('badge_highlight', chars), f"{_max:<5.2f}"])
-            self.text_fields['marker'].set_text(f"{' '*5}{''.join(rating)}{' '*5}")
-        else:
-            self.text_fields['spread'].set_text('')
-            self.text_fields['marker'].set_text('')
-
-
     def update_historic_prices(self, data):
         organic = None if data['organic'] == 'mixed' else data['organic']
         sort = '$/unit' if self.buttons['sort_price'].state else 'ts'
@@ -151,12 +111,16 @@ class PriceCheck(FocusWidget):
         except InvalidOperation:
             quantity = None
 
-        if None not in (sort, product, unit):
-            self.text_fields['dbview'].set_text(
-                self.query_manager.get_historic_prices(
-                    lambda *args: self.update_rating(*args, unit, price=price, quantity=quantity),
-                    sort, product, unit, organic=organic)
-            )
+        if None in (sort, product, unit):
+            return
+
+        df = self.query_manager.get_historic_prices_data(unit, sort=sort, product=product, organic=organic)
+        _avg, _min, _max = [ x for x in df[['avg','min','max']].iloc[0].apply(float) ]
+        self.rating.update_rating(_avg, _min, _max, unit, price=price, quantity=quantity),
+
+        self.text_fields['dbview'].set_text(
+            self.rating.get_historic_prices(df)
+        )
 
     def __init__(self,
         activity_manager: ActivityManager,
@@ -184,6 +148,10 @@ class PriceCheck(FocusWidget):
         self.text_fields = dict((
          (k, Text('')) for k in ('dbview', 'spread', 'rating', 'marker')
         ))
+        self.rating = Rating(dict(filter(
+          lambda x: x[0] in ('spread','rating','marker'),
+          self.text_fields.items()
+        )))
         top_pane = [ 'clear', 'exit', ['sort_price', 'sort_date'], ]
 
         left_pane = [
@@ -202,12 +170,6 @@ class PriceCheck(FocusWidget):
         ]
         bottom_pane = [ 'dbview', ]
 
-        layout = [
-            [ top_pane, ],
-            [ left_pane, right_pane, ],
-            [ badge, ],
-            [ bottom_pane, ],
-        ]
         self.query_manager = query_manager
         self.organic_checkbox = self.checkboxes['organic']
         connect_signal(self.organic_checkbox, 'postchange', lambda _,v: self.update())
@@ -255,51 +217,52 @@ class PriceCheck(FocusWidget):
                 title_align='center',
             ),
         })
-        components = OrderedDict((
-          ('right_pane', (16, Pile(map(
+        components = {
+          'top_pane': Columns([
+              (9, Pile([
+                 Divider(),
+                 AttrMap(self.buttons['clear'], 'streak'),
+                 Divider(),
+              ])),
+              LineBox(
+                Columns([ v for k,v in self.buttons.items() if 'sort' in k]),
+                title="Sort price by",
+                title_align='left',
+              ),
+              (9, Pile([
+                Divider(),
+                AttrMap(self.buttons['exit'], 'streak'),
+                Divider(),
+              ]))
+            ], dividechars=1),
+          'right_pane': (16, Pile(map(
             lambda x: _widgets[x] if x is not None else Divider,
             right_pane
-          )))),
-          ('left_pane', Pile([_widgets['product'], _widgets['organic']])),
-          ('badge', Pile(map(
+          ))),
+          'left_pane': Pile(map(
+            lambda x: _widgets[x] if x is not None else Divider,
+            left_pane
+          )),
+          'badge': Pile(map(
             lambda x: _widgets[x] if x is not None else Divider,
             badge
-          ))),
-          ('bottom_pane', _widgets['dbview']),
-        ))
-
-        left_pane_widget = (Pile([
+          )),
+          'bottom_pane': _widgets['dbview'],
+        }
+        components.update({
+          'left_pane': Pile([
             components['left_pane'],
             LineBox(
                 AttrMap(components['badge'], 'badge'),
                 title="Current Price", title_align='left',
             )
-        ]))
+        ])})
 
         widget = Pile([
             banner,
             Divider(),
-            Columns(
-                [
-                    (9, Pile([
-                        Divider(),
-                        AttrMap(self.buttons['clear'], 'streak'),
-                        Divider(),
-                    ])),
-                    LineBox(
-                        Columns([ v for k,v in self.buttons.items() if 'sort' in k]),
-                        title="Sort price by",
-                        title_align='left',
-                    ),
-                    (9, Pile([
-                        Divider(),
-                        AttrMap(self.buttons['exit'], 'streak'),
-                        Divider(),
-                    ]))
-                ],
-                dividechars=1,
-            ),
-            Columns((left_pane_widget, components['right_pane']),
+            components['top_pane'],
+            Columns((components['left_pane'], components['right_pane']),
                 dividechars=0,
             ),
             components['bottom_pane'],

app/activities/Rating.py

@@ -0,0 +1,59 @@
+#
+# Copyright (c) Daniel Sheffield 2021 - 2022
+#
+# All rights reserved
+#
+# THIS SOFTWARE IS PROVIDED AS IS WITHOUT WARRANTY
+import numpy as np
+class Rating(object):
+    def __init__(self, text_fields):
+        self.text_fields = text_fields
+
+    def get_historic_prices(self, df):
+        if df.empty:
+            self.update_rating(None, None, None)
+            return ''
+
+        return df.drop(labels=[
+            'id', 'avg', 'min', 'max', 'price', 'quantity', 'ts_raw', 'product', 'category', 'group'
+        ], axis=1).to_string(header=[
+            'Date', 'Store', '$/unit', 'Org',
+        ], justify='justify-all', max_colwidth=16, index=False)
+
+    def update_rating(self, _avg, _min, _max, unit, price=None, quantity=None):
+        if None in (_avg, _min, _max):
+            return
+        current = None if None in (price, quantity or None) else float(price/quantity)
+        size = 14
+        chars = ['|', *['-']*(size - 2), '|' ]
+        rating = [' ']*len(chars)
+        _min, _max = min(_min, current or _min), max(_max, current or _max)
+        ls = np.linspace(_min, _max, len(chars))
+        if current is not None:
+            if current <= _avg:
+                p = 'badge_good'
+            else:
+                p = 'badge_bad'
+        else:
+            p = 'badge_neutral'
+        for idx, (e, a) in enumerate(zip(ls, ls[1:])):
+            if e <= _avg <= a:
+                for c, (_idx,_) in zip(''.join(list(f'{_avg:>5.2f}')[:idx+2]), filter(lambda x: idx-2 < x[0] and x[0]>0, enumerate(chars))):
+                    chars[_idx] = (p, c)
+                chars[0] = '|'
+                chars[-1] = '|'
+
+            if current is not None and e <= current < a:
+                rating[idx] = '^'
+
+        if current is not None:
+            self.text_fields['rating'].set_text(f"{current:>5.2f}/{unit} {current/_avg - 1:>4.0%}")
+
+        if _min != _max:
+            if current == _max:
+                rating[-1] = '^'
+            self.text_fields['spread'].set_text([f"{_min:>5.2f}", ('badge_highlight', chars), f"{_max:<5.2f}"])
+            self.text_fields['marker'].set_text(f"{' '*5}{''.join(rating)}{' '*5}")
+        else:
+            self.text_fields['spread'].set_text('')
+            self.text_fields['marker'].set_text('')

app/activities/TransactionEditor.py

@@ -5,7 +5,8 @@
 #
 # THIS SOFTWARE IS PROVIDED AS IS WITHOUT WARRANTY
 import itertools
-from typing import Callable, List, Union
+from decimal import Decimal, InvalidOperation
+from typing import Callable, Union
 from urwid import (
     connect_signal,
     AttrMap,
@@ -18,10 +19,9 @@ from urwid import (
     Pile,
     Text,
 )
-from app.activities import ActivityManager
-from app.db_utils import QueryManager
 
 from .. import COPYRIGHT
+from ..db_utils import QueryManager
 from ..widgets import (
     AutoCompleteEdit,
     AutoCompleteFloatEdit,
@@ -29,7 +29,8 @@ from ..widgets import (
     AutoCompletePopUp,
     NoTabCheckBox
 )
-
+from . import ActivityManager
+from .Rating import Rating
 
 class TransactionEditor(FocusWidget):
 
@@ -108,8 +109,31 @@ class TransactionEditor(FocusWidget):
                 date or None, store or None
             ) else ''
         )
+        self.update_historic_prices(self.data)
         return self
 
+    def update_historic_prices(self, data):
+        organic = None if data['organic'] == 'mixed' else data['organic']
+        #sort = '$/unit' if self.buttons['sort_price'].state else 'ts'
+        product, unit = data['product'] or None, data['unit'] or None
+        try:
+            price = Decimal(data['price'])
+        except InvalidOperation:
+            price = None
+
+        try:
+            quantity = Decimal(data['quantity'])
+        except InvalidOperation:
+            quantity = None
+
+        if None in (product, unit):
+            return
+
+        df = self.query_manager.get_historic_prices_data(unit, product=product, organic=organic)
+        _avg, _min, _max = [ x for x in df[['avg','min','max']].iloc[0].apply(float) ]
+        self.rating.update_rating(_avg, _min, _max, unit, price=price, quantity=quantity),
+
+
     def focus_on_product(self):
         path = self.container.get_focus_path()
         while path != [2,0,1,0,0]:
@@ -145,8 +169,15 @@ class TransactionEditor(FocusWidget):
           'organic': NoTabCheckBox(('bg', "Organic"), state='mixed'),
         }
         self.text_fields = {
-            'dbview': Text('')
+            'dbview': Text(''),
+            'rating': Text(''),
+            'spread': Text(''),
+            'marker': Text(''),
         }
+        self.rating = Rating(dict(filter(
+          lambda x: x[0] in ('spread','rating','marker'),
+          self.text_fields.items()
+        )))
         self.organic_checkbox = self.checkboxes['organic']
         connect_signal(self.organic_checkbox, 'postchange', lambda _,v: self.update())
         layout = [
@@ -163,9 +194,18 @@ class TransactionEditor(FocusWidget):
             'description',
             'dbview',
         ]
+        badge = [
+            'rating',
+            'spread',
+            'marker',
+        ]
         self.clear()
-        widgets = dict()
-        widgets.update({
+        _widgets = dict(itertools.chain(*[
+            [(k, v) for k,v in x] for x in map(lambda x: x.items(), [
+                self.edit_fields, self.text_fields, self.checkboxes
+            ])
+        ]))
+        _widgets.update({
             'dbview': LineBox(
                 AttrMap(self.text_fields['dbview'], 'streak'),
                 title="Session Data",
@@ -176,7 +216,7 @@ class TransactionEditor(FocusWidget):
             connect_signal(ef, 'postchange', lambda _,v: self.update())
             connect_signal(ef, 'apply', lambda w, name: autocomplete_cb(w, name, self.data))
     
-        widgets.update(dict([
+        _widgets.update(dict([
             (k, LineBox(
                 AttrMap(AutoCompletePopUp(
                     self.edit_fields[k],
@@ -188,10 +228,27 @@ class TransactionEditor(FocusWidget):
         header = Text(u'Fill Transaction', 'center')
         _copyright = Text(COPYRIGHT, 'center')
 
-        components = dict()
-        components['bottom_pane'] = Columns(
-            [(8, self.buttons['done']), Divider(), (9, self.buttons['clear'])]
-        )
+        components = {
+          'bottom_button_bar': Columns(
+                [(8, self.buttons['done']), Divider(), (9, self.buttons['clear'])]
+            ),
+          'badge': Pile(map(
+            lambda x: _widgets[x] if x is not None else Divider,
+            badge
+          )),
+        }
+        components.update({
+          'bottom_pane': Columns([
+            Pile(map(
+              lambda x: _widgets[x] if x is not None else Divider(),
+              bottom_pane
+            )),
+            (50, Pile([LineBox(
+              AttrMap(components['badge'], 'badge'),
+              title="Current Price", title_align='left',
+            )])),
+          ])
+        })
         connect_signal(self.buttons['done'], 'click', lambda _: save_and_clear_cb())
         connect_signal(self.buttons['clear'], 'click', lambda _: self.clear())
         
@@ -200,7 +257,7 @@ class TransactionEditor(FocusWidget):
             Padding(_copyright, 'center', width=('relative', 100)),
         ])
         banner = AttrMap(banner, 'banner')
-        widgets.update({
+        _widgets.update({
             'product': LineBox(Columns([
                 AttrMap(AutoCompletePopUp(
                     self.edit_fields['product'],
@@ -211,30 +268,32 @@ class TransactionEditor(FocusWidget):
             ], dividechars=2), title='Product', title_align='left')
         })
         
-        side_pane_widget = (12, Pile([
-            widgets[r] if r is not None else Divider() for r in side_pane
+        components['side_pane'] = (12, Pile([
+            _widgets[r] if r is not None else Divider() for r in side_pane
         ]))
-        main_pane_widgets = []
-        for i, r in enumerate(layout):
-            _widgets = []
+        components['main_pane'] = []
+        for _, r in enumerate(layout):
+            col = []
             for c in r:
                 if c is not None:
                     if c == 'organic':
                         continue
-                    _widgets.append(widgets[c])
+                    col.append(_widgets[c])
                 else:
-                    _widgets.append(Divider())
-            main_pane_widgets.append(Columns(_widgets))
+                    col.append(Divider())
+            components['main_pane'].append(Columns(col))
 
+        components['main_pane'] = Pile(components['main_pane'])
+        
         widget = Pile([
             banner,
             Divider(),
             Columns([
-                Pile(main_pane_widgets), side_pane_widget
+                components['main_pane'], components['side_pane']
             ], dividechars=2),
-            *[ widgets[c] if c is not None else Divider() for c in bottom_pane ],
+            components['bottom_pane'],
             Divider(),
-            components['bottom_pane']
+            components['bottom_button_bar']
         ])
         widget = Filler(widget, 'top') 
         widget = AttrMap(widget, 'bg')

app/db_utils.py

@@ -133,23 +133,10 @@ class QueryManager(object):
         statement = get_historic_prices_statement(unit, sort=sort, product=product, category=category, group=group, organic=organic, limit=limit)
         #print(self.cursor.mogrify(statement).decode('utf-8'))
         #input()
-        return get_data(self.cursor, statement)
-
-    def get_historic_prices(self, rating_cb, sort, product, unit, organic=None, limit=None):
-        df = pd.DataFrame(map(lambda x: dict(
+        data = get_data(self.cursor, statement)
+        return pd.DataFrame(map(lambda x: dict(
             map(lambda k: (k, self.display(x[k], k)), x)
-        ), self.get_historic_prices_data(unit, sort=sort, product=product, organic=organic, limit=limit)))
-        if df.empty:
-            rating_cb(None, None, None)
-            return ''
-        _avg, _min, _max = [ x for x in df[['avg','min','max']].iloc[0].apply(float) ]
-        rating_cb(_avg, _min, _max)
-
-        return df.drop(labels=[
-            'id', 'avg', 'min', 'max', 'price', 'quantity', 'ts_raw', 'product', 'category', 'group'
-        ], axis=1).to_string(header=[
-            'Date', 'Store', '$/unit', 'Org',
-        ], justify='justify-all', max_colwidth=16, index=False)
+        ), data))
 
     def get_session_transactions(self, date, store):
         statement = get_session_transactions_statement(

grocery_transactions.py

@@ -5,7 +5,6 @@
 # All rights reserved
 #
 # THIS SOFTWARE IS PROVIDED AS IS WITHOUT WARRANTY
-import itertools
 import sys
 from sqlite3 import Cursor
 from typing import Union
@@ -44,6 +43,11 @@ dark_palette = [
     ('banner', 'light gray', 'dark red'),
     ('streak', 'light red', 'black'),
     ('bg', 'light red', 'black'),
+    ('badge', 'light gray', 'black'),
+    ('badge_highlight', 'black', 'light gray'),
+    ('badge_good', 'black', 'light green'),
+    ('badge_bad', 'black', 'light red'),
+    ('badge_neutral', 'white', 'dark gray'),
 ]
 
 light_palette = [
@@ -52,6 +56,11 @@ light_palette = [
     ('banner', 'light gray', 'dark red'),
     ('streak', 'white', 'dark blue'),
     ('bg', 'white', 'dark blue'),
+    ('badge', 'white', 'dark blue'),
+    ('badge_highlight', 'dark blue', 'white'),
+    ('badge_good', 'white', 'light green'),
+    ('badge_bad', 'white', 'light red'),
+    ('badge_neutral', 'white', 'dark gray'),
 ]
 
 def _insert_new_product_callback(activity_manager, query_manager, product, category, group):
@@ -63,7 +72,7 @@ def _insert_new_product_callback(activity_manager, query_manager, product, categ
 def _new_product_callback(
     activity_manager: ActivityManager,
     query_manager: QueryManager,
-    name: str, data: dict
+    name: str
 ):
     cur = activity_manager.current()
     txn : TransactionEditor = activity_manager.get('transaction')
@@ -90,7 +99,7 @@ def _autocomplete_callback(
         widget._emit('open', options)
     elif len(options) == 0 and activity_manager.current() is not activity_manager.get('new_product'):
         if name in ('product', 'category', 'group'):
-            _new_product_callback(activity_manager, query_manager, name, data)
+            _new_product_callback(activity_manager, query_manager, name)
 
 
 def _save_and_clear_callback(activity_manager):