-
Notifications
You must be signed in to change notification settings - Fork 227
/
model.py
1935 lines (1627 loc) · 92.2 KB
/
model.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
#!/usr/bin/env python
# -*- coding: utf-8 -*-
###############################################################################
# #
# RMG - Reaction Mechanism Generator #
# #
# Copyright (c) 2002-2019 Prof. William H. Green (whgreen@mit.edu), #
# Prof. Richard H. West (r.west@neu.edu) and the RMG Team (rmg_dev@mit.edu) #
# #
# Permission is hereby granted, free of charge, to any person obtaining a #
# copy of this software and associated documentation files (the 'Software'), #
# to deal in the Software without restriction, including without limitation #
# the rights to use, copy, modify, merge, publish, distribute, sublicense, #
# and/or sell copies of the Software, and to permit persons to whom the #
# Software is furnished to do so, subject to the following conditions: #
# #
# The above copyright notice and this permission notice shall be included in #
# all copies or substantial portions of the Software. #
# #
# THE SOFTWARE IS PROVIDED 'AS IS', WITHOUT WARRANTY OF ANY KIND, EXPRESS OR #
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, #
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE #
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER #
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING #
# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER #
# DEALINGS IN THE SOFTWARE. #
# #
###############################################################################
"""
Contains classes for working with the reaction model generated by RMG.
"""
import gc
import itertools
import logging
import os
import numpy
import rmgpy.data.rmg
from rmgpy import settings
from rmgpy.constraints import failsSpeciesConstraints
from rmgpy.data.kinetics.depository import DepositoryReaction
from rmgpy.data.kinetics.family import KineticsFamily, TemplateReaction
from rmgpy.data.kinetics.library import KineticsLibrary, LibraryReaction
from rmgpy.data.rmg import getDB
from rmgpy.display import display
from rmgpy.exceptions import ForbiddenStructureException
from rmgpy.kinetics import KineticsData, Arrhenius
from rmgpy.quantity import Quantity
from rmgpy.reaction import Reaction
from rmgpy.rmg.pdep import PDepReaction, PDepNetwork
from rmgpy.rmg.react import react_all
from rmgpy.species import Species
from rmgpy.thermo.thermoengine import submit
################################################################################
class ReactionModel:
"""
Represent a generic reaction model. A reaction model consists of `species`,
a list of species, and `reactions`, a list of reactions.
"""
def __init__(self, species=None, reactions=None):
self.species = species or []
self.reactions = reactions or []
def __reduce__(self):
"""
A helper function used when pickling an object.
"""
return (ReactionModel, (self.species, self.reactions))
def merge(self, other):
"""
Return a new :class:`ReactionModel` object that is the union of this
model and `other`.
"""
if not isinstance(other, ReactionModel):
raise ValueError('Expected type ReactionModel for other parameter, got {0}'.format(other.__class__))
# Initialize the merged model
finalModel = ReactionModel()
# Put the current model into the merged model as-is
finalModel.species.extend(self.species)
finalModel.reactions.extend(self.reactions)
# Determine which species in other are already in self
commonSpecies = {}; uniqueSpecies = []
for spec in other.species:
for spec0 in finalModel.species:
if spec.isIsomorphic(spec0):
commonSpecies[spec] = spec0
if spec0.label not in ['Ar','N2','Ne','He']:
if not spec0.thermo.isIdenticalTo(spec.thermo):
print 'Species {0} thermo from model 1 did not match that of model 2.'.format(spec.label)
break
else:
uniqueSpecies.append(spec)
# Determine which reactions in other are already in self
commonReactions = {}; uniqueReactions = []
for rxn in other.reactions:
for rxn0 in finalModel.reactions:
if rxn.isIsomorphic(rxn0, eitherDirection=True):
commonReactions[rxn] = rxn0
if not rxn0.kinetics.isIdenticalTo(rxn.kinetics):
print 'Reaction {0} kinetics from model 1 did not match that of model 2.'.format(str(rxn0))
break
else:
uniqueReactions.append(rxn)
# Add the unique species from other to the final model
finalModel.species.extend(uniqueSpecies)
# Make sure unique reactions only refer to species in the final model
for rxn in uniqueReactions:
for i, reactant in enumerate(rxn.reactants):
try:
rxn.reactants[i] = commonSpecies[reactant]
if rxn.pairs:
for j, pair in enumerate(rxn.pairs):
if reactant in pair:
rxn.pairs[j] = (rxn.reactants[i],pair[1])
except KeyError:
pass
for i, product in enumerate(rxn.products):
try:
rxn.products[i] = commonSpecies[product]
if rxn.pairs:
for j, pair in enumerate(rxn.pairs):
if product in pair:
rxn.pairs[j] = (pair[0], rxn.products[i])
except KeyError:
pass
# Add the unique reactions from other to the final model
finalModel.reactions.extend(uniqueReactions)
# Return the merged model
return finalModel
################################################################################
class CoreEdgeReactionModel:
"""
Represent a reaction model constructed using a rate-based screening
algorithm. The species and reactions in the model itself are called the
*core*; the species and reactions identified as candidates for inclusion in
the model are called the *edge*. The attributes are:
========================= ==============================================================
Attribute Description
========================= ==============================================================
`core` The species and reactions of the current model core
`edge` The species and reactions of the current model edge
`networkDict` A dictionary of pressure-dependent reaction networks (:class:`Network` objects) indexed by source.
`networkList` A list of pressure-dependent reaction networks (:class:`Network` objects)
`networkCount` A counter for the number of pressure-dependent networks created
`indexSpeciesDict` A dictionary with a unique index pointing to the species objects
`solventName` String describing solvent name for liquid reactions. Empty for non-liquid estimation
`surfaceSiteDensity` The surface site density (a SurfaceConcentration quantity) or None if no heterogeneous catalyst.
========================= ==============================================================
"""
def __init__(self, core=None, edge=None, surface=None):
if core is None:
self.core = ReactionModel()
else:
self.core = core
if edge is None:
self.edge = ReactionModel()
else:
self.edge = edge
if surface is None:
self.surface = ReactionModel()
else:
self.surface = surface
# The default tolerances mimic the original RMG behavior; no edge
# pruning takes place, and the simulation is interrupted as soon as
# a species flux higher than the validity
self.networkDict = {}
self.networkList = []
self.networkCount = 0
self.speciesDict = {}
self.reactionDict = {}
self.speciesCache = [None for i in range(4)]
self.speciesCounter = 0
self.reactionCounter = 0
self.newSpeciesList = []
self.newReactionList = []
self.outputSpeciesList = []
self.outputReactionList = []
self.pressureDependence = None
self.quantumMechanics = None
self.verboseComments = False
self.kineticsEstimator = 'rate rules'
self.indexSpeciesDict = {}
self.saveEdgeSpecies = False
self.iterationNum = 0
self.toleranceThermoKeepSpeciesInEdge = numpy.inf
self.Gfmax = numpy.inf
self.Gmax = numpy.inf
self.Gmin = -numpy.inf
self.minCoreSizeForPrune = 50
self.maximumEdgeSpecies = 100000
self.Tmax = 0
self.reactionSystems = []
self.newSurfaceSpcsAdd = set()
self.newSurfaceRxnsAdd = set()
self.newSurfaceSpcsLoss = set()
self.newSurfaceRxnsLoss = set()
self.solventName = ''
self.surfaceSiteDensity = None
def checkForExistingSpecies(self, molecule):
"""
Check to see if an existing species contains the same
:class:`molecule.Molecule` as `molecule`. Comparison is done using
isomorphism without consideration of electrons. Therefore, resonance
structures of a species will all match each other.
Returns the matched species if found and `None` otherwise.
"""
# First check cache and return if species is found
for i, spec in enumerate(self.speciesCache):
if spec is not None and spec.isIsomorphic(molecule, strict=False):
self.speciesCache.pop(i)
self.speciesCache.insert(0, spec)
return spec
# If not found in cache, check all species with matching formula
formula = molecule.getFormula()
try:
species_list = self.speciesDict[formula]
except KeyError:
pass
else:
for spec in species_list:
if spec.isIsomorphic(molecule, strict=False):
self.speciesCache.pop()
self.speciesCache.insert(0, spec)
return spec
# At this point we can conclude that the species is new
return None
def makeNewSpecies(self, object, label='', reactive=True, checkForExisting=True, generateThermo=True):
"""
Formally create a new species from the specified `object`, which can be
either a :class:`Molecule` object or an :class:`rmgpy.species.Species`
object. It is emphasized that `reactive` relates to the :Class:`Species` attribute, while `reactive_structure`
relates to the :Class:`Molecule` attribute.
"""
if isinstance(object, rmgpy.species.Species):
molecule = object.molecule[0]
label = label if label != '' else object.label
reactive = object.reactive
else:
molecule = object
molecule.clearLabeledAtoms()
# If desired, check to ensure that the species is new; return the
# existing species if not new
if checkForExisting:
spec = self.checkForExistingSpecies(molecule)
if spec is not None:
return spec, False
# If we're here then we're ready to make the new species
if reactive:
self.speciesCounter += 1 # count only reactive species
speciesIndex = self.speciesCounter
else:
speciesIndex = -1
try:
spec = Species(index=speciesIndex, label=label, molecule=[molecule], reactive=reactive,
thermo=object.thermo, transportData=object.transportData)
except AttributeError:
spec = Species(index=speciesIndex, label=label, molecule=[molecule], reactive=reactive)
spec.creationIteration = self.iterationNum
spec.generate_resonance_structures()
spec.molecularWeight = Quantity(spec.molecule[0].getMolecularWeight()*1000.,"amu")
if generateThermo:
self.generateThermo(spec)
# If the species still does not have a label, set initial label as the SMILES
# This may change later after getting thermo in self.generateThermo()
if not spec.label:
spec.label = spec.SMILES
logging.debug('Creating new species {0}'.format(spec.label))
formula = molecule.getFormula()
if formula in self.speciesDict:
self.speciesDict[formula].append(spec)
else:
self.speciesDict[formula] = [spec]
# Since the species is new, add it to the list of new species
self.newSpeciesList.append(spec)
if spec.reactive:
self.indexSpeciesDict[spec.index] = spec
return spec, True
def checkForExistingReaction(self, rxn):
"""
Check to see if an existing reaction has the same reactants, products, and
family as `rxn`. Returns :data:`True` or :data:`False` and the matched
reaction (if found).
First, a shortlist of reaction is retrieved that have the same reaction keys
as the parameter reaction.
Next, the reaction ID containing an identifier (e.g. label) of the reactants
and products is compared between the parameter reaction and the each of the
reactions in the shortlist. If a match is found, the discovered reaction is
returned.
If a match is not yet found, the Library (seed mechs, reaction libs)
in the reaction database are iterated over to check if a reaction was overlooked
(a reaction with a different "family" key as the parameter reaction).
"""
# Make sure the reactant and product lists are sorted before performing the check
rxn.reactants.sort()
rxn.products.sort()
# If reactants and products are identical, then something weird happened along
# the way and we got a symmetrical reaction.
if rxn.reactants == rxn.products:
logging.debug("Symmetrical reaction found. Returning no reaction")
return True, None
familyObj = getFamilyLibraryObject(rxn.family)
shortlist = self.searchRetrieveReactions(rxn)
# Now use short-list to check for matches. All should be in same forward direction.
# Make sure the reactant and product lists are sorted before performing the check
rxn_id = generateReactionId(rxn)
for rxn0 in shortlist:
rxn_id0 = generateReactionId(rxn0)
if rxn_id == rxn_id0 and areIdenticalSpeciesReferences(rxn, rxn0):
if isinstance(familyObj, KineticsLibrary) or isinstance(familyObj, KineticsFamily):
if not rxn.duplicate:
return True, rxn0
else:
return True, rxn0
elif (isinstance(familyObj, KineticsFamily)
and rxn_id == rxn_id0[::-1]
and areIdenticalSpeciesReferences(rxn, rxn0)):
if not rxn.duplicate:
return True, rxn0
# Now check seed mechanisms
# We want to check for duplicates in *other* seed mechanisms, but allow
# duplicated *within* the same seed mechanism
_, r1_fwd, r2_fwd = generateReactionKey(rxn)
_, r1_rev, r2_rev = generateReactionKey(rxn, useProducts=True)
for library in self.reactionDict:
libObj = getFamilyLibraryObject(library)
if isinstance(libObj, KineticsLibrary) and library != rxn.family:
# First check seed short-list in forward direction
shortlist = self.retrieve(library, r1_fwd, r2_fwd)
for rxn0 in shortlist:
rxn_id0 = generateReactionId(rxn0)
if (rxn_id == rxn_id0) or (rxn_id == rxn_id0[::-1]):
if areIdenticalSpeciesReferences(rxn, rxn0):
return True, rxn0
# Now get the seed short-list of the reverse reaction
shortlist = self.retrieve(library, r1_rev, r2_rev)
for rxn0 in shortlist:
if areIdenticalSpeciesReferences(rxn, rxn0):
return True, rxn0
return False, None
def makeNewReaction(self, forward, checkExisting=True, generateThermo=True):
"""
Make a new reaction given a :class:`Reaction` object `forward`.
The reaction is added to the global list of reactions.
Returns the reaction in the direction that corresponds to the
estimated kinetics, along with whether or not the reaction is new to the
global reaction list.
The forward direction is determined using the "is_reverse" attribute of the
reaction's family. If the reaction family is its own reverse, then it is
made such that the forward reaction is exothermic at 298K.
The forward reaction is appended to self.newReactionList if it is new.
"""
# Determine the proper species objects for all reactants and products
reactants = [self.makeNewSpecies(reactant, generateThermo=generateThermo)[0] for reactant in forward.reactants]
products = [self.makeNewSpecies(product, generateThermo=generateThermo)[0] for product in forward.products ]
if forward.specificCollider is not None:
forward.specificCollider = self.makeNewSpecies(forward.specificCollider)[0]
if forward.pairs is not None:
for pairIndex in range(len(forward.pairs)):
reactantIndex = forward.reactants.index(forward.pairs[pairIndex][0])
productIndex = forward.products.index(forward.pairs[pairIndex][1])
forward.pairs[pairIndex] = (reactants[reactantIndex], products[productIndex])
if hasattr(forward, 'reverse'):
if forward.reverse:
forward.reverse.pairs[pairIndex] = (products[productIndex], reactants[reactantIndex])
forward.reactants = reactants
forward.products = products
if checkExisting:
found, rxn = self.checkForExistingReaction(forward)
if found: return rxn, False
# Generate the reaction pairs if not yet defined
if forward.pairs is None:
forward.generatePairs()
if hasattr(forward, 'reverse'):
if forward.reverse:
forward.reverse.generatePairs()
# Note in the log
if isinstance(forward, TemplateReaction):
logging.debug('Creating new {0} template reaction {1}'.format(forward.family, forward))
elif isinstance(forward, DepositoryReaction):
logging.debug('Creating new {0} reaction {1}'.format(forward.getSource(), forward))
elif isinstance(forward, LibraryReaction):
logging.debug('Creating new library reaction {0}'.format(forward))
else:
raise Exception("Unrecognized reaction type {0!s}".format(forward.__class__))
self.registerReaction(forward)
forward.index = self.reactionCounter + 1
self.reactionCounter += 1
# Since the reaction is new, add it to the list of new reactions
self.newReactionList.append(forward)
# Return newly created reaction
return forward, True
def makeNewPDepReaction(self, forward):
"""
Make a new pressure-dependent reaction based on a list of `reactants` and a
list of `products`. The reaction belongs to the specified `network` and
has pressure-dependent kinetics given by `kinetics`.
No checking for existing reactions is made here. The returned PDepReaction
object is not added to the global list of reactions, as that is intended
to represent only the high-pressure-limit set. The reactionCounter is
incremented, however, since the returned reaction can and will exist in
the model edge and/or core.
"""
# Don't create reverse reaction: all such reactions are treated as irreversible
# The reverse direction will come from a different partial network
# Note that this isn't guaranteed to satisfy thermodynamics (but will probably be close)
forward.reverse = None
forward.reversible = False
# Generate the reaction pairs if not yet defined
if forward.pairs is None:
forward.generatePairs()
# Set reaction index and increment the counter
forward.index = self.reactionCounter + 1
self.reactionCounter += 1
return forward
def enlarge(self, newObject=None, reactEdge=False,
unimolecularReact=None, bimolecularReact=None, trimolecularReact=None):
"""
Enlarge a reaction model by processing the objects in the list `newObject`.
If `newObject` is a
:class:`rmg.species.Species` object, then the species is moved from
the edge to the core and reactions generated for that species, reacting
with itself and with all other species in the model core. If `newObject`
is a :class:`rmg.unirxn.network.Network` object, then reactions are
generated for the species in the network with the largest leak flux.
If the `reactEdge` flag is `True`, then no newObject is needed,
and instead the algorithm proceeds to react the core species together
to form edge reactions.
"""
numOldCoreSpecies = len(self.core.species)
numOldCoreReactions = len(self.core.reactions)
numOldEdgeSpecies = len(self.edge.species)
numOldEdgeReactions = len(self.edge.reactions)
reactionsMovedFromEdge = []
self.newReactionList = []; self.newSpeciesList = []
# Determine number of parallel processes.
from rmgpy.rmg.main import determine_procnum_from_RAM
procnum = determine_procnum_from_RAM()
if reactEdge is False:
# We are adding core species
newReactions = []
pdepNetwork = None
objectWasInEdge = False
if isinstance(newObject, Species):
newSpecies = newObject
objectWasInEdge = newSpecies in self.edge.species
if not newSpecies.reactive:
logging.info('NOT generating reactions for unreactive species {0}'.format(newSpecies))
else:
logging.info('Adding species {0} to model core'.format(newSpecies))
display(newSpecies) # if running in IPython --pylab mode, draws the picture!
# Add new species
reactionsMovedFromEdge = self.addSpeciesToCore(newSpecies)
elif isinstance(newObject, tuple) and isinstance(newObject[0], PDepNetwork) and self.pressureDependence:
pdepNetwork, newSpecies = newObject
newReactions.extend(pdepNetwork.exploreIsomer(newSpecies))
self.processNewReactions(newReactions, newSpecies, pdepNetwork, generateThermo=False)
else:
raise TypeError('Unable to use object {0} to enlarge reaction model; expecting an object of class rmg.model.Species or rmg.model.PDepNetwork, not {1}'.format(newObject, newObject.__class__))
# If there are any core species among the unimolecular product channels
# of any existing network, they need to be made included
for network in self.networkList:
network.updateConfigurations(self)
index = 0
isomers = [isomer.species[0] for isomer in network.isomers]
while index < len(self.core.species):
species = self.core.species[index]
if species in isomers and species not in network.explored:
network.explored.append(species)
continue
for products in network.products:
products = products.species
if len(products) == 1 and products[0] == species:
newReactions = network.exploreIsomer(species)
self.processNewReactions(newReactions, species, network, generateThermo=False)
network.updateConfigurations(self)
index = 0
break
else:
index += 1
if isinstance(newObject, Species) and objectWasInEdge:
# moved one species from edge to core
numOldEdgeSpecies -= 1
# moved these reactions from edge to core
numOldEdgeReactions -= len(reactionsMovedFromEdge)
else:
# Generate reactions between all core species which have not been
# reacted yet and exceed the reaction filter thresholds
rxnLists, spcsTuples = react_all(self.core.species, numOldCoreSpecies,
unimolecularReact, bimolecularReact,
trimolecularReact=trimolecularReact,
procnum=procnum)
for rxnList, spcTuple in zip(rxnLists, spcsTuples):
if rxnList:
# Identify a core species which was used to generate the reaction
# This is only used to determine the reaction direction for processing
spc = spcTuple[0]
self.processNewReactions(rxnList, spc, generateThermo=False)
################################################################
# Begin processing the new species and reactions
# Generate thermo for new species
if self.newSpeciesList:
logging.info('Generating thermo for new species...')
self.applyThermoToSpecies(procnum)
# Do thermodynamic filtering
if not numpy.isinf(self.toleranceThermoKeepSpeciesInEdge) and self.newSpeciesList != []:
self.thermoFilterSpecies(self.newSpeciesList)
# Generate kinetics of new reactions
if self.newReactionList:
logging.info('Generating kinetics for new reactions...')
for reaction in self.newReactionList:
# If the reaction already has kinetics (e.g. from a library),
# assume the kinetics are satisfactory
if reaction.kinetics is None:
self.applyKineticsToReaction(reaction)
# For new reactions, convert ArrheniusEP to Arrhenius, and fix barrier heights.
# self.newReactionList only contains *actually* new reactions, all in the forward direction.
for reaction in self.newReactionList:
# convert KineticsData to Arrhenius forms
if isinstance(reaction.kinetics, KineticsData):
reaction.kinetics = reaction.kinetics.toArrhenius()
# correct barrier heights of estimated kinetics
if isinstance(reaction,TemplateReaction) or isinstance(reaction,DepositoryReaction): # i.e. not LibraryReaction
reaction.fixBarrierHeight() # also converts ArrheniusEP to Arrhenius.
if self.pressureDependence and reaction.isUnimolecular():
# If this is going to be run through pressure dependence code,
# we need to make sure the barrier is positive.
reaction.fixBarrierHeight(forcePositive=True)
# Update unimolecular (pressure dependent) reaction networks
if self.pressureDependence:
# Recalculate k(T,P) values for modified networks
self.updateUnimolecularReactionNetworks()
logging.info('')
# Check new core and edge reactions for Chemkin duplicates
# The same duplicate reaction gets brought into the core
# at the same time, so there is no danger in checking all of the edge.
newCoreReactions = self.core.reactions[numOldCoreReactions:]
newEdgeReactions = self.edge.reactions[numOldEdgeReactions:]
checkedReactions = self.core.reactions[:numOldCoreReactions] + self.edge.reactions[:numOldEdgeReactions]
from rmgpy.chemkin import markDuplicateReaction
for rxn in newCoreReactions:
markDuplicateReaction(rxn, checkedReactions)
checkedReactions.append(rxn)
if self.saveEdgeSpecies:
for rxn in newEdgeReactions:
markDuplicateReaction(rxn, checkedReactions)
checkedReactions.append(rxn)
self.printEnlargeSummary(
newCoreSpecies=self.core.species[numOldCoreSpecies:],
newCoreReactions=self.core.reactions[numOldCoreReactions:],
reactionsMovedFromEdge=reactionsMovedFromEdge,
newEdgeSpecies=self.edge.species[numOldEdgeSpecies:],
newEdgeReactions=self.edge.reactions[numOldEdgeReactions:],
reactEdge=reactEdge,
)
logging.info('')
def addNewSurfaceObjects(self,obj,newSurfaceSpecies,newSurfaceReactions,reactionSystem):
"""
obj is the list of objects for enlargement coming from simulate
newSurfaceSpecies and newSurfaceReactions are the current lists of surface species and surface reactions
following simulation
reactionSystem is the current reactor
manages surface species and reactions being moved to and from the surface
moves them to appropriate newSurfaceSpc/RxnsAdd/loss sets
returns false if the surface has changed
"""
surfSpcs = set(self.surface.species)
surfRxns = set(self.surface.reactions)
newSurfaceSpecies = set(newSurfaceSpecies)
newSurfaceReactions = set(newSurfaceReactions)
addedRxns = {k for k in obj if isinstance(k,Reaction)}
addedSurfaceRxns = newSurfaceReactions - surfRxns
addedBulkRxns = addedRxns-addedSurfaceRxns
lostSurfaceRxns = (surfRxns - newSurfaceReactions) | addedBulkRxns
addedSpcs = {k for k in obj if isinstance(k,Species)} | {k.getMaximumLeakSpecies(reactionSystem.T.value_si, reactionSystem.P.value_si) for k in obj if isinstance(k,PDepNetwork)}
lostSurfaceSpcs = (surfSpcs-newSurfaceSpecies) | addedSpcs
addedSurfaceSpcs = newSurfaceSpecies - surfSpcs
self.newSurfaceSpcsAdd = self.newSurfaceSpcsAdd | addedSurfaceSpcs
self.newSurfaceRxnsAdd = self.newSurfaceRxnsAdd | addedSurfaceRxns
self.newSurfaceSpcsLoss = self.newSurfaceSpcsLoss | lostSurfaceSpcs
self.newSurfaceRxnsLoss = self.newSurfaceRxnsLoss | lostSurfaceRxns
return not (self.newSurfaceRxnsAdd != set() or self.newSurfaceRxnsLoss != set() or self.newSurfaceSpcsLoss != set() or self.newSurfaceSpcsAdd != set())
def adjustSurface(self):
"""
Here we add species intended to be added and remove any species that need to be moved out of the core.
For now we remove reactions from the surface that have become part of a PDepNetwork by
intersecting the set of surface reactions with the core so that all surface reactions are in the core
thus the surface algorithm currently (June 2017) is not implemented for pdep networks
(however it will function fine for non-pdep reactions on a pdep run)
"""
self.surface.species = list(((set(self.surface.species) | self.newSurfaceSpcsAdd)-self.newSurfaceSpcsLoss) & set(self.core.species))
self.surface.reactions = list(((set(self.surface.reactions) | self.newSurfaceRxnsAdd)-self.newSurfaceRxnsLoss) & set(self.core.reactions))
self.clearSurfaceAdjustments()
def clearSurfaceAdjustments(self):
"""
empties surface tracking varaibles
"""
self.newSurfaceSpcsAdd = set()
self.newSurfaceRxnsAdd = set()
self.newSurfaceSpcsLoss = set()
self.newSurfaceRxnsLoss = set()
def processNewReactions(self, newReactions, newSpecies, pdepNetwork=None, generateThermo=True):
"""
Process a list of newly-generated reactions involving the new core
species or explored isomer `newSpecies` in network `pdepNetwork`.
Makes a reaction and decides where to put it: core, edge, or PDepNetwork.
"""
for rxn in newReactions:
rxn, isNew = self.makeNewReaction(rxn, generateThermo=generateThermo)
if rxn is None:
# Skip this reaction because there was something wrong with it
continue
if isNew:
# We've made a new reaction, so make sure the species involved
# are in the core or edge
allSpeciesInCore = True
# Add the reactant and product species to the edge if necessary
# At the same time, check if all reactants and products are in the core
for spec in rxn.reactants:
if spec not in self.core.species:
allSpeciesInCore = False
if spec not in self.edge.species:
self.addSpeciesToEdge(spec)
for spec in rxn.products:
if spec not in self.core.species:
allSpeciesInCore = False
if spec not in self.edge.species:
self.addSpeciesToEdge(spec)
isomerAtoms = sum([len(spec.molecule[0].atoms) for spec in rxn.reactants])
# Decide whether or not to handle the reaction as a pressure-dependent reaction
pdep = True
if not self.pressureDependence:
# The pressure dependence option is turned off entirely
pdep = False
elif self.pressureDependence.maximumAtoms is not None and self.pressureDependence.maximumAtoms < isomerAtoms:
# The reaction involves so many atoms that pressure-dependent effects are assumed to be negligible
pdep = False
elif not (rxn.isIsomerization() or rxn.isDissociation() or rxn.isAssociation()):
# The reaction is not unimolecular in either direction, so it cannot be pressure-dependent
pdep = False
elif isinstance(rxn,LibraryReaction):
# Try generating the high pressure limit kinetics. If successful, set pdep to ``True``, and vice versa.
pdep = rxn.generate_high_p_limit_kinetics()
# If pressure dependence is on, we only add reactions that are not unimolecular;
# unimolecular reactions will be added after processing the associated networks
if not pdep:
if not isNew:
# The reaction is not new, so it should already be in the core or edge
continue
if allSpeciesInCore:
self.addReactionToCore(rxn)
else:
self.addReactionToEdge(rxn)
else:
# Add the reaction to the appropriate unimolecular reaction network
# If pdepNetwork is not None then that will be the network the
# (path) reactions are added to
# Note that this must be done even with reactions that are not new
# because of the way partial networks are explored
# Since PDepReactions are created as irreversible, not doing so
# would cause you to miss the reverse reactions!
self.addReactionToUnimolecularNetworks(rxn, newSpecies=newSpecies, network=pdepNetwork)
if isinstance(rxn, LibraryReaction):
# If reaction came from a reaction library, omit it from the core and edge so that it does
# not get double-counted with the pdep network
if rxn in self.core.reactions:
self.core.reactions.remove(rxn)
if rxn in self.edge.reactions:
self.edge.reactions.remove(rxn)
def applyThermoToSpecies(self, procnum):
"""
Generate thermo for species. QM calculations are parallelized if requested.
"""
from rmgpy.rmg.input import getInput
quantumMechanics = getInput('quantumMechanics')
if quantumMechanics:
quantumMechanics.runJobs(self.newSpeciesList, procnum=procnum)
# Serial thermo calculation for other methods
for spc in self.newSpeciesList:
self.generateThermo(spc, rename=True)
def generateThermo(self, spc, rename=False):
"""
Generate thermo for species.
"""
if not spc.thermo:
submit(spc, self.solventName)
if rename and spc.thermo and spc.thermo.label != '': # check if thermo libraries have a name for it
logging.info('Species {0} renamed {1} based on thermo library name'.format(spc.label, spc.thermo.label))
spc.label = spc.thermo.label
spc.generateEnergyTransferModel()
def applyKineticsToReaction(self, reaction):
"""
retrieve the best kinetics for the reaction and apply it towards the forward
or reverse direction (if reverse, flip the direaction).
"""
from rmgpy.data.rmg import getDB
# Find the reaction kinetics
kinetics, source, entry, isForward = self.generateKinetics(reaction)
# Flip the reaction direction if the kinetics are defined in the reverse direction
if not isForward:
family = getDB('kinetics').families[reaction.family]
reaction.reactants, reaction.products = reaction.products, reaction.reactants
reaction.pairs = [(p,r) for r,p in reaction.pairs]
if family.ownReverse and hasattr(reaction,'reverse'):
if reaction.reverse:
reaction.template = reaction.reverse.template
# replace degeneracy
reaction.degeneracy = reaction.reverse.degeneracy
# We're done with the "reverse" attribute, so delete it to save a bit of memory
reaction.reverse = None
reaction.kinetics = kinetics
def generateKinetics(self, reaction):
"""
Generate best possible kinetics for the given `reaction` using the kinetics database.
"""
# Only reactions from families should be missing kinetics
assert isinstance(reaction, TemplateReaction)
family = getFamilyLibraryObject(reaction.family)
# Get the kinetics for the reaction
kinetics, source, entry, isForward = family.getKinetics(reaction, templateLabels=reaction.template, degeneracy=reaction.degeneracy, estimator=self.kineticsEstimator, returnAllKinetics=False)
# Get the gibbs free energy of reaction at 298 K
G298 = reaction.getFreeEnergyOfReaction(298)
gibbsIsPositive = G298 > -1e-8
if family.ownReverse and hasattr(reaction,'reverse'):
if reaction.reverse:
# The kinetics family is its own reverse, so we could estimate kinetics in either direction
# First get the kinetics for the other direction
rev_kinetics, rev_source, rev_entry, rev_isForward = family.getKinetics(reaction.reverse, templateLabels=reaction.reverse.template, degeneracy=reaction.reverse.degeneracy, estimator=self.kineticsEstimator, returnAllKinetics=False)
# Now decide which direction's kinetics to keep
keepReverse = False
if (entry is not None and rev_entry is None):
# Only the forward has an entry, meaning an exact match in a depository or template
# the reverse must have used an averaged estimated node - so use forward.
reason = "This direction matched an entry in {0}, the other was just an estimate.".format(reaction.family)
elif (entry is None and rev_entry is not None):
# Only the reverse has an entry (see above) - use reverse.
keepReverse = True
reason = "This direction matched an entry in {0}, the other was just an estimate.".format(reaction.family)
elif (entry is not None and rev_entry is not None
and entry is rev_entry):
# Both forward and reverse have the same source and entry
# Use the one for which the kinetics is the forward kinetics
keepReverse = gibbsIsPositive and isForward and rev_isForward
reason = "Both directions matched the same entry in {0}, but this direction is exergonic.".format(reaction.family)
elif self.kineticsEstimator == 'group additivity' and (kinetics.comment.find("Fitted to 1 rate")>0
and not rev_kinetics.comment.find("Fitted to 1 rate")>0) :
# forward kinetics were fitted to only 1 rate, but reverse are hopefully better
keepReverse = True
reason = "Other direction matched a group only fitted to 1 rate."
elif self.kineticsEstimator == 'group additivity' and (not kinetics.comment.find("Fitted to 1 rate")>0
and rev_kinetics.comment.find("Fitted to 1 rate")>0) :
# reverse kinetics were fitted to only 1 rate, but forward are hopefully better
keepReverse = False
reason = "Other direction matched a group only fitted to 1 rate."
elif entry is not None and rev_entry is not None:
# Both directions matched explicit rate rules
# Keep the direction with the lower (but nonzero) rank
if entry.rank < rev_entry.rank and entry.rank != 0:
keepReverse = False
reason = "Both directions matched explicit rate rules, but this direction has a rule with a lower rank ({0} vs {1}).".format(entry.rank, rev_entry.rank)
elif rev_entry.rank < entry.rank and rev_entry.rank != 0:
keepReverse = True
reason = "Both directions matched explicit rate rules, but this direction has a rule with a lower rank ({0} vs {1}).".format(rev_entry.rank, entry.rank)
# Otherwise keep the direction that is exergonic at 298 K
else:
keepReverse = gibbsIsPositive and isForward and rev_isForward
reason = "Both directions matched explicit rate rules, but this direction is exergonic."
else:
# Keep the direction that is exergonic at 298 K
# This must be done after the thermo generation step
keepReverse = gibbsIsPositive and isForward and rev_isForward
reason = "Both directions are estimates, but this direction is exergonic."
if keepReverse:
kinetics = rev_kinetics
source = rev_source
entry = rev_entry
isForward = not rev_isForward
G298 = -G298
if self.verboseComments:
kinetics.comment += "\nKinetics were estimated in this direction instead of the reverse because:\n{0}".format(reason)
kinetics.comment += "\ndGrxn(298 K) = {0:.2f} kJ/mol".format( G298 / 1000.)
# The comments generated by the database for estimated kinetics can
# be quite long, and therefore not very useful
# We don't want to waste lots of memory storing these long,
# uninformative strings, so here we replace them with much shorter ones
if not self.verboseComments:
# Only keep a short comment (to save memory)
if 'Exact' in kinetics.comment:
# Exact match of rate rule
pass
elif 'Matched reaction' in kinetics.comment:
# Stems from matching a reaction from a depository
pass
else:
# Estimated (averaged) rate rule
kinetics.comment = kinetics.comment[kinetics.comment.find('Estimated'):]
return kinetics, source, entry, isForward
def printEnlargeSummary(self, newCoreSpecies, newCoreReactions, newEdgeSpecies, newEdgeReactions, reactionsMovedFromEdge=None, reactEdge=False):
"""
Output a summary of a model enlargement step to the log. The details of
the enlargement are passed in the `newCoreSpecies`, `newCoreReactions`,
`newEdgeSpecies`, and `newEdgeReactions` objects.
"""
logging.info('')
if reactEdge:
logging.info('Summary of Secondary Model Edge Enlargement')
else:
logging.info('Summary of Model Enlargement')
logging.info('---------------------------------')
logging.info('Added {0:d} new core species'.format(len(newCoreSpecies)))
for spec in newCoreSpecies:
display(spec)
logging.info(' {0}'.format(spec))
logging.info('Created {0:d} new edge species'.format(len(newEdgeSpecies)))
for spec in newEdgeSpecies:
display(spec)
logging.info(' {0}'.format(spec))
if reactionsMovedFromEdge:
logging.info('Moved {0:d} reactions from edge to core'.format(len(reactionsMovedFromEdge)))
for rxn in reactionsMovedFromEdge:
for r in newCoreReactions:
if ((r.reactants == rxn.reactants and r.products == rxn.products) or
(r.products == rxn.reactants and r.reactants == rxn.products)):
logging.info(' {0}'.format(r))
newCoreReactions.remove(r)
break
logging.info('Added {0:d} new core reactions'.format(len(newCoreReactions)))
for rxn in newCoreReactions:
logging.info(' {0}'.format(rxn))
logging.info('Created {0:d} new edge reactions'.format(len(newEdgeReactions)))
for rxn in newEdgeReactions:
logging.info(' {0}'.format(rxn))
coreSpeciesCount, coreReactionCount, edgeSpeciesCount, edgeReactionCount = self.getModelSize()
# Output current model size information after enlargement
logging.info('')
logging.info('After model enlargement:')
logging.info(' The model core has {0:d} species and {1:d} reactions'.format(coreSpeciesCount, coreReactionCount))
logging.info(' The model edge has {0:d} species and {1:d} reactions'.format(edgeSpeciesCount, edgeReactionCount))
logging.info('')
def addSpeciesToCore(self, spec):
"""
Add a species `spec` to the reaction model core (and remove from edge if
necessary). This function also moves any reactions in the edge that gain
core status as a result of this change in status to the core.
If this are any such reactions, they are returned in a list.
"""
assert spec not in self.core.species, "Tried to add species {0} to core, but it's already there".format(spec.label)
forbidden_structures = getDB('forbidden')
# check RMG globally forbidden structures
if not spec.explicitlyAllowed and forbidden_structures.isMoleculeForbidden(spec.molecule[0]):