Changeset - ceb0c451fade
[Not reviewed]
0 1 0
Brett Smith - 4 years ago 2020-07-01 19:57:58
brettcsmith@brettcsmith.org
fund: Improve type signature.
1 file changed with 1 insertions and 1 deletions:
0 comments (0 inline, 0 general)
conservancy_beancount/reports/fund.py
Show inline comments
...
 
@@ -80,193 +80,193 @@ from .. import cliutil
 
from .. import config as configmod
 
from .. import data
 

	
 
AccountsMap = Mapping[data.Account, core.PeriodPostings]
 
FundPosts = Tuple[MetaValue, AccountsMap]
 

	
 
EQUITY_ACCOUNTS = ['Equity', 'Income', 'Expenses']
 
INFO_ACCOUNTS = [
 
    'Assets:Receivable',
 
    'Assets:Prepaid',
 
    'Liabilities:Payable',
 
    'Liabilities:UnearnedIncome',
 
]
 
PROGNAME = 'fund-report'
 
UNRESTRICTED_FUND = 'Conservancy'
 
logger = logging.getLogger('conservancy_beancount.reports.fund')
 

	
 
class ODSReport(core.BaseODS[FundPosts, None]):
 
    def __init__(self, start_date: datetime.date, stop_date: datetime.date) -> None:
 
        super().__init__()
 
        self.start_date = start_date
 
        self.stop_date = stop_date
 

	
 
    def section_key(self, row: FundPosts) -> None:
 
        return None
 

	
 
    def start_spreadsheet(self, *, expanded: bool=True) -> None:
 
        headers = [["Fund"], ["Balance as of", self.start_date.isoformat()]]
 
        if expanded:
 
            sheet_name = "With Breakdowns"
 
            headers += [["Income"], ["Expenses"], ["Equity"]]
 
        else:
 
            sheet_name = "Fund Report"
 
            headers += [["Additions"], ["Releases from", "Restrictions"]]
 
        headers.append(["Balance as of", self.stop_date.isoformat()])
 
        if expanded:
 
            headers += [
 
                ["Of which", "Receivable"],
 
                ["Of which", "Prepaid Expenses"],
 
                ["Of which", "Payable"],
 
                ["Of which", "Unearned Income"],
 
            ]
 

	
 
        self.use_sheet(sheet_name)
 
        for header in headers:
 
            first_line = header[0]
 
            if first_line == 'Fund':
 
                width = 2.0
 
            elif first_line == 'Balance as of':
 
                width = 1.5
 
            elif first_line == 'Of which':
 
                width = 1.3
 
            else:
 
                width = 1.2
 
            col_style = self.column_style(width)
 
            self.sheet.addElement(odf.table.TableColumn(stylename=col_style))
 

	
 
        center_bold = self.merge_styles(self.style_centertext, self.style_bold)
 
        row = self.add_row(*(
 
            self.multiline_cell(header, stylename=center_bold)
 
            for header in headers
 
        ))
 
        row.firstChild.setAttribute(
 
            'stylename', self.merge_styles(self.style_endtext, self.style_bold),
 
        )
 
        self.lock_first_row()
 
        self.lock_first_column()
 
        self.add_row()
 
        self.add_row(self.string_cell(
 
            f"Fund Report From {self.start_date.isoformat()} To {self.stop_date.isoformat()}",
 
            stylename=center_bold,
 
            numbercolumnsspanned=6,
 
        ))
 
        self.add_row()
 

	
 
    def end_spreadsheet(self) -> None:
 
        start_sheet = self.sheet
 
        self.set_open_sheet(self.sheet)
 
        self.start_spreadsheet(expanded=False)
 
        bal_indexes = [0, 1, 2, 4]
 
        totals = [core.MutableBalance() for _ in bal_indexes]
 
        threshold = Decimal('.5')
 
        for fund, balances in self.balances.items():
 
            balances = [balances[index] for index in bal_indexes]
 
            if (not all(bal.clean_copy(threshold).le_zero() for bal in balances)
 
                and fund != UNRESTRICTED_FUND):
 
                self.write_balances(fund, balances)
 
                for total, bal in zip(totals, balances):
 
                    total += bal
 
        self.write_balances('', totals, self.merge_styles(
 
            self.border_style(core.Border.TOP, '.75pt'),
 
            self.border_style(core.Border.BOTTOM, '1.5pt', 'double'),
 
        ))
 
        self.document.spreadsheet.childNodes.reverse()
 
        self.sheet = start_sheet
 

	
 
    def _row_balances(self, accounts_map: AccountsMap) -> Iterable[core.Balance]:
 
    def _row_balances(self, accounts_map: AccountsMap) -> Iterator[core.Balance]:
 
        acct_order = ['Income', 'Expenses', 'Equity']
 
        key_order = [core.OPENING_BALANCE_NAME, *acct_order, core.ENDING_BALANCE_NAME]
 
        balances: Dict[str, core.Balance] = {key: core.MutableBalance() for key in key_order}
 
        for acct_s, balance in core.account_balances(accounts_map, acct_order):
 
            if acct_s in balances:
 
                balances[acct_s] = balance
 
            else:
 
                acct_root, _, _ = acct_s.partition(':')
 
                balances[acct_root] += balance
 
        for key in key_order:
 
            if key == 'Expenses':
 
                yield balances[key]
 
            else:
 
                yield -balances[key]
 
        for info_key in INFO_ACCOUNTS:
 
            for _, balance in core.account_balances(accounts_map, [info_key]):
 
                pass
 
            yield core.normalize_amount_func(info_key)(balance)
 

	
 
    def write_balances(self,
 
                       fund: str,
 
                       balances: Iterable[core.Balance],
 
                       style: Union[None, str, odf.style.Style]=None,
 
    ) -> odf.table.TableRow:
 
        return self.add_row(
 
            self.string_cell(fund, stylename=self.style_endtext),
 
            *(self.balance_cell(bal, stylename=style) for bal in balances),
 
        )
 

	
 
    def write_row(self, row: FundPosts) -> None:
 
        fund, accounts_map = row
 
        self.balances[fund] = list(self._row_balances(accounts_map))
 
        if fund != UNRESTRICTED_FUND:
 
            self.write_balances(fund, self.balances[fund])
 

	
 
    def write(self, rows: Iterable[FundPosts]) -> None:
 
        self.balances: Dict[str, Sequence[core.Balance]] = collections.OrderedDict()
 
        super().write(rows)
 
        try:
 
            unrestricted = self.balances[UNRESTRICTED_FUND]
 
        except KeyError:
 
            pass
 
        else:
 
            self.add_row()
 
            self.write_balances("Unrestricted", unrestricted)
 

	
 

	
 
