NPO-Accounting/conservancy_beancount Changeset - e26dffa21428

Changeset - e26dffa21428

Parent rev.

Child rev.

[Not reviewed]

0 2 1

Brett Smith - 4 years ago 2020-06-09 13:04:27
brettcsmith@brettcsmith.org

reports: Add normalize_amount_func() function.

3 files changed with 94 insertions and 10 deletions:

conservancy_beancount/reports/accrual.py

conservancy_beancount/reports/core.py

tests/test_reports_core.py

0 comments (0 inline, 0 general)

conservancy_beancount/reports/accrual.py

➞

Show inline comments

@@ ... / @@ -116,66 +116,66 @@ from . import core @@
 from .. import cliutil
 from .. import config as configmod
 from .. import data
 from .. import filters
 from .. import rtutil
 PROGNAME = 'accrual-report'
 STANDARD_PATH = Path('-')
 CompoundAmount = TypeVar('CompoundAmount', data.Amount, core.Balance)
 PostGroups = Mapping[Optional[MetaValue], 'AccrualPostings']
 RTObject = Mapping[str, str]
 T = TypeVar('T')
 logger = logging.getLogger('conservancy_beancount.reports.accrual')
 class Sentinel:
     pass
 class Account(NamedTuple):
     name: str
     norm_func: Callable[[CompoundAmount], CompoundAmount]
     aging_thresholds: Sequence[int]
 class AccrualAccount(enum.Enum):
     # Note the aging report uses the same order accounts are defined here.
     # See AgingODS.start_spreadsheet().
     RECEIVABLE = Account(
         'Assets:Receivable', lambda bal: bal, [365, 120, 90, 60],
+    )
     PAYABLE = Account(
         'Liabilities:Payable', operator.neg, [365, 90, 60, 30],
+    )
     RECEIVABLE = Account('Assets:Receivable', [365, 120, 90, 60])
     PAYABLE = Account('Liabilities:Payable', [365, 90, 60, 30])
     @classmethod
     def account_names(cls) -> Iterator[str]:
         return (acct.value.name for acct in cls)
     @classmethod
     def by_account(cls, name: data.Account) -> 'AccrualAccount':
         for account in cls:
             if name.is_under(account.value.name):
                 return account
         raise ValueError(f"unrecognized account {name!r}")
     @classmethod
     def classify(cls, related: core.RelatedPostings) -> 'AccrualAccount':
         for account in cls:
             account_name = account.value.name
             if all(post.account.is_under(account_name) for post in related):
                 return account
         raise ValueError("unrecognized account set in related postings")
     @property
     def normalize_amount(self) -> Callable[[T], T]:
         return core.normalize_amount_func(self.value.name)
 class AccrualPostings(core.RelatedPostings):
     def _meta_getter(key: MetaKey) -> Callable[[data.Posting], MetaValue]:  # type:ignore[misc]
         def meta_getter(post: data.Posting) -> MetaValue:
             return post.meta.get(key)
         return meta_getter
     _FIELDS: Dict[str, Callable[[data.Posting], MetaValue]] = {
         'account': operator.attrgetter('account'),
         'contract': _meta_getter('contract'),
         'invoice': _meta_getter('invoice'),
         'purchase_order': _meta_getter('purchase-order'),
@@ ... / @@ -212,26 +212,25 @@ class AccrualPostings(core.RelatedPostings): @@
             if len(values) == 1:
                 one_value = next(iter(values))
             else:
                 one_value = self.INCONSISTENT
             setattr(self, name, one_value)
         if self.account is self.INCONSISTENT:
             self.accrual_type: Optional[AccrualAccount] = None
             self.end_balance = self.balance_at_cost()
             self.accrued_entities = self._collect_entities()
             self.paid_entities = self.accrued_entities
         else:
             self.accrual_type = AccrualAccount.classify(self)
             accrual_acct: Account = self.accrual_type.value
             norm_func = accrual_acct.norm_func
             norm_func = self.accrual_type.normalize_amount
             self.end_balance = norm_func(self.balance_at_cost())
             self.accrued_entities = self._collect_entities(
                 lambda post: norm_func(post.units).number > 0,
+            )
             self.paid_entities = self._collect_entities(
                 lambda post: norm_func(post.units).number < 0,
+            )
     def _collect_entities(self,
                           pred: Callable[[data.Posting], bool]=bool,
                           default: str='<empty>',
     ) -> FrozenSet[MetaValue]:
@@ ... @@
     def write_row(self, row: AccrualPostings) -> None:
         age = (self.date - row[0].meta.date).days
         if row.end_balance.ge_zero():
             for index, threshold in enumerate(self.age_thresholds):
                 if age >= threshold:
                     self.age_balances[index] += row.end_balance
                     break
             else:
                 return
         raw_balance = row.balance()
         if row.accrual_type is not None:
-            raw_balance = row.accrual_type.value.norm_func(raw_balance)
+            raw_balance = row.accrual_type.normalize_amount(raw_balance)
         if raw_balance == row.end_balance:
             amount_cell = odf.table.TableCell()
         else:
             amount_cell = self.balance_cell(raw_balance)
         self.add_row(
             self.date_cell(row[0].meta.date),
             self.multiline_cell(row.entities()),
             amount_cell,
             self.balance_cell(row.end_balance),
             self.multilink_cell(self._link_seq(row, 'rt-id')),
             self.multilink_cell(self._link_seq(row, 'invoice')),
+        )

conservancy_beancount/reports/core.py

➞

Show inline comments

@@ ... / @@ -64,24 +64,25 @@ from typing import ( @@
 from ..beancount_types import (
     MetaKey,
     MetaValue,
+)
 DecimalCompat = data.DecimalCompat
 BalanceType = TypeVar('BalanceType', bound='Balance')
 ElementType = Callable[..., odf.element.Element]
 LinkType = Union[str, Tuple[str, Optional[str]]]
 RelatedType = TypeVar('RelatedType', bound='RelatedPostings')
 RT = TypeVar('RT', bound=Sequence)
 ST = TypeVar('ST')
 T = TypeVar('T')
 class Balance(Mapping[str, data.Amount]):
     """A collection of amounts mapped by currency
     Each key is a Beancount currency string, and each value represents the
     balance in that currency.
     """
     __slots__ = ('_currency_map', 'tolerance')
     TOLERANCE = Decimal('0.01')
     def __init__(self,
                  source: Iterable[data.Amount]=(),
@@ ... @@
             else:
                 cell = self.string_cell(cell_source)
             out_row.addElement(cell)
         self.sheet.addElement(out_row)
     def save_file(self, out_file: BinaryIO) -> None:
         self.document.write(out_file)
     def save_path(self, path: Path, mode: str='w') -> None:
         with path.open(f'{mode}b') as out_file:
             out_file = cast(BinaryIO, out_file)
             self.save_file(out_file)
 def normalize_amount_func(account_name: str) -> Callable[[T], T]:
     """Get a function to normalize amounts for reporting
     Given an account name, return a function that can be used on "amounts"
     under that account (including numbers, Amount objects, and Balance objects)
     to normalize them for reporting. Right now that means make flipping the
     sign for accounts where "normal" postings are negative.
     """
     if account_name.startswith(('Assets:', 'Expenses:')):
         # We can't just return operator.pos because Beancount's Amount class
         # doesn't implement __pos__.
         return lambda amt: amt
     elif account_name.startswith(('Equity:', 'Income:', 'Liabilities:')):
         return operator.neg
     else:
         raise ValueError(f"unrecognized account name {account_name!r}")

tests/test_reports_core.py

➞

Show inline comments

 new file 100644
 """test_reports_core - Unit tests for basic reports functions"""
 # Copyright © 2020  Brett Smith
+#
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU Affero General Public License for more details.
+#
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 import pytest
 from decimal import Decimal
 from . import testutil
 from conservancy_beancount.reports import core
 AMOUNTS = [
 ,
     Decimal('4.40'),
     testutil.Amount('6.60', 'CHF'),
     core.Balance([testutil.Amount('8.80')]),
+]
 @pytest.mark.parametrize('acct_name', [
     'Assets:Checking',
     'Assets:Receivable:Accounts',
     'Expenses:Other',
     'Expenses:FilingFees',
 ])
 def test_normalize_amount_func_pos(acct_name):
     actual = core.normalize_amount_func(acct_name)
     for amount in AMOUNTS:
         assert actual(amount) == amount
 @pytest.mark.parametrize('acct_name', [
     'Equity:Funds:Restricted',
     'Equity:Realized:CurrencyConversion',
     'Income:Donations',
     'Income:Other',
     'Liabilities:CreditCard',
     'Liabilities:Payable:Accounts',
 ])
 def test_normalize_amount_func_neg(acct_name):
     actual = core.normalize_amount_func(acct_name)
     for amount in AMOUNTS:
         assert actual(amount) == -amount
 @pytest.mark.parametrize('acct_name', [
     '',
     'Assets',
     'Equity',
     'Expenses',
     'Income',
     'Liabilities',
 ])
 def test_normalize_amount_func_bad_acct_name(acct_name):
     with pytest.raises(ValueError):
         core.normalize_amount_func(acct_name)

0 comments (0 inline, 0 general)