forked from windelbouwman/lognplot
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathchart.py
More file actions
141 lines (110 loc) · 4.21 KB
/
Copy pathchart.py
File metadata and controls
141 lines (110 loc) · 4.21 KB
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
import math
from .axis import Axis
from .curve import Curve
from ..utils import bench_it
from ..time import TimeSpan
from ..tsdb import Aggregation, Metrics
class Chart:
""" Chart object.
Note that this is only the datatype of the chart. All drawing
is performed in a seperate function. This only binds all the
axis, traces into a single object.
"""
def __init__(self, db):
self.x_axis = Axis()
self.y_axis = Axis()
self.curves = []
self.cursor = None
self.db = db
def has_curve(self, name):
for curve in self.curves:
if curve.name == name:
return True
return False
def add_curve(self, name, color):
if not self.has_curve(name):
curve = Curve(self.db, name, color)
self.curves.append(curve)
def clear_curves(self):
self.curves.clear()
def info(self):
print(f"Chart with {len(self.curves)} series")
for index, curve in enumerate(self.curves):
print(f"serie {index} with {len(curve)} samples")
def set_cursor(self, value):
""" Set cursor position onto this chart.
Use None to hide the cursor.
"""
self.cursor = value
def horizontal_zoom(self, amount, around):
""" Zoom in horizontal manner. """
self.x_axis.zoom(amount, around=around)
def vertical_zoom(self, amount):
self.y_axis.zoom(amount)
def horizontal_pan_relative(self, amount):
""" Pan a percentage of the current axis range. """
self.x_axis.pan_relative(amount)
def horizontal_pan_absolute(self, amount):
""" Pan horizontally by a certain amount. """
self.x_axis.pan_absolute(amount)
def vertical_pan_relative(self, amount):
self.y_axis.pan_relative(amount)
def autoscale_y(self):
""" Automatically adjust the Y-axis to fit data in range. """
timespan = TimeSpan(self.x_axis.minimum, self.x_axis.maximum)
summary = self.data_summary(timespan=timespan)
if summary:
self.fit_metrics_y_axis(summary.metrics)
def fit_metrics_y_axis(self, metric: Metrics):
""" Adjust Y-axis to fit metrics into view. """
domain = metric.maximum - metric.minimum
# If we have a single value, increase the domain.
if math.isclose(domain, 0):
domain = 1
minimum = metric.minimum - domain * 0.05
maximum = metric.maximum + domain * 0.05
self.y_axis.set_limits(minimum, maximum)
def fit_timespan_on_x_axis(self, timespan: TimeSpan):
""" Adjust X-axis to fit timespan in view. """
domain = timespan.end - timespan.begin
if math.isclose(domain, 0):
domain = 1
minimum = timespan.begin - domain * 0.05
maximum = timespan.end + domain * 0.05
self.x_axis.set_limits(minimum, maximum)
def get_region(self):
""" Get the current viewed region.
"""
return (
self.x_axis.minimum,
self.y_axis.minimum,
self.x_axis.maximum,
self.y_axis.maximum,
)
def zoom_fit(self):
""" Adjust axis to fit all curves. """
summary = self.data_summary()
# If we have metrics, adjust axis.
if summary:
self.fit_timespan_on_x_axis(summary.timespan)
self.fit_metrics_y_axis(summary.metrics)
def zoom_to_last(self, duration):
""" To to the last duration in view. """
summary = self.data_summary()
if summary:
end = summary.timespan.end
begin = end - duration
timespan = TimeSpan(begin, end)
self.fit_timespan_on_x_axis(timespan)
self.autoscale_y()
def data_summary(self, timespan=None) -> Aggregation:
""" Metrics of all signals in the plot. """
# Gather bounding boxes of all curves:
aggregations = []
for curve in self.curves:
aggregation = curve.query_summary(timespan=timespan)
if aggregation:
aggregations.append(aggregation)
# If we have bounds, merge them and adjust axis.
if aggregations:
return Aggregation.from_aggregations(aggregations)