Changeset - 6213bc1e5da9
[Not reviewed]
0 2 1
Brett Smith - 4 years ago 2020-06-20 23:04:53
brettcsmith@brettcsmith.org
reports: Add PeriodPostings class.

This is AccountPostings from the ledger report, cleaned up to be more
general.
3 files changed with 126 insertions and 20 deletions:
0 comments (0 inline, 0 general)
conservancy_beancount/reports/core.py
Show inline comments
...
 
@@ -204,384 +204,435 @@ class Balance(Mapping[str, data.Amount]):
 
    def format(self,
 
               fmt: Optional[str]='#,##0.00 ¤¤',
 
               sep: str=', ',
 
               empty: str="Zero balance",
 
               zero: Optional[str]=None,
 
               tolerance: Optional[Decimal]=None,
 
    ) -> str:
 
        """Formats the balance as a string with the given parameters
 

	
 
        If the balance is completely empty, return ``empty``.
 
        If the balance is zero (within tolerance) and ``zero`` is specified,
 
        return ``zero``.
 
        Otherwise, return a string with each amount in the balance formatted
 
        as ``fmt``, separated by ``sep``.
 

	
 
        If you set ``fmt`` to None, amounts will be formatted according to the
 
        user's locale. The default format is Beancount's input format.
 
        """
 
        balance = self.clean_copy(tolerance) or self.copy(tolerance)
 
        if not balance:
 
            return empty
 
        elif zero is not None and balance.is_zero():
 
            return zero
 
        else:
 
            amounts = list(balance.values())
 
            amounts.sort(key=lambda amt: (-abs(amt.number), amt.currency))
 
            return sep.join(
 
                babel.numbers.format_currency(amt.number, amt.currency, fmt)
 
                for amt in amounts
 
            )
 

	
 

	
 
class MutableBalance(Balance):
 
    __slots__ = ()
 

	
 
    def __iadd__(self: BalanceType, other: Union[data.Amount, Balance]) -> BalanceType:
 
        self._add_other(self._currency_map, other)
 
        return self
 

	
 
    def __isub__(self: BalanceType, other: Union[data.Amount, Balance]) -> BalanceType:
 
        self._add_other(self._currency_map, -other)
 
        return self
 

	
 

	
 
class RelatedPostings(Sequence[data.Posting]):
 
    """Collect and query related postings
 

	
 
    This class provides common functionality for collecting related postings
 
    and running queries on them: iterating over them, tallying their balance,
 
    etc.
 

	
 
    This class doesn't know anything about how the postings are related. That's
 
    entirely up to the caller.
 

	
 
    A common pattern is to use this class with collections.defaultdict
 
    to organize postings based on some key. See the group_by_meta classmethod
 
    for an example.
 
    """
 
    __slots__ = ('_postings',)
 

	
 
    def __init__(self,
 
                 source: Iterable[data.Posting]=(),
 
                 *,
 
                 _can_own: bool=False,
 
    ) -> None:
 
        self._postings: List[data.Posting]
 
        if _can_own and isinstance(source, list):
 
            self._postings = source
 
        else:
 
            self._postings = list(source)
 

	
 
    @classmethod
 
    def _group_by(cls: Type[RelatedType],
 
                  postings: Iterable[data.Posting],
 
                  key: Callable[[data.Posting], T],
 
    ) -> Iterator[Tuple[T, RelatedType]]:
 
        mapping: Dict[T, List[data.Posting]] = collections.defaultdict(list)
 
        for post in postings:
 
            mapping[key(post)].append(post)
 
        for value, posts in mapping.items():
 
            yield value, cls(posts, _can_own=True)
 

	
 
    @classmethod
 
    def group_by_account(cls: Type[RelatedType],
 
                         postings: Iterable[data.Posting],
 
    ) -> Iterator[Tuple[data.Account, RelatedType]]:
 
        return cls._group_by(postings, operator.attrgetter('account'))
 

	
 
    @classmethod
 
    def group_by_meta(cls: Type[RelatedType],
 
                      postings: Iterable[data.Posting],
 
                      key: MetaKey,
 
                      default: Optional[MetaValue]=None,
 
    ) -> Iterator[Tuple[Optional[MetaValue], RelatedType]]:
 
        """Relate postings by metadata value
 

	
 
        This method takes an iterable of postings and returns a mapping.
 
        The keys of the mapping are the values of post.meta.get(key, default).
 
        The values are RelatedPostings instances that contain all the postings
 
        that had that same metadata value.
 
        """
 
        def key_func(post: data.Posting) -> Optional[MetaValue]:
 
            return post.meta.get(key, default)
 
        return cls._group_by(postings, key_func)
 

	
 
    @classmethod
 
    def group_by_first_meta_link(
 
            cls: Type[RelatedType],
 
            postings: Iterable[data.Posting],
 
            key: MetaKey,
 
    ) -> Iterator[Tuple[Optional[str], RelatedType]]:
 
        """Relate postings by the first link in metadata
 

	
 
        This method takes an iterable of postings and returns a mapping.
 
        The keys of the mapping are the values of
 
        post.meta.first_link(key, None).
 
        The values are RelatedPostings instances that contain all the postings
 
        that had that same first metadata link.
 
        """
 
        def key_func(post: data.Posting) -> Optional[MetaValue]:
 
            return post.meta.first_link(key, None)
 
        return cls._group_by(postings, key_func)
 

	
 
    def __repr__(self) -> str:
 
        return f'<{type(self).__name__} {self._postings!r}>'
 

	
 
    @overload
 
    def __getitem__(self: RelatedType, index: int) -> data.Posting: ...
 

	
 
    @overload
 
    def __getitem__(self: RelatedType, s: slice) -> RelatedType: ...
 

	
 
    def __getitem__(self: RelatedType,
 
                    index: Union[int, slice],
 
    ) -> Union[data.Posting, RelatedType]:
 
        if isinstance(index, slice):
 
            return type(self)(self._postings[index], _can_own=True)
 
        else:
 
            return self._postings[index]
 

	
 
    def __len__(self) -> int:
 
        return len(self._postings)
 

	
 
    def all_meta_links(self, key: MetaKey) -> Iterator[str]:
 
        return filters.iter_unique(
 
            link for post in self for link in post.meta.report_links(key)
 
        )
 

	
 
    @overload
 
    def first_meta_links(self, key: MetaKey, default: str='') -> Iterator[str]: ...
 

	
 
    @overload
 
    def first_meta_links(self, key: MetaKey, default: None) -> Iterator[Optional[str]]: ...
 

	
 
    def first_meta_links(self,
 
                         key: MetaKey,
 
                         default: Optional[str]='',
 
    ) -> Iterator[Optional[str]]:
 
        retval = filters.iter_unique(
 
            post.meta.first_link(key, default) for post in self
 
        )
 
        if default == '':
 
            retval = (s for s in retval if s)
 
        return retval
 

	
 
    def iter_with_balance(self) -> Iterator[Tuple[data.Posting, Balance]]:
 
        balance = MutableBalance()
 
        for post in self:
 
            balance += post.units
 
            yield post, balance
 

	
 
    def balance(self) -> Balance:
 
        return Balance(post.units for post in self)
 

	
 
    def balance_at_cost(self) -> Balance:
 
        return Balance(post.at_cost() for post in self)
 

	
 
    def balance_at_cost_by_date(self, date: datetime.date) -> Balance:
 
        for index, post in enumerate(self):
 
            if post.meta.date >= date:
 
                break
 
        else:
 
            index += 1
 
        return Balance(post.at_cost() for post in self._postings[:index])
 

	
 
    def meta_values(self,
 
                    key: MetaKey,
 
                    default: Optional[MetaValue]=None,
 
    ) -> Set[Optional[MetaValue]]:
 
        return {post.meta.get(key, default) for post in self}
 

	
 

	
 
