ledger: Prepare LedgerODS for subclasses.

This commit reorganized the class internals to make it more straightforward
to add a transaction-oriented reporting subclass.
This commit is contained in:
Brett Smith 2020-07-21 10:42:52 -04:00
parent 56114cc66e
commit 6960425571

View file

@ -86,7 +86,7 @@ PostTally = List[Tuple[int, data.Account]]
PROGNAME = 'ledger-report' PROGNAME = 'ledger-report'
logger = logging.getLogger('conservancy_beancount.reports.ledger') logger = logging.getLogger('conservancy_beancount.reports.ledger')
class LedgerODS(core.BaseODS[data.Posting, data.Account]): class LedgerODS(core.BaseODS[data.Posting, None]):
CORE_COLUMNS: Sequence[str] = [ CORE_COLUMNS: Sequence[str] = [
'Date', 'Date',
data.Metadata.human_name('entity'), data.Metadata.human_name('entity'),
@ -270,16 +270,21 @@ class LedgerODS(core.BaseODS[data.Posting, data.Account]):
for sheet_name in cls._split_sheet(split_tally[key], sheet_size, key) for sheet_name in cls._split_sheet(split_tally[key], sheet_size, key)
] ]
def section_key(self, row: data.Posting) -> data.Account: # The write() override does its own sectioning by account, so this method
return row.account # goes unused.
def section_key(self, row: data.Posting) -> None:
return None
def start_sheet(self, sheet_name: str) -> None: def metadata_columns_for(self, sheet_name: str) -> Sequence[str]:
self.use_sheet(sheet_name.replace(':', ' '))
columns_key = data.Account(sheet_name).is_under(*self.ACCOUNT_COLUMNS) columns_key = data.Account(sheet_name).is_under(*self.ACCOUNT_COLUMNS)
# columns_key must not be None because ACCOUNT_COLUMNS has an entry # columns_key must not be None because ACCOUNT_COLUMNS has an entry
# for all five root accounts. # for all five root accounts.
assert columns_key is not None assert columns_key is not None
self.metadata_columns = self.ACCOUNT_COLUMNS[columns_key] return self.ACCOUNT_COLUMNS[columns_key]
def start_sheet(self, sheet_name: str) -> None:
self.use_sheet(sheet_name.replace(':', ' '))
self.metadata_columns = self.metadata_columns_for(sheet_name)
self.sheet_columns: Sequence[str] = [ self.sheet_columns: Sequence[str] = [
*self.CORE_COLUMNS, *self.CORE_COLUMNS,
*(data.Metadata.human_name(meta_key) for meta_key in self.metadata_columns), *(data.Metadata.human_name(meta_key) for meta_key in self.metadata_columns),
@ -320,7 +325,7 @@ class LedgerODS(core.BaseODS[data.Posting, data.Account]):
self.balance_cell(self.norm_func(balance), stylename=self.style_bold), self.balance_cell(self.norm_func(balance), stylename=self.style_bold),
) )
def start_section(self, key: data.Account, *, force_total: bool=False) -> None: def write_header(self, key: data.Account) -> None:
self.add_row() self.add_row()
self.add_row( self.add_row(
odf.table.TableCell(), odf.table.TableCell(),
@ -332,29 +337,6 @@ class LedgerODS(core.BaseODS[data.Posting, data.Account]):
numbercolumnsspanned=len(self.sheet_columns) - 1, numbercolumnsspanned=len(self.sheet_columns) - 1,
), ),
) )
self.norm_func = core.normalize_amount_func(key)
if force_total or key.is_under(*self.totals_with_entries):
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.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 write_balance_sheet(self) -> None: def write_balance_sheet(self) -> None:
self.use_sheet("Balance") self.use_sheet("Balance")
@ -388,6 +370,27 @@ class LedgerODS(core.BaseODS[data.Posting, data.Account]):
self.balance_cell(-balance, stylename=style), self.balance_cell(-balance, stylename=style),
) )
def _account_tally(self, account: data.Account) -> int:
return len(self.account_groups[account])
def write_entries(self, account: data.Account, rows: Iterable[data.Posting]) -> None:
for row in rows:
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 write(self, rows: Iterable[data.Posting]) -> None: def write(self, rows: Iterable[data.Posting]) -> None:
related_cls = core.PeriodPostings.with_start_date(self.date_range.start) related_cls = core.PeriodPostings.with_start_date(self.date_range.start)
self.account_groups = dict(related_cls.group_by_account( self.account_groups = dict(related_cls.group_by_account(
@ -397,7 +400,7 @@ class LedgerODS(core.BaseODS[data.Posting, data.Account]):
self.account_groups[empty_acct] = related_cls() self.account_groups[empty_acct] = related_cls()
self.write_balance_sheet() self.write_balance_sheet()
tally_by_account_iter = ( tally_by_account_iter = (
(account, len(self.account_groups[account])) (account, self._account_tally(account))
for account in self.accounts for account in self.accounts
) )
tally_by_account = { tally_by_account = {
@ -412,17 +415,25 @@ class LedgerODS(core.BaseODS[data.Posting, data.Account]):
for sheet_index, account in core.sort_and_filter_accounts( for sheet_index, account in core.sort_and_filter_accounts(
tally_by_account, sheet_names, tally_by_account, sheet_names,
): ):
if not account.is_open_on_date(self.date_range.start):
continue
while using_sheet_index < sheet_index: while using_sheet_index < sheet_index:
using_sheet_index += 1 using_sheet_index += 1
self.start_sheet(sheet_names[using_sheet_index]) self.start_sheet(sheet_names[using_sheet_index])
self.norm_func = core.normalize_amount_func(account)
postings = self.account_groups[account] postings = self.account_groups[account]
if postings: if postings:
super().write(postings) totals_set = self.totals_with_entries
elif not account.is_open_on_date(self.date_range.start): else:
pass totals_set = self.totals_without_entries
elif account.is_under(*self.totals_without_entries): want_totals = account.is_under(*totals_set) is not None
self.start_section(account, force_total=True) if postings or want_totals:
self.end_section(account) self.write_header(account)
if want_totals:
self._report_section_balance(account, 'start')
self.write_entries(account, postings)
if want_totals:
self._report_section_balance(account, 'stop')
for index in range(using_sheet_index + 1, len(sheet_names)): for index in range(using_sheet_index + 1, len(sheet_names)):
self.start_sheet(sheet_names[index]) self.start_sheet(sheet_names[index])