-
Notifications
You must be signed in to change notification settings - Fork 3
/
ibfm.py
1477 lines (1400 loc) · 50.3 KB
/
ibfm.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
'''ibfm.py
Inherent Behavior in Functional Models
Author: Matthew G McIntire
2016
'''
#These flags can be changed in the file or at runtime after import ibfm
track_states = False
printWarnings = False
print_iterations = False
print_scenarios = False
run_parallel = False
n_workers = 3
from math import inf
from time import time
from glob import glob
import networkx as nx
import re
#import multiprocessing
import pickle
class InconsistencyError(Exception):
pass
class ModelError(Exception):
pass
def resetClock():
'''Reset the global clock variable to zero for a new simulation.'''
global last_clock,clock
last_clock = 0
clock = 0
resetClock()
def all_subclasses(cls):
'''Return all subclasses of a class recursively.
Written by Vebjorn Ljosa on Stack Overflow
'''
return cls.__subclasses__() + [g for s in cls.__subclasses__()
for g in all_subclasses(s)]
def getSubclass(cls,name):
'''Return the subclass of cls named name'''
for c in all_subclasses(cls):
if c.__name__ == name:
return c
class State(float):
'''Ordinal type for qualitatively representing flow values'''
def __new__(cls,value=None,flow=None):
'''Return a new State object.
Keyword arguments:
value=None -- If State() is called, rather than a subclass, value will be
used to determine the appropriate subclass
'''
if cls is State:
for subclass in State.__subclasses__():
if subclass.value == value:
cls = subclass
break
else:
lowest_value = 0
highest_value = 0
for subclass in State.__subclasses__():
if subclass.value < lowest_value:
lowest_value = subclass.value
lowest_cls = subclass
if subclass.value > highest_value:
highest_value = subclass.value
highest_cls = subclass
if value > highest_value:
cls = highest_cls
value = highest_value
elif value < lowest_value:
cls = lowest_cls
value = lowest_value
else:
raise Exception('Undefined State value')
return float.__new__(cls,cls.value)
def __init__(self,value=None,flow=None):
self.flow = flow
def __repr__(self):
return self.__class__.__qualname__
def __str__(self):
return self.__class__.__qualname__
@staticmethod
def getMethod(string):
'''Return a lambda method that returns the state specified in string.'''
state_class = getSubclass(State,string)
try:
state = state_class()
return lambda: [state]
except TypeError:
return None
@classmethod
def getUnaryMethod(cls,string):
'''Return the appropriate single-argument operator specified by string.'''
if string == 'effort':
return cls.setValueToEffort
if string == 'rate':
return cls.setValueToRate
if string == 'max':
return cls._max
if string == 'min':
return cls._min
if string == '++':
return cls._increment
if string == '--':
return cls._decrement
if string == 'invert':
return cls._inverse
if string == 'any':
return cls._any
if string == 'all':
return cls._all
if string == 'not':
return cls._not
return None
@classmethod
def getBinaryMethod(cls,string):
'''Return the appropriate double-argument operator specified by string.'''
if string == ',':
return cls._combine
if string == '*':
return cls._times
if string == '==':
return cls._eq
if string == '!=':
return cls._neq
if string == '>=':
return cls._geq
if string == '<=':
return cls._leq
if string == '<':
return cls._lt
if string == '>':
return cls._gt
if string == 'and':
return cls._and
if string == 'or':
return cls._or
return None
@classmethod
def getSetMethod(cls,string):
'''Return the appropriate method for setting either the effort or rate of
the flow specified during simulation.'''
if string == 'effort':
return cls.setEffort
elif string == 'rate':
return cls.setRate
@staticmethod
def setValueToEffort(states):
'''Set the value of each state in states to the effort of its flow'''
return [State(state.flow.effort,state.flow) for state in states]
@staticmethod
def setValueToRate(states):
'''Set the value of each state in states to the rate of its flow'''
return [State(state.flow.rate,state.flow) for state in states]
@staticmethod
def _max(states):
value = max(states)
return [state for state in states if state == value]
@staticmethod
def _min(states):
value = min(states)
return [state for state in states if state == value]
@staticmethod
def _increment(states):
return [State(state+1,state.flow) for state in states]
@staticmethod
def _decrement(states):
return [State(state-1,state.flow) for state in states]
@staticmethod
def _inverse(states):
for i,state in enumerate(states):
if state == Zero():
states[i] = Highest(flow = state.flow)
elif state == Low():
states[i] = High(flow = state.flow)
elif state >= High():
states[i] = Low(flow = state.flow)
return states
@staticmethod
def _any(x):
return [any(x)]
@staticmethod
def _all(x):
return [all(x)]
@staticmethod
def _not(xs):
return [not x for x in xs]
@staticmethod
def _combine(x1,x2):
return x1+x2
@staticmethod
def _times(x1s,x2s):
if len(x1s) == 1:
if x1s[0].flow == None:
return [State(x1s[0]*x2,x2.flow) for x2 in x2s]
if all([x2.flow == None for x2 in x2s]):
return [State(x1s[0]*x2,x1s[0].flow) for x2 in x2s]
return [State(x1s[0]*x2) for x2 in x2s]
if len(x2s) == 1:
if x2s[0].flow == None:
return [State(x2s[0]*x1,x1.flow) for x1 in x1s]
if all([x1.flow == None for x1 in x1s]):
return [State(x2s[0]*x1,x2s[0].flow) for x1 in x1s]
return [State(x2s[0]*x1) for x1 in x1s]
if len(x1s) == len(x2s):
return [State(x1*x2) for x1,x2 in zip(x1s, x2s)]
raise Exception('Unforseen multiplication case!')
@staticmethod
def _eq(x1s,x2s):
if len(x2s) == 1:
return [x2s[0] == x1 for x1 in x1s]
if len(x1s) == 1:
return [x1s[0] == x2 for x2 in x2s]
if len(x1s) == len(x2s):
return [x1 == x2 for x1,x2 in zip(x1s,x2s)]
raise Exception('Unforseen eq case')
@staticmethod
def _neq(x1s,x2s):
if len(x2s) == 1:
return [x2s[0] != x1 for x1 in x1s]
if len(x1s) == 1:
return [x1s[0] != x2 for x2 in x2s]
if len(x1s) == len(x2s):
return [x1 != x2 for x1,x2 in zip(x1s,x2s)]
raise Exception('Unforseen neq case')
@staticmethod
def _geq(x1s,x2s):
if len(x2s) == 1:
return [x1 >= x2s[0] for x1 in x1s]
if len(x1s) == 1:
return [x1s[0] >= x2 for x2 in x2s]
if len(x1s) == len(x2s):
return [x1 >= x2 for x1,x2 in zip(x1s,x2s)]
raise Exception('Unforseen geq case')
@staticmethod
def _leq(x1s,x2s):
if len(x2s) == 1:
return [x1 <= x2s[0] for x1 in x1s]
if len(x1s) == 1:
return [x1s[0] <= x2 for x2 in x2s]
if len(x1s) == len(x2s):
return [x1 <= x2 for x1,x2 in zip(x1s,x2s)]
raise Exception('Unforseen leq case')
@staticmethod
def _gt(x1s,x2s):
if len(x2s) == 1:
return [x1 > x2s[0] for x1 in x1s]
if len(x1s) == 1:
return [x1s[0] > x2 for x2 in x2s]
if len(x1s) == len(x2s):
return [x1 > x2 for x1,x2 in zip(x1s,x2s)]
raise Exception('Unforseen gt case')
@staticmethod
def _lt(x1s,x2s):
if len(x2s) == 1:
return [x1 < x2s[0] for x1 in x1s]
if len(x1s) == 1:
return [x1s[0] < x2 for x2 in x2s]
if len(x1s) == len(x2s):
return [x1 < x2 for x1,x2 in zip(x1s,x2s)]
raise Exception('Unforseen lt case')
@staticmethod
def _and(x1s,x2s):
if len(x2s) == 1:
return [x1 and x2s[0] for x1 in x1s]
if len(x1s) == 1:
return [x1s[0] and x2 for x2 in x2s]
if len(x1s) == len(x2s):
return [x1 and x2 for x1,x2 in zip(x1s,x2s)]
raise Exception('Unforseen and case')
@staticmethod
def _or(x1s,x2s):
if len(x2s) == 1:
return [x1 or x2s[0] for x1 in x1s]
if len(x1s) == 1:
return [x1s[0] or x2 for x2 in x2s]
if len(x1s) == len(x2s):
return [x1 or x2 for x1,x2 in zip(x1s,x2s)]
raise Exception('Unforseen or case')
@staticmethod
def setRate(lhs,rhs):
value = rhs[0]
for state in rhs[1:]:
if value != state:
print(rhs)
print([state for state in rhs])
raise InconsistencyError()
for state in lhs:
state.flow.setRate(value)
@staticmethod
def setEffort(lhs,rhs):
value = rhs[0]
for state in rhs[1:]:
if value != state:
print(rhs)
print([state for state in rhs])
raise InconsistencyError()
for state in lhs:
state.flow.setEffort(value)
############################# States #######################################
class Negative(State):
value = -1
class Zero(State):
value = 0
class Low(State):
value = 1
class Nominal(State):
value = 2
class High(State):
value = 3
class Highest(State):
value = 4
class ModeHealth(object):
pass
class Operational(ModeHealth):
pass
class Degraded(ModeHealth):
pass
class Failed(ModeHealth):
pass
class ModeConditionParent(object):
'''Superclass for the Mode and Condition classes.
Contains methods used by both Mode and Condition objects.
'''
def stack(self,x):
'''Take a list of strings and functions and folds it into a single function.
Recursively identifies keywords in x as operators in order to construct a
single function out of an ibfm expression in mode and condition definitions.
'''
# Check if done
#if len(x) == 0:
# return
if len(x) == 1:
if callable(x[0]):
return x[0]
else:
raise Exception('Unexpected: '+str(x[0])+' in '+self.__class__.__name__)
# Collapse parentheses
count = 0
for i,element in enumerate(x):
if element == '(':
count = count + 1
if count == 1:
i1 = i
elif element == ')':
count = count - 1
if count == 0:
return self.stack(x[:i1]+[self.stack(x[i1+1:i])]+x[i+1:])
if count:
raise Exception('Mismatched parenthesis in ' + self.__class__.__name__)
# Stack Unary Operators from both ends
for i in [-1,0]:
f = State.getUnaryMethod(x[i])
if f:
x.pop(i)
g = self.stack(x)
if not callable(g):
raise Exception('Unforseen stack case: '+str(x))
return self.getLambda(f,g)
# Stack binary operators
for i in [-2,1]:
f = State.getBinaryMethod(x[i])
if f:
g = self.stack(x[:i])
h = self.stack(x[i+1:])
return self.getLambda(f,g,h)
raise Exception('Unforseen stack case: '+str(x))
def getLambda(self,f=None,g=None,h=None,arg=None):
'''A lambda factory to ensure proper argument closure.'''
if arg:
if f:
return lambda: f(arg)
else:
return lambda: arg
if h:
return lambda: f(g(),h())
if g:
return lambda: f(g())
raise Exception('Unforseen getLambda')
def flowAndStateMethods(self,behavior,optional=False):
'''Replaces ibfm references to states and flows with the appropriate objects.
Behavior is a list of strings, some of which refer to states and flows.
Each reference to a state or flow is replaced by a method which returns the
state or flow refered to.
'''
# Replace strings with flow and state object methods
with_methods = []
flow_class = None
flow = None
for element in behavior:
if flow_class: #If a flow class is already identified
if element == 'input':
flow = self.in_flow[flow_class]
elif element == 'output':
flow = self.out_flow[flow_class]
else:
raise Exception('Expected "input" or "output" following '+
flow_class.__name__+' in '+str(self)+' '+
self.__class__.__name__+' definition.')
if flow == None:
raise ModelError('Expected '+flow_class.__name__+' '+element+' in '+
self.function.name)
flow_class = None
elif flow: #If a flow class and its direction have been identified
if element == 'effort':
f = self.getLambda(f=State.setValueToEffort,arg=flow)
elif element == 'rate':
f = self.getLambda(f=State.setValueToRate,arg=flow)
else:
raise Exception('Expected "effort" or "rate" following the input/output of '+
flow_class.__name__+' in '+str(self)+' '+
self.__class__.__name__+' definition.')
with_methods.append(f)
flow = None
else:
#Test for references to a state
f = State.getMethod(element)
if f:
with_methods.append(f)
else:
#Test for references to a flow
flow_class = Flow._subclasses.get(element)
if not flow_class:
if element in ['input','output']:
raise Exception(element+' not following flow type in '+str(self)+
' '+self.__class__.__name__+' definition.')
#Everything else gets passed through unchanged.
with_methods.append(element)
if flow:
#The effort/rate distinction has already been identifed in self.behavior(s)
with_methods.append(self.getLambda(arg=flow))
return with_methods
class Mode(ModeConditionParent):
'''Class for operational modes that functions may use.
'''
_subclasses = {}
def __init__(self,name,function,health,prob,when,cost, **attr):
'''Return a Mode object.
Required arguments:
name -- a unique name (for easier to read function definitions)
function -- the object the mode belongs to (for access to its flows),
health -- a ModeHealth object designating the health of the function
represented by the mode.
'''
self.name = name
self.function = function
self.out_flow = function.out_flow
self.in_flow = function.in_flow
self.health = health
self.prob = prob
self.when = when
self.attr = attr
self.cost = cost
def __repr__(self):
return self.__class__.__qualname__
def __hash__(self):
'''Return hash to identify unique Mode objects.
Used by dictionaries in NetworkX.
'''
return hash((self.__class__,self.health))
def __eq__(self,other):
'''Return boolean to identify unique Mode objects.
Used by dictionaries in NetworkX.
'''
return self.__class__ == other.__class__ and self.health == other.health
def createSetMethod(self,setMethod,lhs,rhs):
if not callable(lhs) or not callable(rhs) or not callable(setMethod):
#print(lhs)
#print(rhs)
#print(setMethod)
#print(self)
raise Exception('Bad creatSetMethod call')
return lambda: setMethod(lhs(),rhs())
def behaviors(self,mode=None):
'''Yield Behavior methods.
Runs recursively (but in a different subclass) when using the 'import'
keyword.
'''
if mode == None:
mode = self.__class__
for behavior in mode._behaviors:
optional = False
if 'optional' == behavior[0].lower():
optional = True
behavior = behavior[1:]
elif 'import' == behavior[0].lower():
from_mode = Mode._subclasses[behavior[1]]
if from_mode:
yield from from_mode.behaviors(self,from_mode)
else:
raise Exception(behavior[1]+' is not a defined mode. '+str(self)+
' tried to use behaviors from it.')
continue
# Separate the left-hand side from the right-hand side of the assignment
assignment = behavior.index('=')
lhs = behavior[:assignment]
rhs = behavior[assignment+1:]
# Get value setter method
setMethod = State.getSetMethod(lhs.pop())
# Replace flows and states with appropriate methods
try:
lhs = self.flowAndStateMethods(lhs,optional)
rhs = self.flowAndStateMethods(rhs,optional)
except KeyError:
if optional:
continue
else:
raise
# Replace everything else
lhs = self.stack(lhs)
rhs = self.stack(rhs)
yield self.createSetMethod(setMethod,lhs,rhs)
def textBehaviors(self,mode=None):
if mode == None:
mode = self
for behavior in mode._behaviors:
optional = 'required'
if 'optional' == behavior[0].lower():
optional = 'optional'
behavior = behavior[1:]
elif 'from' == behavior[0].lower():
from_Mode = Mode._subclasses[behavior[1]]
if from_Mode:
yield from from_Mode.textBehaviors(self,from_Mode)
else:
print(behavior)
raise Exception(behavior[1]+' is not a defined mode. '+str(self)+
' tried to use behaviors from it.')
continue
ins = []
outs = []
cls = getSubclass(Behavior,behavior[0])
if not cls:
raise Exception(behavior[0]+' is not a defined behavior.')
try:
for word in behavior[1:]:
if word.lower() in 'inout':
entry = word.lower()
elif entry == 'in':
ins.append(word)
elif entry == 'out':
outs.append(word)
else:
raise Exception('Looking for keywords in or out, found: '+word)
yield (behavior[0],optional,ins,outs)
except KeyError:
if not optional:
raise
def reset(self):
'''Reset the Mode object for a new simulation.
Required because Mode objects share NetworkX graphs with Condition
objects, which have their timers reset, and this is easier than generating
a different iterator for only Condition objects.
'''
pass
class Condition(ModeConditionParent):
'''Class for a condition to advance to another mode.
'''
_subclasses = {}
def __init__(self,function,delay=0,logical_not=False):
'''Return a Condition object
Required arguments:
function -- the object it belongs to (for access to its flows)
delay -- the amound of time the condition must be met for a mode change.
logical_not -- boolean flag indicating whether the condition should be
logically negated.
'''
self.out_flow = function.out_flow
self.in_flow = function.in_flow
self.delay = delay
self.reset()
if logical_not:
self.test = lambda: not self.behavior()
else:
self.test = self.behavior()
def behavior(self,condition=None):
if condition == None:
condition = self.__class__
if len(condition._behaviors) != 1:
raise Exception('Conditions must have exactly one test behavior! '+
condition.__name__+' does not.')
behavior = condition._behaviors[0]
if 'import' == behavior[0].lower():
from_condition = Mode._subclasses[behavior[1]]
if from_condition:
return from_condition.behavior(self,from_condition)
else:
raise Exception(behavior[1]+' is not a defined condition. '+str(self)+
' tried to use the behavior from it.')
behavior = self.flowAndStateMethods(behavior)
test = self.stack(behavior)
return lambda: test()[0]
def reset(self):
'''Resets the delay timer.
'''
self.timer = inf
self.time = clock
def time_remaining(self):
'''Returns the time remaining before mode change.
Calculates the time under current conditions before the condition is
met. Returns 0 if the condition test has been met for self.delay amount of
time, inf if the condition test has not been met, and the remaining time
otherwise.
'''
if last_clock != self.time and clock != self.time:
self.timer = inf
self.time = clock
if self.test(): #This is where the actual test method is called.
self.timer = min(self.timer,self.delay+clock)
else:
self.timer = inf
return self.timer
class Function(object):
'''Abstract class for functions in the functional model.
Required methods:
construct(self) -- call self.addMode(name,health,mode_class,prob, default=False) and
self.addCondition(source_modes,condition,next_mode,delay=0)
repeatedly to specify all of the modes and conditions
attainable by the function. The default default is the
first given mode with health=Operational.
'''
_subclasses = {}
def __init__(self,model,name=None,allow_faults=True,**attr):
'''Return a Function object
Required arguments:
name -- a unique name for use in defining flows
allow_faults -- a flag to allow degraded and failure modes to be tested in
experiments. If False, off-nominal modes may still be
entered conditionally during simulation.
'''
self.allow_faults = allow_faults
self.attr = attr
self.default = None
#The graph containing the conditional relationships between the modes
self.condition_graph = nx.DiGraph()
#The graph containing the behaviors enacted by each mode
self.behavior_graph = nx.DiGraph()
self.in_flow = {}
self.out_flow = {}
self.all_flows = []
self.modes = []
if name != None:
if name not in model.names:
model.names.append(name)
else:
raise Exception(name+' already used as a function name.')
else:
for n in range(1,10):
name = self.__class__.__qualname__+format(n,'01d')
if name not in model.names:
model.names.append(name)
break
else:
raise Exception('Too many ' +self.__class__.__qualname__+' functions. '+
'Raise the limit.')
self.name = name
def __str__(self):
return self.name
def __repr__(self):
return self.name
def __hash__(self):
'''Return hash to identify unique Function objects.
Used by dictionaries in NetworkX.
'''
return hash(self.name)
def __lt__(self,other):
return str(self)<str(other)
def __eq__(self,other):
'''Return boolean to identify unique Function objects.
Used by dictionaries in NetworkX.
'''
return repr(self) == repr(other) or self.name == str(other)
def construct(self):
'''Build the modes and conditions for the function.
For functions defined in ibfm files.
'''
for mode in self.__class__._modes:
ident = mode[0]
health = getSubclass(ModeHealth,mode[1])
mode_class = Mode._subclasses.get(mode[2])
try:
prob=mode[4]
except IndexError:
prob='NA'
try:
when=mode[5]
except IndexError:
when='NA'
try:
cost=mode[7]
except IndexError:
cost='NA'
if mode_class is None:
print(mode)
raise Exception(mode[2]+' is not a defined mode')
self.addMode(ident,health,mode_class, prob, when, cost)
for condition in self.__class__._conditions:
entry = None
delay = 0
source_modes = []
for word in condition:
if word.lower() == 'to':
entry = 'to'
elif word.lower() == 'delay':
entry = 'delay'
elif entry == 'to':
next_mode = word
entry = 'class'
elif entry == 'delay':
delay = float(word)
elif entry == 'class':
condition_class = Condition._subclasses.get(word)
if condition_class is None:
raise Exception(condition[0]+condition[1]+condition[2]+condition[3]+' is not a defined mode')
else:
source_modes.append(word)
#print(source_modes)
#print(next_mode)
#print(condition_class)
self.addCondition(source_modes,condition_class,next_mode,delay=delay)
def reset(self):
'''Reset all modes and set mode to default.'''
self.mode = self.default
for node in list(self.condition_graph.nodes()):
node.reset()
def addOutFlow(self,flow):
'''Attach an outflow flow to the function.
Also fills out a list of all flows in or out of the function.
'''
self._addFlow(flow,flow.__class__,self.out_flow)
self.all_flows.append(flow)
def addInFlow(self,flow):
'''Attach an inflow flow to the function.
Also fills out a list of all flows in or out of the function.
'''
self._addFlow(flow,flow.__class__,self.in_flow)
self.all_flows.append(flow)
def _addFlow(self,flow,flow_class,flows):
'''Add a flow to the flows dictionary (recursively).
Makes sure the flow is reachable by its class or any of its superclasses
as the key. Also fills out a list of all flows in or out of the function.
'''
previous = flows.get(flow_class)
if previous:
previous.append(flow)
else:
flows[flow_class] = [flow]
if Flow not in flow_class.__bases__:
for base in flow_class.__bases__:
self._addFlow(flow,base,flows)
def addMode(self,name,health,mode_class, prob, when, cost, default=False,**attr):
'''Add a mode to the function.
Required arguments:
name -- a unique identifier for the mode to be used in calls to
self.addCondition()
health -- a ModeHealth subclass designating the health of the function
when represented by the mode.
prob -- probability of failure mode
when -- when the failure mode occurs (outside of pseudo-time)
mode_class -- the Mode subclass representing the mode being added
default -- whether the mode should be the default for the function. The
default default is the first given mode with health=Operational.
'''
mode = mode_class(name,self,health(),prob,when,cost, **attr)
self.modes.append(mode)
if default or (self.default == None and health == Operational):
self.default = mode
self.condition_graph.add_node(mode)
try:
for behavior in mode.behaviors():
self.behavior_graph.add_edge(mode,behavior)
except KeyError as error:
raise Exception("{0} missing {1} flow.".format(self,error.args[0]))
def getMode(self,name):
'''Return the function mode with the given name, if it exists.'''
for mode in self.modes:
if mode.name == name:
return mode
def addCondition(self,source_modes,condition,next_mode,delay=0):
'''Add a conditional change to the function from one mode to another.
Required arguments:
source_modes -- a list of names of modes from which the condition is applied.
condition -- the Condition subclass representing the condition being added.
next_mode -- the name of the mode assigned to the function should be
condition be satisfied.
'''
condition = condition(self,delay)
if type(source_modes) is not list:
source_modes = [source_modes]
next_mode = self.getMode(next_mode)
self.condition_graph.add_edge(condition,next_mode)
for sourceMode in source_modes:
sourceMode = self.getMode(sourceMode)
self.condition_graph.add_edge(sourceMode,condition)
def step(self):
'''Evaluate the function.
Test each condition reachable from the current mode. If there is only one,
advance to its child mode. If there is more than one, fork the simulation,
advancing to each mode in a different fork. <------------------------------ IMPLEMENT THIS!
In any case, finish by applying the behavior defined by the current mode.
'''
transition = False
minimum_timer = inf
for condition in self.condition_graph.successors(self.mode):
timer = condition.time_remaining()
minimum_timer = min(minimum_timer,timer)
if timer <= clock:
if timer < clock:
raise Exception('Missed condition')
if not transition:
transition = True
self.mode = list(self.condition_graph.successors(condition))[0]
for behavior in self.behavior_graph.successors(self.mode):
try:
behavior()
except:
print(self)
print(self.mode)
raise
return minimum_timer
class Flow(object):
'''Superclass for flows in the functional model.'''
_subclasses = {}
def __init__(self,source,drain):
self.source = source
self.drain = drain
self.name = source.name+'_'+drain.name+'_'+self.__class__.__name__
self.reset()
self.flow = self #For State.setValueToEffort and setValueToFlow
def __repr__(self):
return self.__class__.__qualname__
def reset(self,effort=Zero(),rate=Zero()):
'''Set the effort and rate of the flow to Zero() unless otherwise specified.'''
self.effort = effort
self.rate = rate
self.effort_queue = None
self.rate_queue = None
def setEffort(self,value):
'''Set the effort of the flow to value. Queued until step(self) is called.'''
if printWarnings and not self.effort_queue is None:
print('Warning! Competing causality in '+self.name+' effort.')
self.effort_queue = value
def setRate(self,value):
'''Set the rate of the flow to value. Queued until step(self) is called.'''
if printWarnings and not self.rate_queue is None:
print('Warning! Competing causality in '+self.name+' rate.')
self.rate_queue = value
def step(self):
'''Resolve the effort and rate values in the flow.'''
changed = False
if self.effort != self.effort_queue and self.effort_queue != None:
self.effort = self.effort_queue
changed = True
if printWarnings and self.rate != self.rate_queue:
print('Warning! Overlapping causality in '+self.name)
if self.rate != self.rate_queue and self.rate_queue != None:
self.rate = self.rate_queue
changed =True
self.effort_queue = self.rate_queue = None
return changed
Flow._subclasses['Flow'] = Flow
class Model(object):
'''Class for functional models.
Replaceable methods:
construct(self) -- Call self.addFunction(function) and
self.addFlow(in_function_name,out_function_name) repeatedly
to describe the functions and flows that make up the
functional model.
'''
_subclasses = {}
def __init__(self,graph=None):
'''Construct the model and run it under nominal conditions.
Keyword Arguments:
graph -- a NetworkX graph representing the functional model
'''
#This graph contains all of the functions as nodes and flows as edges.
self.names = []
self.imported_graph = graph
self.graph = nx.MultiDiGraph()
self.functions = self.graph.nodes() #for code readability
self.construct()
self.connect()
self.reset()
self.run()
self.nominal_state = self.getState()
def construct(self):
'''Construct a model from an imported graph or model defined in a .ibfm file.
Replace this method when directly defining models as subclasses.
'''
if self.imported_graph:
#Find keys
function_key = None
flow_key = None
for _,data in list(self.imported_graph.nodes(data=True)):
for key in data:
if Function._subclasses.get(data[key]) is not None:
function_key = key
break
if function_key:
break
else:
raise Exception('No defined function names found.')
for _,_,data in list(self.imported_graph.edges(data=True)):
for key in data:
if Flow._subclasses.get(str(data[key])) is not None:
flow_key = key
break
if flow_key:
break
else:
raise Exception('No defined flow names found.')
#Build model
for node,data in list(self.imported_graph.nodes(data=True)):
function = Function._subclasses.get(str(data[function_key]))
if function is None:
raise Exception(data[function_key]+' is not a defined function name.')
self.addFunction(function(self,node), node)
for node1,node2,data in list(self.imported_graph.edges(data=True)):
flow_class = Flow._subclasses.get(str(data[flow_key]))
if flow_class is None:
raise Exception(data[flow_key]+' is not a defined flow name.')
self.addFlow(flow_class,node1,node2)
else: #The model was defined in a .ibfm file
for words in self.__class__._functions:
ident = words[0]
function = Function._subclasses[words[1]]
self.addFunction(function(self,ident),words[1])
for words in self.__class__._flows:
ident = words[:2]
flow = Flow._subclasses[words[2]]
self.addFlow(flow,ident[0],ident[1])
def flows(self,functions=False):
'''Generate flows for iterating.
NetworkX does not allow edges to be arbitrary objects; objects must be
stored as edge attributes.
'''
if functions:
for in_function,out_function,attr in list(self.graph.edges(data=True)):
yield (attr['attr_dict'][Flow],in_function,out_function)
else:
for _,_,attr in list(self.graph.edges(data=True)):
yield attr['attr_dict'][Flow]
def reset(self):
'''Reset the clock, all functions, and all flows.'''
resetClock()
for function in self.functions():
function.reset()
for flow in self.flows():
flow.reset()
def connect(self):
'''Finish initialization of each function.
Give each function handles to every flow connected to it, then run each
function's construct method to initialize its modes and conditions.