class PeriodPostings(RelatedPostings):
 
    """Postings filtered and balanced over a date range
 

	
 
    Create a subclass with ``PeriodPostings.with_start_date(date)``.
 
    Note that there is no explicit stop date. The expectation is that the
 
    caller has already filtered out posts past the stop date from the input.
 

	
 
    Instances of that subclass will have three Balance attributes:
 

	
 
    * ``start_bal`` is the balance at cost of postings to your start date
 
    * ``period_bal`` is the balance at cost of postings from your start date
 
    * ``stop_bal`` is the balance at cost of all postings
 

	
 
    Use this subclass when your report includes a lot of balances over time to
 
    help you get the math right.
 
    """
 
    __slots__ = (
 
        'begin_bal',
 
        'end_bal',
 
        'period_bal',
 
        'start_bal',
 
        'stop_bal',
 
    )
 
    START_DATE = datetime.date(datetime.MINYEAR, 1, 1)
 

	
 
    def __init__(self,
 
                 source: Iterable[data.Posting]=(),
 
                 *,
 
                 _can_own: bool=False,
 
    ) -> None:
 
        start_posts: List[data.Posting] = []
 
        period_posts: List[data.Posting] = []
 
        for post in source:
 
            if post.meta.date < self.START_DATE:
 
                start_posts.append(post)
 
            else:
 
                period_posts.append(post)
 
        super().__init__(period_posts, _can_own=True)
 
        self.start_bal = RelatedPostings(start_posts, _can_own=True).balance_at_cost()
 
        self.period_bal = self.balance_at_cost()
 
        self.stop_bal = self.start_bal + self.period_bal
 
        # Convenience aliases
 
        self.begin_bal = self.start_bal
 
        self.end_bal = self.stop_bal
 

	
 
    @classmethod
 
    def with_start_date(cls: Type[RelatedType], start_date: datetime.date) -> Type[RelatedType]:
 
        name = f'BalancePostings{start_date.strftime("%Y%m%d")}'
 
        return type(name, (cls,), {'START_DATE': start_date})
 

	
 

	
 
