The purpose of this exercise is to develop the model-building software, so everything should be done with interactive scripts.
In [1]:
import re
import os
import sys
import json
sys.path.append('../..')
import lcatools
In [204]:
from lcatools.foreground.fragment_flows import LcFragment
from lcatools.entities import LcFlow, LcQuantity
from lcatools.exchanges import Exchange, comp_dir
from lcatools.catalog import ExchangeRef
In [3]:
F = lcatools.ForegroundManager('pet')
In [4]:
F.show_all()
In [5]:
[str(q) for q in F[0].lcia_methods()]
Out[5]:
In [6]:
F.load_lcia_cfs('LCIA')
In [7]:
['%s %-70.70s' % (x.get_uuid(), x) for x in F['LCIA'].lcia_methods() if re.search('midpoint', x['Name'])]
Out[7]:
In [82]:
ilcd_qs = ('ec7836be-83eb-41da-bcda-1a6a3fe2d149', 'f6cbd466-253f-4145-a4bb-8dae7d266e89',
'f4602cba-f746-4351-b4da-e91181853e33', 'a4630d5b-df2f-48ae-b3f2-5e5257606d68',
'72ce3013-f678-4c6a-935b-f4752ff8901b', '370960f4-0a3a-415d-bf3e-e5ce63160bb9',
'5296e2be-060b-4e50-b033-d45f85f6ac92', '40054970-5936-477c-9bec-29fa23b1cb77'
)
In [83]:
for q in ilcd_qs:
F.add_to_foreground(F['LCIA'][q])
In [6]:
F.load_lcia_cfs('EI-LCIA')
In [7]:
F.db.factors_for_quantity('f6cbd466-253f-4145-a4bb-8dae7d266e89')
In [8]:
F.load(8)
In [9]:
from lcatools.interact import pick_one, cyoa, ifinput, pick_list, _pick_list
In [10]:
def find_flow(name, index=None, elementary=False):
if name is None:
name = input('Enter flow name search string: ')
res = F.search(index, 'flow', Name=name)
if elementary is not None:
res = list(filter(lambda x: F.db.is_elementary(x.entity()) == elementary, res))
pick = pick_one(res)
print('Picked: %s' % pick)
return pick
In [203]:
current_archive = None
def select_archive():
F.show_all()
ch1 = input('Search which catalog? or blank to search all loaded catalogs')
if len(ch1) == 0:
index = None
else:
try:
index = int(ch1)
except ValueError:
index = F._catalog.get_index(ch1)
current_archive = index
return index
def create_fragment(parent=None):
ch = cyoa('(N)ew flow or (S)earch for flow? ', 'ns')
if ch == 'n':
flow = F.new_flow()
elif ch == 's':
index = current_archive or select_archive()
elem = {'i': False,
'e': True,
'a': None}[cyoa('(I)ntermediate, (E)lementary, or (A)ll flows? ', 'aei', 'I').lower()]
flow = find_flow(None, index=index, elementary=elem)
if flow is None:
return None
else:
raise ValueError
direction = { 'i': 'Input', 'o': 'Output'}[cyoa('flow is (I)nput or (O)utput?', 'IO').lower()]
comment = ifinput('Enter comment: ', '')
if parent is None:
# direction reversed for UX! user inputs direction w.r.t. fragment, not w.r.t. parent
frag = F.new_fragment(flow, comp_dir(direction), Comment=comment)
else:
frag = F[0].add_child_fragment_flow(parent, flow.entity(), direction, Comment=comment)
return frag
In [11]:
current_archive = select_archive()
In [12]:
frag = create_fragment()
In [13]:
def find_termination(ref, index=None, direction=None):
if isinstance(ref, LcFragment):
if index is None:
index = current_archive or select_archive()
terms = F._catalog.terminate_fragment(index, ref)
else:
if direction is None:
direction = { 'i': 'Input', 'o': 'Output'}[cyoa('(I)nput or (O)utput?', 'IO').lower()]
terms = F._catalog.terminate_flow(flow_ref, direction)
pick = pick_one(terms)
print('Picked: %s' % pick)
return pick
In [14]:
def add_termination(frag, term, scenario=None):
if isinstance(term, ExchangeRef):
frag.term_from_exch(term, scenario=scenario)
else:
frag.terminate(term, scenario=scenario)
F.build_child_flows(frag, background_children=True)
In [15]:
def auto_terminate(frag, index=None):
ex = find_termination(frag, index=index)
add_termination(frag, ex)
In [16]:
def foreground(frag, index=None):
F.fragment_to_foreground(frag)
auto_terminate(frag)
In [15]:
auto_terminate(frag)
In [208]:
def del_orphans(self, for_real=False):
"""
self is a foreground archive
"""
for f in self.fragments(show_all=True):
if f.reference_entity is not None:
continue
try:
next(_find_links(self, f))
print('Found a link for %s' % f)
except StopIteration:
print('Deleting %s' % f)
if for_real:
del self._entities[f._uuid]
def _find_links(self, frag):
for i in self.fragments(show_all=True):
if i.reference_entity is frag:
yield i
else:
for v in i._terminations.values():
if v.term_node is frag:
yield i
In [211]:
del_orphans(F[0])
In [12]:
F.draw_fragment(F.frag('1b7'))
In [54]:
F.frag('b14').terminate(None)
In [30]:
[str(x) for x in F[0].fragments(show_all=True, background=False, match='combust')]
Out[30]:
In [18]:
F.show_fragments(show_all=False, background=False)
In [13]:
F.compute_unit_scores()
In [14]:
F.show_all()
In [15]:
frag = F.frag('0c4')
In [224]:
_qs = (x for x in F[0].lcia_methods())
In [223]:
results = F.fg_lcia(frag.term.term_node)
In [225]:
q = next(_qs)
str(q)
Out[225]:
In [242]:
F.db.flowables.synonyms_for('010102-44-0')
Out[242]:
In [69]:
[x.flow.get_uuid() for x in frag.term.term_node.entity().exchanges() if x.flow['Name'] == 'Nitrogen oxides ']
Out[69]:
In [240]:
F.db.flowables.add_synonym('010102-44-0', '2fb4ca68-6eb1-3f0b-b907-70cdc6f050f2')
Out[240]:
In [226]:
results[q.get_uuid()].show_details(show_all=True)
In [227]:
F.db.factors_for_quantity(q)
In [233]:
for fb in F.db.flowables.search('chromium III'):
print(F.db.flowables.synonym_set(fb))
In [234]:
for fb in F.db.flowables.search('chromium VI'):
print(F.db.flowables.synonym_set(fb))
In [111]:
F.show_fragments()
In [95]:
F.frag('1b7').direction
Out[95]:
In [80]:
ios = F.get_fragment_inventory(F.frag('1b7'))
In [144]:
ffs, _ = F.frag('1b7').traverse(lambda x: F.child_flows(x), 1.0, None)
In [145]:
ffs
Out[145]:
In [151]:
F.draw_fragment(F.frag('1b7'))
In [158]:
F.draw_fragment(F.frag('78a').term.term_node)
In [159]:
F.draw_fragment(F.frag('cf3').term.term_node)
In [174]:
tot = 0.0
for f in ffs:
if f.term.score_cache(q) is None:
score = '--'
total = ' '
else:
scr = f.term.score_cache(q).total()
score = '%-10.3g' % scr
total = '%-10.3g' % (scr * f.node_weight)
tot += (scr * f.node_weight)
print('%-10.10s %-10.10s %s' % (total, score, f))
print('%-10.3g Total' % tot)
In [183]:
['%10.6g %s' % (x.value, x.flow) for x in F.db.filter_exch(F.frag('b78').term.term_node, elem=False)]
Out[183]:
In [184]:
['%10.6g %s' % (x.value, x.flow) for x in F.db.filter_exch(F.frag('1b7').term.term_node, elem=False)]
Out[184]:
In [220]:
(.0121378 )/ (1.0 - .0144)
Out[220]:
In [191]:
F.frag('1b7').term.term_flow.origin
Out[191]:
In [157]:
ffs[1].__dict__
Out[157]:
In [156]:
F.elementary(ffs[0].term.term_node)
In [192]:
fnew = create_fragment()
In [198]:
F.terminate_to_foreground(fnew)
In [212]:
fnew2 = create_fragment(fnew)
In [216]:
F.draw_fragment(fnew)
In [219]:
F.frag('b98').exchange_value(0)
Out[219]:
In [214]:
F.frag('6c6e').terminate(F.frag('1b7'))
In [215]:
F.build_child_flows(F.frag('6c6e'))
Out[215]:
In [146]:
R = traversal_to_lcia(ffs)
In [147]:
lcatools.show_lcia(R)
In [148]:
R[q.get_uuid()].show_details()
In [149]:
R[q.get_uuid()].show_components()
In [116]:
str(q)
Out[116]:
In [119]:
F.draw_fragment(F.frag('1b7'))
In [140]:
R[q.get_uuid()].show_components()
In [120]:
F.draw_fragment(F.frag('68d'))
In [136]:
R = F.fg_lcia(F.frag('114').term.term_node)
In [137]:
R[q.get_uuid()].show_details()
In [132]:
for f in F[0].fragments(show_all=True):
f.term.clear_score_cache()
In [133]:
%time F.compute_unit_scores()
In [134]:
F.fragment_lcia(F.frag('114'))[q.get_uuid()].show_details()
In [124]:
from lcatools.foreground.fragment_flows import traversal_to_lcia
In [125]:
F.show_fragments()
In [94]:
[str(x) for x in ffs]
Out[94]:
In [109]:
['%6.3g %s %s' % (z.value, z.direction, z.flow) for z in ios]
Out[109]:
In [217]:
(6.56e-5 + 9.33e-7) / (1 - 0)
Out[217]:
In [105]:
ios[1].value
Out[105]:
In [ ]:
f.
In [53]:
F.find_flowable('Nitrogen oxides '.strip())
Out[53]:
In [47]:
fb
Out[47]:
In [41]:
cm = F.db.find_matching_compartment(['Emissions to air'])
In [22]:
F.draw_fragment(F.frag('0c4'))
In [21]:
lcatools.show_lcia(F.frag('0c4').term._score_cache)
In [55]:
foreground(F.frag('d00'))
In [51]:
F.frag('db3').terminate(None)
In [92]:
F.save()
In [61]:
for f in F[0].fragments(background=False, show_all=True, match='combus'):
if not f.term.is_null:
if f.term.is_bg:
foreground(f)
# print('%s' % f)
In [73]:
del_orphans(F[0])
In [69]:
del F[0]._entities[F.frag('4ec')._uuid]
In [67]:
F[0]._entities.remove(F.frag('4ec'))
In [70]:
F.show_fragments(show_all=True, background=True, match='transport')
In [49]:
F.frag('68f').terminate(F.frag('ba3'))
In [40]:
[str(x) for x in frag.io_flows(lambda x: F.child_flows(x), None)]
Out[40]:
In [36]:
F.frag('b8e').reference_entity = None
In [28]:
F.fragment_to_background(F.frag('094'))
Out[28]:
In [ ]:
for f in ('')
In [33]:
F.frag('ba3c')._background=False
In [36]:
F.build_child_flows(F.frag('ba3'), background_children=True)
Out[36]:
In [31]:
auto_terminate(F.frag('ba3c'))
In [25]:
F.frag('15f1').terminate(None)
In [29]:
foreground(F.frag('28b5'))
In [19]:
auto_terminate(F.frag('491'))
In [9]:
def create_quantity(self):
name = input('Enter quantity name: ')
unit = ifinput('Enter reference unit: ', 'kg')
comment = ifinput('Enter comment: ', '')
q = LcQuantity.new(name, unit, Comment=comment)
self[0].add(q)
return q
In [10]:
def edit_flow(self):
flow = pick_one(self._catalog[0].flows())
ch = cyoa('Edit (P)roperties or (C)haracterizations? ', 'pc').lower()
if ch == 'p':
_edit_entity(flow)
elif ch == 'c':
edit_characterizations(flow)
def _edit_entity(entity):
print('Select field to edit:')
field = menu_list(*entity.keys())
if field == -1 or field is None:
return True
new = ifinput('Enter new value for %s: ' % field, flow[field])
if len(new) > 0:
entity[field] = new
else:
print('Not updating.')
def edit_characterizations(flow):
char = pick_one(cf for cf in flow.characterizations())
val = float(ifinput('enter new characterization value: ', char.value ))
char.value = val
In [13]:
def pick_fragment(background=False):
"""
self is a foreground manager
"""
frag = pick_list([f for f in F[0].fragments(background=background)])
print('%s' % frag)
return frag
In [14]:
def show_info(frag):
frag.show()
print('Exchange Values:')
evs = list(frag._exchange_values.keys())
evs.remove(0)
evs.remove(1)
print('%20.20s: %g' % ('Cached', frag.cached_ev))
print('%20.20s: %g' % ('Observed', frag.observed_ev))
for k in evs:
print('%20.20s: %g' % (k, frag.exchange_value(k)))
if frag._balance_flow:
print('\nBalance flow: True (%s)' % frag.flow.reference_entity)
else:
print('\nBalance flow: False')
def show_terminations(frag):
print('Terminations: ')
print('%20s: %s' % ('Scenario', 'Termination'))
for k, v in frag._terminations.items():
if v.descend:
desc = '(agg)'
else:
desc = ' '
print('%20.20s: %s %s %s' % (k, v, desc, v.term_node))
In [28]:
def _pick_scenario(scs):
sc = _pick_list(scs, 'New scenario', 'Cancel')
if sc == (None, None):
return None
if sc == (None, 1):
scen = input('Enter scenario Name: ')
elif sc == (None, 2):
return None
else:
scen = scs[sc[0]]
if scen == '0':
scen = 0
return scen
def _f_add_child(frag):
return create_fragment(frag)
def _f_info(frag):
_edit_entity(frag)
def _f_exch(frag):
action = cyoa('Set (O)bserved exchange value, (S)cenario exchange value, (B)alance? ', 'osb')
if action == 'o':
frag.observed_ev = float(ifinput('Enter observed exchange value: ', frag.observed_ev))
elif action == 's':
scs = list(frag._exchange_values.keys())
if frag.cached_ev != 1.0:
scs.remove(0)
scs.remove(1)
sc = _pick_scenario(scs)
if sc == None:
return
elif sc == 0:
cev = frag.cached_ev
elif sc in frag._exchange_values.keys():
cev = frag.exchange_value(sc)
else:
cev = 1.0
ev = float(ifinput('Enter exchange value: ', cev))
frag.set_exchange_value(sc, ev)
elif action == 'b':
if frag._balance_flow:
frag.unset_balance_flow()
else:
frag.set_balance_flow()
def _f_terms(frag):
scs = list(frag.terminations())
sc = _pick_scenario(scs)
term = find_termination(frag)
if term is None:
ch = cyoa('No terminations found. Search for termination by (P)rocess or by (F)low? ', 'pf')
else:
add_termination(frag, term, scenario=sc)
def _f_select_child(frag):
ch = pick_list([f for f in F.child_flows(frag)])
edit_fragment(ch)
def edit_fragment(frag):
print('%s' % frag)
F._show_frag_children(frag)
show_info(frag)
show_terminations(frag)
action = {'a': _f_add_child, 'i': _f_info, 'e': _f_exch, 't': _f_terms, 's': _f_select_child, 'q': None
}[cyoa('Choices: (A)dd child, (I)nfo, (E)xchange values, (T)erminations; (S)elect child, (Q)uit: ', 'aeitsq', 'i')]
if action is None:
return
action(frag)
In [21]:
F.show_fragments(background=None)
In [29]:
edit_fragment(frag)
In [72]:
frag.term.__dict__
Out[72]:
In [64]:
frbc = F.frag('ebfc')
In [66]:
help(del)
In [63]:
F.draw_fragment(frag)
In [58]:
auto_terminate(F.frag('2f5d'), index=8)
In [ ]:
In [ ]:
In [ ]:
In [ ]:
In [ ]:
In [ ]:
In [ ]:
In [ ]:
In [48]:
str(flow_ref)
Out[48]:
In [51]:
x = find_termination(flow_ref, direction)
In [40]:
str(x.exchange)
Out[40]: