Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def open_opt(entries, date, options_map):
"""Convenience function to open() using an options map.
"""
account_types = options.get_account_types(options_map)
previous_accounts = options.get_previous_accounts(options_map)
conversion_currency = options_map['conversion_currency']
return open(entries, date, account_types, conversion_currency, *previous_accounts)
c_where = None
if parsed_query.where_clause:
c_where = query_compile.compile_expression(parsed_query.where_clause, query_env.FilterPostingsEnvironment())
# Figure out if we need to compute balance.
balance = None
if c_where and query_execute.uses_balance_column(c_where):
balance = inventory.Inventory()
context = query_execute.RowContext()
context.balance = balance
# Initialize some global properties for use by some of the accessors.
context.options_map = options_map
context.account_types = options.get_account_types(options_map)
context.open_close_map = getters.get_account_open_close(entries)
#context.commodity_map = getters.get_commodity_map(entries)
context.price_map = prices.build_price_map(entries)
if c_from is not None:
filtered_entries = query_execute.filter_entries(c_from, entries, options_map)
else:
filtered_entries = entries
for entry in filtered_entries:
if isinstance(entry, Transaction):
context.entry = entry
matching_postings = []
for posting in entry.postings:
context.posting = posting
if c_where is None or c_where(context):
matching_postings.append(posting)
c_where = None
if parsed_query.where_clause:
c_where = query_compile.compile_expression(parsed_query.where_clause, query_env.FilterPostingsEnvironment())
# Figure out if we need to compute balance.
balance = None
if c_where and query_execute.uses_balance_column(c_where):
balance = inventory.Inventory()
context = query_execute.RowContext()
context.balance = balance
# Initialize some global properties for use by some of the accessors.
context.options_map = options_map
context.account_types = options.get_account_types(options_map)
context.open_close_map = getters.get_account_open_close(entries)
#context.commodity_map = getters.get_commodity_map(entries)
context.price_map = prices.build_price_map(entries)
if c_from is not None:
filtered_entries = query_execute.filter_entries(c_from, entries, options_map)
else:
filtered_entries = entries
return filtered_entries
# for entry in filtered_entries:
def get_assets_holdings(entries, options_map, currency=None):
"""Return holdings for all assets and liabilities.
Args:
entries: A list of directives.
options_map: A dict of parsed options.
currency: If specified, a string, the target currency to convert all
holding values to.
Returns:
A list of Holding instances and a price-map.
"""
# Compute a price map, to perform conversions.
price_map = prices.build_price_map(entries)
# Get the list of holdings.
account_types = options.get_account_types(options_map)
holdings_list = holdings.get_final_holdings(entries,
(account_types.assets,
account_types.liabilities),
price_map)
# Convert holdings to a unified currency.
if currency:
holdings_list = holdings.convert_to_currency(price_map, currency, holdings_list)
return holdings_list, price_map
parser.add_argument('filename', help='Ledger filename')
parser.add_argument('-r', '--related', action='store',
default='.*:ScotiaBank:.*',
help="Related accounts.")
opts = parser.parse_args()
entries, errors, options_map = loader.load(opts.filename)
# Find and print the lis tof external entries.
external_entries, dates = find_external_flows(entries, opts.related)
# Create a predicate for the Assets accounts.
acc_types = options.get_account_types(options_map)
match = re.compile(opts.related).match
is_asset_account = lambda account_: (
match(account_) and
account_types.get_account_type(account_) == acc_types.assets)
# Verify that external flow entries only affect balance sheet accounts and
# not income or expenses accounts.
for entry in external_entries:
for posting in entry.postings:
if (match(posting.account) and
not account_types.get_account_type(posting.account) == acc_types.assets):
raise ValueError(
"External flow may not affect non-asset accounts: {}".format(entry))
# Build a price database.
def create_row_context(entries, options_map):
"""Create the context container which we will use to evaluate rows."""
context = RowContext()
context.balance = inventory.Inventory()
# Initialize some global properties for use by some of the accessors.
context.options_map = options_map
context.account_types = options.get_account_types(options_map)
context.open_close_map = getters.get_account_open_close(entries)
context.commodity_map = getters.get_commodity_directives(entries)
context.price_map = prices.build_price_map(entries)
return context
# Get the filtered list of entries.
self.entries, self.begin_index = self.apply_filter(self.all_entries, options_map)
# Compute the list of entries for the opening balances sheet.
self.opening_entries = (self.entries[:self.begin_index]
if self.begin_index is not None
else [])
# Compute the list of entries that includes transfer entries of the
# income/expenses amounts to the balance sheet's equity (as "net
# income"). This is used to render the end-period balance sheet, with
# the current period's net income, closing the period.
self.closing_entries = summarize.cap_opt(self.entries, options_map)
# Realize the three sets of entries.
account_types = options.get_account_types(options_map)
with misc_utils.log_time('realize_opening', logging.info):
self.opening_real_accounts = realization.realize(self.opening_entries,
account_types)
with misc_utils.log_time('realize', logging.info):
self.real_accounts = realization.realize(self.entries,
account_types)
with misc_utils.log_time('realize_closing', logging.info):
self.closing_real_accounts = realization.realize(self.closing_entries,
account_types)
assert self.real_accounts is not None
assert self.closing_real_accounts is not None
def load_file(self, beancount_file_path=None):
"""Load self.beancount_file_path and compute things that are independent
of how the entries might be filtered later"""
if beancount_file_path:
self.beancount_file_path = beancount_file_path
self.all_entries, self.errors, self.options = \
loader.load_file(self.beancount_file_path)
self.price_map = prices.build_price_map(self.all_entries)
self.account_types = options.get_account_types(self.options)
self.title = self.options['title']
if self.options['render_commas']:
self.format_string = '{:,f}'
self.default_format_string = '{:,.2f}'
else:
self.format_string = '{:f}'
self.default_format_string = '{:.2f}'
self.dcontext = self.options['dcontext']
self.active_years = list(getters.get_active_years(self.all_entries))
self.active_tags = list(getters.get_all_tags(self.all_entries))
self.active_payees = list(getters.get_all_payees(self.all_entries))
self.queries = self._entries_filter_type(self.all_entries, Query)