class BaseSpreadsheet(Generic[RT, ST], metaclass=abc.ABCMeta):
 
    """Abstract base class to help write spreadsheets
 

	
 
    This class provides the very core logic to write an arbitrary set of data
 
    rows to arbitrary output. It calls hooks when it starts writing the
 
    spreadsheet, starts a new "section" of rows, ends a section, and ends the
 
    spreadsheet.
 

	
 
    RT is the type of the input data rows. ST is the type of the section
 
    identifier that you create from each row. If you don't want to use the
 
    section logic at all, set ST to None and define section_key to return None.
 
    """
 

	
 
    @abc.abstractmethod
 
    def section_key(self, row: RT) -> ST:
 
        """Return the section a row belongs to
 

	
 
        Given a data row, this method should return some identifier for the
 
        "section" the row belongs to. The write method uses this to
 
        determine when to call start_section and end_section.
 

	
 
        If your spreadsheet doesn't need sections, define this to return None.
 
        """
 
        ...
 

	
 
    @abc.abstractmethod
 
    def write_row(self, row: RT) -> None:
 
        """Write a data row to the output spreadsheet
 

	
 
        This method is called once for each data row in the input.
 
        """
 
        ...
 

	
 
    # The next four methods are all called by the write method when the name
 
    # says. You may override them to output headers or sums, record
 
    # state, etc. The default implementations are all noops.
 

	
 
    def start_spreadsheet(self) -> None:
 
        pass
 

	
 
    def start_section(self, key: ST) -> None:
 
        pass
 

	
 
    def end_section(self, key: ST) -> None:
 
        pass
 

	
 
    def end_spreadsheet(self) -> None:
 
        pass
 

	
 
    def write(self, rows: Iterable[RT]) -> None:
 
        prev_section: Optional[ST] = None
 
        self.start_spreadsheet()
 
        for row in rows:
 
            section = self.section_key(row)
 
            if section != prev_section:
 
                if prev_section is not None:
 
                    self.end_section(prev_section)
 
                self.start_section(section)
 
                prev_section = section
 
            self.write_row(row)
 
        try:
 
            should_end = section is not None
 
        except NameError:
 
            should_end = False
 
        if should_end:
 
            self.end_section(section)
 
        self.end_spreadsheet()
 

	
 

	
 
class BaseODS(BaseSpreadsheet[RT, ST], metaclass=abc.ABCMeta):
 
    """Abstract base class to help write OpenDocument spreadsheets
 

	
 
    This class provides the very core logic to write an arbitrary set of data
 
    rows to an OpenDocument spreadsheet. It provides helper methods for
 
    building sheets, rows, and cells.
 

	
 
    See also the BaseSpreadsheet base class for additional documentation about
 
    methods you must and can define, the definition of RT and ST, etc.
 
    """
 
    # Defined in the XSL spec, "Definitions of Units of Measure"
 
    MEASUREMENT_UNITS = frozenset([
 
        'cm',
 
        'em',
 
        'in',
 
        'mm',
 
        'pc',
 
        'pt',
 
        'px',
 
    ])
 
    MEASUREMENT_RE = re.compile(
 
        r'([-+]?(?:\d+\.?|\.\d+|\d+\.\d+))({})'.format('|'.join(MEASUREMENT_UNITS)),
 
        re.ASCII,
 
    )
 

	
 
    def __init__(self, rt_wrapper: Optional[rtutil.RT]=None) -> None:
 
        self.rt_wrapper = rt_wrapper
 
        self.locale = babel.core.Locale.default('LC_MONETARY')
 
        self.currency_fmt_key = 'accounting'
 
        self._name_counter = itertools.count(1)
 
        self._currency_style_cache: MutableMapping[str, odf.style.Style] = {}
 
        self.document = odf.opendocument.OpenDocumentSpreadsheet()
 
        self.init_settings()
 
        self.init_styles()
 
        self.sheet = self.use_sheet("Report")
 

	
 
    ### Low-level document tree manipulation
 
    # The *intent* is that you only need to use these if you're adding new
 
    # methods to manipulate document settings or styles.
 

	
 
    def copy_element(self, elem: odf.element.Element) -> odf.element.Element:
 
        qattrs = dict(self.iter_qattributes(elem))
 
        retval = odf.element.Element(qname=elem.qname, qattributes=qattrs)
 
        try:
 
            orig_name = retval.getAttribute('name')
 
        except ValueError:
 
            orig_name = None
 
        if orig_name is not None:
 
            retval.setAttribute('name', f'{orig_name}{next(self._name_counter)}')
 
        return retval
 

	
 
    def ensure_child(self,
 
                     parent: odf.element.Element,
 
                     child_type: ElementType,
 
                     **kwargs: Any,
 
    ) -> odf.element.Element:
 
        new_child = child_type(**kwargs)
 
        found_child = self.find_child(parent, new_child)
 
        if found_child is None:
 
            parent.addElement(new_child)
 
            return parent.lastChild
 
        else:
 
            return found_child
 

	
 
    def ensure_config_map_entry(self,
 
                                root: odf.element.Element,
 
                                map_name: str,
 
                                entry_name: str,
 
    ) -> odf.element.Element:
 
        """Return a ``ConfigItemMapEntry`` under ``root``
 

	
 
        This method ensures there's a ``ConfigItemMapNamed`` named ``map_name``
 
        under ``root``, and a ``ConfigItemMapEntry`` named ``entry_name`` under
 
        that. Return the ``ConfigItemMapEntry`` element.
 
        """
 
        config_map = self.ensure_child(root, odf.config.ConfigItemMapNamed, name=map_name)
 
        return self.ensure_child(config_map, odf.config.ConfigItemMapEntry, name=entry_name)
 

	
 
    def find_child(self,
 
                   parent: odf.element.Element,
 
                   child: odf.element.Element,
 
    ) -> Optional[odf.element.Element]:
 
        attrs = {k: v for k, v in self.iter_attributes(child)}
 
        if not attrs:
 
            return None
 
        for elem in parent.childNodes:
 
            if (elem.qname == child.qname
 
                and all(elem.getAttribute(k) == v for k, v in attrs.items())):
 
                return elem
 
        return None
 

	
 
    def iter_attributes(self, elem: odf.element.Element) -> Iterator[Tuple[str, str]]:
 
        for (_, key), value in self.iter_qattributes(elem):
 
            yield key.lower().replace('-', ''), value
 

	
 
    def iter_qattributes(self, elem: odf.element.Element) -> Iterator[Tuple[Tuple[str, str], str]]:
 
        if elem.attributes:
 
            yield from elem.attributes.items()
 

	
 
    def replace_child(self,
 
                     parent: odf.element.Element,
 
                     child_type: ElementType,
 
                     **kwargs: Any,
 
    ) -> odf.element.Element:
 
        new_child = child_type(**kwargs)
 
        found_child = self.find_child(parent, new_child)
 
        parent.insertBefore(new_child, found_child)
 
        if found_child is not None:
 
            parent.removeChild(found_child)
 
        return new_child
 

	
 
    def set_config(self,
 
                   root: odf.element.Element,
 
                   name: str,
 
                   value: Union[bool, int, str],
 
                   config_type: Optional[str]=None,
 
    ) -> None:
 
        """Ensure ``root`` has a ``ConfigItem`` with the given name, type, and value"""
 
        value_s = str(value)
 
        if isinstance(value, bool):
 
            value_s = str(value).lower()
 
            default_type = 'boolean'
 
        elif isinstance(value, str):
conservancy_beancount/reports/ledger.py
Show inline comments
 
"""ledger.py - General ledger report from Beancount
 

	
 
This tool produces a spreadsheet that shows postings in Beancount, organized
 
by account.
 

	
 
Specify the date range you want to report with the ``--begin`` and ``--end``
 
options.
 

	
 
Select the accounts you want to report with the ``--account`` option. You can
 
specify this option multiple times. The report will include at least one sheet
 
for each account you specify. Subaccounts will be reported on that sheet as
 
well.
 

	
 
Select the postings you want to report by passing metadata search terms in
 
``name=value`` format.
 

	
 
Run ``ledger-report --help`` for abbreviations and other options.
 

	
 
Examples
 
--------
 

	
 
Report all activity related to a given project::
 

	
 
    ledger-report project=NAME
 

	
 
Get all Assets postings for a given month to help with reconciliation::
 

	
 
    ledger-report -a Assets -b 2018-05-01 -e 2018-06-01
 
"""
 
# 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 argparse
 
import collections
 
import datetime
 
import enum
 
import itertools
 
import operator
 
import logging
 
import sys
 

	
 
from typing import (
 
    Callable,
 
    Dict,
 
    Iterable,
 
    Iterator,
 
    List,
 
    Mapping,
 
    Optional,
 
    Sequence,
 
    TextIO,
 
    Tuple,
 
    Union,
 
)
 

	
 
from pathlib import Path
 

	
 
import odf.table  # type:ignore[import]
 

	
 
from beancount.core import data as bc_data
 
from beancount.parser import printer as bc_printer
 

	
 
from . import core
 
from .. import books
 
from .. import cliutil
 
from .. import config as configmod
 
from .. import data
 
from .. import ranges
 
from .. import rtutil
 

	
 
PostTally = List[Tuple[int, data.Account]]
 

	
 
PROGNAME = 'ledger-report'
 
logger = logging.getLogger('conservancy_beancount.reports.ledger')
 

	
 
class AccountPostings(core.RelatedPostings):
 
    START_DATE: datetime.date
 

	
 
    def __init__(self,
 
                 source: Iterable[data.Posting]=(),
 
                 *,
 
                 _can_own: bool=False,
 
    ) -> None:
 
        super().__init__(source, _can_own=_can_own)
 
        self.start_bal = self.balance_at_cost_by_date(self.START_DATE)
 
        self.stop_bal = self.balance_at_cost()
 
        self.period_bal = self.stop_bal - self.start_bal
 

	
 

	
 
