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
|
# This program is in the public domain
"""
Progress monitors.
Process monitors accept a :class:`bumps.history.History` object each cycle
and perform some sort of work.
Monitors have a :meth:`Monitor.config_history` method which calls
*history.requires()* to set the amount of history it needs and a
*Monitor.__call__* method which takes the updated history and
generates the monitor output.
Most monitors are subclassed from :class:`TimedUpdate` to set a minimum
time between updates and to only show updates when there is an improvement.
The *TimedUpdate* subclasses must override :meth:`TimedUpdate.show_progress`
and :meth:`TimedUpdate.show_improvement` to control the output form. History
must be updated with time, value, point and step. The
bumps.fitters.MonitorRunner class manages history and updates.
"""
__all__ = ["Monitor", "Logger", "TimedUpdate"]
from numpy import inf
class Monitor(object):
"""
Base class for monitors.
"""
def config_history(self, history):
"""
Indicate which fields are needed by the monitor and for what duration.
"""
pass
def __call__(self, history):
"""
Give the monitor a new piece of history to work with.
"""
pass
def _getfield(history, field):
"""
Return the last value in the trace, or None if there is no
last value or no trace.
"""
trace = getattr(history, field, [])
try:
return trace[0]
except IndexError:
return None
class Logger(Monitor):
"""
Keeps a record of all values for the desired fields.
*fields* is a list of history fields to store.
*table* is an object with a *store(field=value,...)* method, which gets
the current value of each field every time the history is updated.
Call :meth:`config_history` with the :class:`bumps.history.History`
object before starting so that the correct fields are stored.
"""
def __init__(self, fields=(), table=None):
self.fields = fields
self.table = table
def config_history(self, history):
"""
Make sure history records each logged field.
"""
kwargs = dict((key, 1) for key in self.fields)
history.requires(**kwargs)
def __call__(self, history):
"""
Record the next piece of history.
"""
record = dict((f, _getfield(history, f)) for f in self.fields)
self.table.store(**record)
class TimedUpdate(Monitor):
"""
Indicate progress every n seconds.
The process should provide time, value, point, and step to the
history update. Call :meth:`config_history` with the
:class:`bumps.history.History` object before starting so that
these fields are stored.
*progress* is the number of seconds to go before showing progress, such
as time or step number.
*improvement* is the number of seconds to go before showing
improvements to value.
By default, the update only prints step number and improved value.
Subclass TimedUpdate with replaced :meth:`show_progress` and
:meth:`show_improvement` to trigger GUI updates or show parameter
values.
"""
def __init__(self, progress=60, improvement=5):
self.progress_delta = progress
self.improvement_delta = improvement
self.progress_time = -inf
self.improvement_time = -inf
self.value = inf
self.improved = False
def config_history(self, history):
history.requires(time=1, value=1, point=1, step=1)
def show_improvement(self, history):
print("step", history.step, "value", history.value)
def show_progress(self, history):
pass
def __call__(self, history):
t = history.time[0]
v = history.value[0]
if v < self.value:
self.improved = True
if t > self.progress_time + self.progress_delta:
self.progress_time = t
self.show_progress(history)
if self.improved and t > self.improvement_time + self.improvement_delta:
self.improved = False
self.improvement_time = t
self.show_improvement(history)
|