class TextReport:
 
    def __init__(self,
 
                 start_date: datetime.date,
 
                 stop_date: datetime.date,
 
                 out_file: TextIO) -> None:
 
        self.start_date = start_date
 
        self.stop_date = stop_date
 
        self.out_file = out_file
 

	
 
    def _account_balances(self,
 
                          fund: str,
 
                          account_map: AccountsMap,
 
    ) -> Iterator[Tuple[str, Sequence[str]]]:
 
        total_fmt = f'{fund} balance as of {{}}'
 
        for acct_s, balance in core.account_balances(account_map, EQUITY_ACCOUNTS):
 
            if acct_s is core.OPENING_BALANCE_NAME:
 
                acct_s = total_fmt.format(self.start_date.isoformat())
 
            elif acct_s is core.ENDING_BALANCE_NAME:
 
                acct_s = total_fmt.format(self.stop_date.isoformat())
 
            yield acct_s, (-balance).format(None, sep='\0').split('\0')
 
        for _, account in core.sort_and_filter_accounts(account_map, INFO_ACCOUNTS):
 
            balance = account_map[account].stop_bal
 
            if not balance.is_zero():
 
                balance = core.normalize_amount_func(account)(balance)
 
                yield account, balance.format(None, sep='\0').split('\0')
 

	
 
    def write(self, rows: Iterable[FundPosts]) -> None:
 
        output = [
 
            line
 
            for fund, account_map in rows
 
            for line in self._account_balances(fund, account_map)
 
        ]
 
        acct_width = max(len(acct_s) for acct_s, _ in output) + 2
 
        bal_width = max(len(s) for _, bal_s in output for s in bal_s)
 
        bal_width = max(bal_width, 8)
 
        line_fmt = f'{{:>{acct_width}}}  {{:>{bal_width}}}'
 
        print(line_fmt.replace('{:>', '{:^').format("ACCOUNT", "BALANCE"),
 
              file=self.out_file)
 
        fund_start = f' balance as of {self.start_date.isoformat()}'
 
        for acct_s, bal_seq in output:
 
            if acct_s.endswith(fund_start):
 
                print(line_fmt.format('―' * acct_width, '―' * bal_width),
 
                      file=self.out_file)
 
            bal_iter = iter(bal_seq)
 
            print(line_fmt.format(acct_s, next(bal_iter)), file=self.out_file)
 
            for bal_s in bal_iter:
 
                print(line_fmt.format('', bal_s), file=self.out_file)
 

	
 

	
0 comments (0 inline, 0 general)