class LedgerODS(core.BaseODS[data.Posting, data.Account]):
 
    CORE_COLUMNS: Sequence[str] = [
 
        'Date',
 
        data.Metadata.human_name('entity'),
 
        'Description',
 
        'Original Amount',
 
        'Booked Amount',
 
    ]
 
    ACCOUNT_COLUMNS: Dict[str, Sequence[str]] = collections.OrderedDict([
 
        ('Income', ['project', 'rt-id', 'receipt', 'income-type']),
 
        ('Expenses', ['project', 'rt-id', 'receipt', 'approval', 'expense-allocation']),
 
        ('Equity', ['rt-id']),
 
        ('Assets:Receivable', ['project', 'rt-id', 'invoice', 'approval', 'contract', 'purchase-order']),
 
        ('Liabilities:Payable', ['project', 'rt-id', 'invoice', 'approval', 'contract', 'purchase-order']),
 
        ('Assets:PayPal', ['rt-id', 'paypal-id', 'receipt', 'approval']),
 
        ('Assets', ['rt-id', 'receipt', 'approval', 'bank-statement']),
 
        ('Liabilities', ['rt-id', 'receipt', 'approval', 'bank-statement']),
 
    ])
 
    # Excel 2003 was limited to 65,536 rows per worksheet.
 
    # While we can probably count on all our users supporting more modern
 
    # formats (Excel 2007 supports over 1 million rows per worksheet),
 
    # keeping the default limit conservative seems good to avoid running into
 
    # other limits (like the number of hyperlinks per worksheet), plus just
 
    # better for human organization and readability.
 
    SHEET_SIZE = 65000
 

	
 
    def __init__(self,
 
                 start_date: datetime.date,
 
                 stop_date: datetime.date,
 
                 sheet_names: Optional[Sequence[str]]=None,
 
                 rt_wrapper: Optional[rtutil.RT]=None,
 
                 sheet_size: Optional[int]=None,
 
    ) -> None:
 
        if sheet_names is None:
 
            sheet_names = list(self.ACCOUNT_COLUMNS)
 
        if sheet_size is None:
 
            sheet_size = self.SHEET_SIZE
 
        super().__init__(rt_wrapper)
 
        self.date_range = ranges.DateRange(start_date, stop_date)
 
        self.required_sheet_names = sheet_names
 
        self.sheet_size = sheet_size
 

	
 
    def init_styles(self) -> None:
 
        super().init_styles()
 
        self.amount_column = self.column_style(1.2)
 
        self.default_column = self.column_style(1.5)
 
        self.column_styles: Mapping[str, Union[str, odf.style.Style]] = {
 
            'Date': '',
 
            'Description': self.column_style(2),
 
            'Original Amount': self.amount_column,
 
            'Booked Amount': self.amount_column,
 
            data.Metadata.human_name('project'): self.amount_column,
 
            data.Metadata.human_name('rt-id'): self.amount_column,
 
        }
 

	
 
    @classmethod
 
    def _group_tally(
 
            cls,
 
            tally_by_account: PostTally,
 
            key: Callable[[data.Account], Optional[str]],
 
    ) -> Dict[str, PostTally]:
 
        retval: Dict[str, PostTally] = collections.defaultdict(list)
 
        for count, account in tally_by_account:
 
            item_key = key(account)
 
            if item_key is not None:
 
                retval[item_key].append((count, account))
 
        return retval
 

	
 
    @classmethod
 
    def _split_sheet(
 
            cls,
 
            tally_by_account: PostTally,
 
            sheet_size: int,
 
            sheet_name: str,
 
    ) -> Iterator[str]:
 
        total = 0
 
        for index, (count, account) in enumerate(tally_by_account):
 
            total += count
 
            if total > sheet_size:
 
                break
 
        else:
 
            # All the accounts fit in this sheet.
 
            yield sheet_name
 
            return
 
        if index == 0 and len(tally_by_account) == 1:
 
            # With one account, we can't split any further, so warn and stop.
 
            logger.warning(
 
                "%s has %s rows, over size %s",
 
                account, f'{count:,g}', f'{sheet_size:,g}',
 
            )
 
            yield sheet_name
 
            return
 
        group_func = operator.methodcaller('root_part', sheet_name.count(':') + 2)
 
        maybe_split = cls._group_tally(tally_by_account[:index], group_func)
 
        must_split = cls._group_tally(tally_by_account[index:], group_func)
 
        for subkey, must_split_tally in sorted(must_split.items()):
 
            split_names = cls._split_sheet(
 
                maybe_split.get(subkey, []) + must_split_tally, sheet_size, subkey,
 
            )
 
            # We must be willing to split out at least as many sheets as there
 
            # are accounts that didn't fit. Do that first.
 
            yield from itertools.islice(split_names, len(must_split_tally))
 
            # After that, we can be in one of two cases:
 
            # 1. There is no next sheet. All the accounts, including the
 
            #    maybe_splits and must_splits, fit on planned subsheets.
 
            #    Update state to note we don't need a sheet for them anymore.
 
            # 2. The next sheet is named `subkey`, and is planned to include
 
            #    all of our maybe_split accounts. However, we don't need to
 
            #    yield that sheet name, because those accounts already fit in
 
            #    the sheet we're planning, and it would be a needless split.
 
            next_sheet_name = next(split_names, None)
 
            if next_sheet_name is None:
 
                maybe_split.pop(subkey, None)
 
            else:
 
                assert next_sheet_name == subkey
 
                assert not any(split_names)
 
        if maybe_split:
 
            yield sheet_name
 

	
 
    @classmethod
 
    def plan_sheets(
 
            cls,
 
            tally_by_account: Mapping[data.Account, int],
 
            base_sheets: Sequence[str],
 
            sheet_size: int,
 
    ) -> Sequence[str]:
 
        sorted_tally: PostTally = [
 
            (count, account)
 
            for account, count in tally_by_account.items()
 
        ]
 
        sorted_tally.sort()
 
        split_tally = cls._group_tally(
 
            sorted_tally,
 
            operator.methodcaller('is_under', *base_sheets),
 
        )
 
        return [
 
            sheet_name
 
            for key in base_sheets
 
            for sheet_name in cls._split_sheet(split_tally[key], sheet_size, key)
 
        ]
 

	
 
    def section_key(self, row: data.Posting) -> data.Account:
 
        return row.account
 

	
 
    def start_sheet(self, sheet_name: str) -> None:
 
        self.use_sheet(sheet_name.replace(':', ' '))
 
        columns_key = data.Account(sheet_name).is_under(*self.ACCOUNT_COLUMNS)
 
        # columns_key must not be None because ACCOUNT_COLUMNS has an entry
 
        # for all five root accounts.
 
        assert columns_key is not None
 
        self.metadata_columns = self.ACCOUNT_COLUMNS[columns_key]
 
        self.sheet_columns: Sequence[str] = [
 
            *self.CORE_COLUMNS,
 
            *(data.Metadata.human_name(meta_key) for meta_key in self.metadata_columns),
 
        ]
 
        for col_name in self.sheet_columns:
 
            self.sheet.addElement(odf.table.TableColumn(
 
                stylename=self.column_styles.get(col_name, self.default_column),
 
            ))
 
        self.add_row(*(
 
            self.string_cell(col_name, stylename=self.style_bold)
 
            for col_name in self.sheet_columns
 
        ))
 
        self.lock_first_row()
 

	
 
    def _report_section_balance(self, key: data.Account, date_key: str) -> None:
 
        uses_opening = key.is_under('Assets', 'Equity', 'Liabilities')
 
        related = self.account_groups[key]
 
        if date_key == 'start':
 
            if not uses_opening:
 
                return
 
            date = self.date_range.start
 
            balance = related.start_bal
 
            description = "Opening Balance"
 
        else:
 
            date = self.date_range.stop
 
            if uses_opening:
 
                balance = related.stop_bal
 
                description = "Ending Balance"
 
            else:
 
                balance = related.period_bal
 
                description = "Period Total"
 
        self.add_row(
 
            self.date_cell(date, stylename=self.merge_styles(
 
                self.style_bold, self.style_date,
 
            )),
 
            odf.table.TableCell(),
 
            self.string_cell(description, stylename=self.style_bold),
 
            odf.table.TableCell(),
 
            self.balance_cell(self.norm_func(balance), stylename=self.style_bold),
 
        )
 

	
 
    def start_section(self, key: data.Account) -> None:
 
        self.add_row()
 
        self.add_row(
 
            odf.table.TableCell(),
 
            self.string_cell(
 
                f"{key} Ledger"
 
                f" From {self.date_range.start.isoformat()}"
 
                f" To {self.date_range.stop.isoformat()}",
 
                stylename=self.style_bold,
 
                numbercolumnsspanned=len(self.sheet_columns) - 1,
 
            ),
 
        )
 
        self.norm_func = core.normalize_amount_func(key)
 
        self._report_section_balance(key, 'start')
 

	
 
    def end_section(self, key: data.Account) -> None:
 
        self._report_section_balance(key, 'stop')
 

	
 
    def write_row(self, row: data.Posting) -> None:
 
        if row.meta.date not in self.date_range:
 
            return
 
        elif row.cost is None:
 
        if row.cost is None:
 
            amount_cell = odf.table.TableCell()
 
        else:
 
            amount_cell = self.currency_cell(self.norm_func(row.units))
 
        self.add_row(
 
            self.date_cell(row.meta.date),
 
            self.string_cell(row.meta.get('entity') or ''),
 
            self.string_cell(row.meta.txn.narration),
 
            amount_cell,
 
            self.currency_cell(self.norm_func(row.at_cost())),
 
            *(self.meta_links_cell(row.meta.report_links(key))
 
              if key in data.LINK_METADATA
 
              else self.string_cell(row.meta.get(key, ''))
 
              for key in self.metadata_columns),
 
        )
 

	
 
    def _combined_balance_row(self,
 
                              balance_accounts: Sequence[str],
 
                              attr_name: str,
 
    ) -> None:
 
        date = getattr(self.date_range, attr_name)
 
        balance_attrname = f'{attr_name}_bal'
 
        balance = -sum((
 
            getattr(related, balance_attrname)
 
            for account, related in self.account_groups.items()
 
            if account.is_under(*balance_accounts)
 
        ), core.MutableBalance())
 
        self.add_row(
 
            self.string_cell(
 
                f"Balance as of {date.isoformat()}",
 
                stylename=self.merge_styles(self.style_bold, self.style_endtext),
 
            ),
 
            self.balance_cell(balance, stylename=self.style_bold),
 
        )
 

	
 
    def write_balance_sheet(self) -> None:
 
        balance_accounts = ['Equity', 'Income', 'Expenses']
 
        self.use_sheet("Balance")
 
        self.sheet.addElement(odf.table.TableColumn(stylename=self.column_style(3)))
 
        self.sheet.addElement(odf.table.TableColumn(stylename=self.column_style(1.5)))
 
        self.add_row(
 
            self.string_cell("Account", stylename=self.style_bold),
 
            self.string_cell("Balance", stylename=self.style_bold),
 
        )
 
        self.lock_first_row()
 
        self.add_row()
 
        self.add_row(self.string_cell(
 
            f"Ledger From {self.date_range.start.isoformat()}"
 
            f" To {self.date_range.stop.isoformat()}",
 
            stylename=self.merge_styles(self.style_centertext, self.style_bold),
 
            numbercolumnsspanned=2,
 
        ))
 
        self.add_row()
 
        self._combined_balance_row(balance_accounts, 'start')
 
        for _, account in core.sort_and_filter_accounts(
 
                self.account_groups, balance_accounts,
 
        ):
 
            balance = self.account_groups[account].period_bal
 
            if not balance.is_zero():
 
                self.add_row(
 
                    self.string_cell(account, stylename=self.style_endtext),
 
                    self.balance_cell(-balance),
 
                )
 
        self._combined_balance_row(balance_accounts, 'stop')
 

	
 
    def write(self, rows: Iterable[data.Posting]) -> None:
 
        AccountPostings.START_DATE = self.date_range.start
 
        self.account_groups = dict(AccountPostings.group_by_account(
 
        related_cls = core.PeriodPostings.with_start_date(self.date_range.start)
 
        self.account_groups = dict(related_cls.group_by_account(
 
            post for post in rows if post.meta.date < self.date_range.stop
 
        ))
 
        self.write_balance_sheet()
 
        tally_by_account_iter = (
 
            (account, sum(1 for post in related if post.meta.date in self.date_range))
 
            (account, len(related))
 
            for account, related in self.account_groups.items()
 
        )
 
        tally_by_account = {
 
            account: count
 
            for account, count in tally_by_account_iter
 
            if count
 
        }
 
        sheet_names = self.plan_sheets(
 
            tally_by_account, self.required_sheet_names, self.sheet_size,
 
        )
 
        using_sheet_index = -1
 
        for sheet_index, account in core.sort_and_filter_accounts(
 
                tally_by_account, sheet_names,
 
        ):
 
            while using_sheet_index < sheet_index:
 
                using_sheet_index += 1
 
                self.start_sheet(sheet_names[using_sheet_index])
 
            super().write(self.account_groups[account])
 
        for index in range(using_sheet_index + 1, len(sheet_names)):
 
            self.start_sheet(sheet_names[index])
 

	
 

	
 
class ReturnFlag(enum.IntFlag):
 
    LOAD_ERRORS = 1
 
    NOTHING_TO_REPORT = 8
 

	
 

	
 
def parse_arguments(arglist: Optional[Sequence[str]]=None) -> argparse.Namespace:
 
    parser = argparse.ArgumentParser(prog=PROGNAME)
 
    cliutil.add_version_argument(parser)
 
    parser.add_argument(
 
        '--begin', '--start', '-b',
 
        dest='start_date',
 
        metavar='DATE',
 
        type=cliutil.date_arg,
 
        help="""Date to start reporting entries, inclusive, in YYYY-MM-DD format.
 
The default is one year ago.
 
""")
 
    parser.add_argument(
 
        '--end', '--stop', '-e',
 
        dest='stop_date',
 
        metavar='DATE',
 
        type=cliutil.date_arg,
 
        help="""Date to stop reporting entries, exclusive, in YYYY-MM-DD format.
 
The default is a year after the start date, or 30 days from today if the start
 
date was also not specified.
 
""")
 
    parser.add_argument(
 
        '--account', '-a',
 
        dest='sheet_names',
 
        metavar='ACCOUNT',
 
        action='append',
 
        help="""Show this account in the report. You can specify this option
 
multiple times. If not specified, the default set adapts to your search
 
criteria.
 
""")
 
    parser.add_argument(
 
        '--sheet-size', '--size',
 
        metavar='SIZE',
 
        type=int,
 
        default=LedgerODS.SHEET_SIZE,
 
        help="""Try to limit sheets to this many rows. The report will
 
automatically create new sheets to make this happen. When that's not possible,
 
it will issue a warning.
 
""")
 
    parser.add_argument(
 
        '--output-file', '-O',
 
        metavar='PATH',
 
        type=Path,
 
        help="""Write the report to this file, or stdout when PATH is `-`.
 
The default is stdout for the balance and outgoing reports, and a generated
 
filename for other reports.
 
""")
 
    cliutil.add_loglevel_argument(parser)
 
    parser.add_argument(
 
        'search_terms',
 
        metavar='FILTER',
 
        type=cliutil.SearchTerm.arg_parser('project', 'rt-id'),
 
        nargs=argparse.ZERO_OR_MORE,
 
        help="""Report on postings that match this criteria. The format is
 
NAME=TERM. TERM is a link or word that must exist in a posting's NAME
 
metadata to match. A single ticket number is a shortcut for
 
`rt-id=rt:NUMBER`. Any other word is a shortcut for `project=TERM`.
 
""")
 
    args = parser.parse_args(arglist)
 
    if args.sheet_names is None:
 
        if any(term.meta_key == 'project' for term in args.search_terms):
 
            args.sheet_names = ['Income', 'Expenses', 'Assets:Receivable', 'Liabilities:Payable']
 
        else:
 
            args.sheet_names = list(LedgerODS.ACCOUNT_COLUMNS)
 
    return args
 

	
 
def diff_year(date: datetime.date, diff: int) -> datetime.date:
 
    new_year = date.year + diff
 
    try:
 
        return date.replace(year=new_year)
 
    except ValueError:
 
        # The original date is Feb 29, which doesn't exist in the new year.
 
        if diff < 0:
 
            return datetime.date(new_year, 2, 28)
 
        else:
 
            return datetime.date(new_year, 3, 1)
 

	
 
def main(arglist: Optional[Sequence[str]]=None,
 
         stdout: TextIO=sys.stdout,
 
         stderr: TextIO=sys.stderr,
 
         config: Optional[configmod.Config]=None,
 
) -> int:
 
    args = parse_arguments(arglist)
 
    cliutil.set_loglevel(logger, args.loglevel)
 
    if config is None:
 
        config = configmod.Config()
 
        config.load_file()
 

	
 
    today = datetime.date.today()
 
    if args.start_date is None:
 
        args.start_date = diff_year(today, -1)
 
        if args.stop_date is None:
 
            args.stop_date = today + datetime.timedelta(days=30)
 
    elif args.stop_date is None:
 
        args.stop_date = diff_year(args.start_date, 1)
 

	
 
    returncode = 0
 
    books_loader = config.books_loader()
 
    if books_loader is None:
 
        entries, load_errors, _ = books.Loader.load_none(config.config_file_path())
 
    else:
 
        entries, load_errors, _ = books_loader.load_fy_range(args.start_date, args.stop_date)
 
    for error in load_errors:
 
        bc_printer.print_error(error, file=stderr)
 
        returncode |= ReturnFlag.LOAD_ERRORS
 

	
 
    postings = data.Posting.from_entries(entries)
 
    for search_term in args.search_terms:
 
        postings = search_term.filter_postings(postings)
 

	
 
    rt_wrapper = config.rt_wrapper()
 
    if rt_wrapper is None:
 
        logger.warning("could not initialize RT client; spreadsheet links will be broken")
 
    report = LedgerODS(
 
        args.start_date,
 
        args.stop_date,
 
        args.sheet_names,
 
        rt_wrapper,
 
        args.sheet_size,
 
    )
 
    report.write(postings)
 
    if not report.account_groups:
 
        logger.warning("no matching postings found to report")
 
        returncode |= ReturnFlag.NOTHING_TO_REPORT
 

	
 
    if args.output_file is None:
 
        out_dir_path = config.repository_path() or Path()
 
        args.output_file = out_dir_path / 'LedgerReport_{}_{}.ods'.format(
 
            args.start_date.isoformat(), args.stop_date.isoformat(),
 
        )
 
        logger.info("Writing report to %s", args.output_file)
 
    ods_file = cliutil.bytes_output(args.output_file, stdout)
 
    report.save_file(ods_file)
 
    return 0 if returncode == 0 else 16 + returncode
 

	
 
entry_point = cliutil.make_entry_point(__name__, PROGNAME)
 

	
 
if __name__ == '__main__':
 
    exit(entry_point())
tests/test_reports_period_postings.py
Show inline comments
 
new file 100644
 
"""test_reports_period_postings - Unit tests for PeriodPostings"""
 
# 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 datetime
 
import operator
 

	
 
from decimal import Decimal
 

	
 
import pytest
 

	
 
from . import testutil
 

	
 
from conservancy_beancount import data
 
from conservancy_beancount.reports import core
 

	
 
def check_balance_attrs(balance_postings, start_usd, stop_usd):
 
    if start_usd:
 
        expected = {'USD': testutil.Amount(start_usd)}
 
        assert balance_postings.start_bal == expected
 
        assert balance_postings.begin_bal == expected
 
    else:
 
        assert balance_postings.start_bal.is_zero()
 
        assert balance_postings.begin_bal.is_zero()
 
    expected = {'USD': testutil.Amount(stop_usd)}
 
    assert balance_postings.stop_bal == expected
 
    assert balance_postings.end_bal == expected
 
    expected = {'USD': testutil.Amount(stop_usd - start_usd)}
 
    assert balance_postings.period_bal == expected
 

	
 
@pytest.mark.parametrize('start_date,expect_start_bal', [
 
    (datetime.date(2019, 2, 1), 0),
 
    (datetime.date(2019, 4, 1), 30),
 
    (datetime.date(2019, 6, 1), 120),
 
])
 
def test_balance_postings_attrs(start_date, expect_start_bal):
 
    entries = [testutil.Transaction(date=datetime.date(2019, n, 15), postings=[
 
        ('Income:Donations', -n * 10),
 
        ('Assets:Cash', n * 10),
 
    ]) for n in range(3, 7)]
 
    cls = core.PeriodPostings.with_start_date(start_date)
 
    actual = dict(cls.group_by_account(data.Posting.from_entries(entries)))
 
    assert len(actual) == 2
 
    check_balance_attrs(actual['Assets:Cash'], expect_start_bal, 180)
 
    check_balance_attrs(actual['Income:Donations'], -expect_start_bal, -180)
 

	
 
@pytest.mark.parametrize('start_date,expect_count', [
 
    (datetime.date(2019, 2, 1), 4),
 
    (datetime.date(2019, 4, 1), 3),
 
    (datetime.date(2019, 6, 1), 1),
 
])
 
def test_balance_postings_filter(start_date, expect_count):
 
    entries = [testutil.Transaction(date=datetime.date(2019, n, 15), postings=[
 
        ('Income:Donations', -n * 10),
 
        ('Assets:Cash', n * 10),
 
    ]) for n in range(3, 7)]
 
    cls = core.PeriodPostings.with_start_date(start_date)
 
    for _, related in cls.group_by_account(data.Posting.from_entries(entries)):
 
        assert len(related) == expect_count
0 comments (0 inline, 0 general)