-
Notifications
You must be signed in to change notification settings - Fork 3
Expand file tree
/
Copy pathpyutilib.py
More file actions
1870 lines (1571 loc) · 73.6 KB
/
pyutilib.py
File metadata and controls
1870 lines (1571 loc) · 73.6 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
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
from __future__ import division
from concurrent.futures import ThreadPoolExecutor
from functools import wraps
import datetime as dt
import gzip
import json
import re
import threading
import types
import warnings
import webbrowser
from asyncio_helpers import cancellable, sync
from debounce.async import Debounce
from dropbot import EVENT_ENABLE, EVENT_CHANNELS_UPDATED, EVENT_SHORTS_DETECTED
from flatland import Integer, Float, Form, Boolean
from flatland.validation import ValueAtLeast, ValueAtMost
from matplotlib.backends.backend_gtkagg import (FigureCanvasGTKAgg as
FigureCanvas)
from matplotlib.figure import Figure
from logging_helpers import _L #: .. versionadded:: 2.24
from microdrop.app_context import (get_app, get_hub_uri, MODE_RUNNING_MASK,
MODE_REAL_TIME_MASK)
from microdrop.interfaces import (IApplicationMode, IElectrodeActuator,
IPlugin, IWaveformGenerator)
from microdrop.plugin_helpers import (StepOptionsController, AppDataController,
hub_execute)
from microdrop.plugin_manager import (Plugin, implements, PluginGlobals,
ScheduleRequest, emit_signal,
get_service_instance_by_name)
from microdrop_utility.gui import yesno
from pygtkhelpers.gthreads import gtk_threadsafe
from pygtkhelpers.ui.dialogs import animation_dialog
from zmq_plugin.plugin import Plugin as ZmqPlugin
from zmq_plugin.schema import decode_content_data
import blinker
import dropbot as db
import dropbot.hardware_test
import dropbot.monitor
import dropbot.self_test
import dropbot.threshold
import gobject
import gtk
# XXX Use `json_tricks` rather than standard `json` to support serializing
# [Numpy arrays and scalars][1].
#
# [1]: http://json-tricks.readthedocs.io/en/latest/#numpy-arrays
import json_tricks
import markdown2pango
import numpy as np
import pandas as pd
import path_helpers as ph
import si_prefix as si
import tables
import trollius as asyncio
import zmq
from ._version import get_versions
from .noconflict import classmaker
from .execute import execute
from .status import DropBotStatusView
__version__ = get_versions()['version']
del get_versions
# Prevent warning about potential future changes to Numpy scalar encoding
# behaviour.
json_tricks.NumpyEncoder.SHOW_SCALAR_WARNING = False
# Ignore natural name warnings from PyTables [1].
#
# [1]: https://www.mail-archive.com/pytables-users@lists.sourceforge.net/msg01130.html
warnings.simplefilter('ignore', tables.NaturalNameWarning)
PluginGlobals.push_env('microdrop.managed')
def gtk_on_link_clicked(widget, uri):
'''
.. versionadded:: 2.38.1
Callback to workaround the following error:
GtkWarning: No application is registered as handling this file
This is a known issue, e.g., https://bitbucket.org/tortoisehg/hgtk/issues/1656/link-in-about-box-doesnt-work#comment-312511
'''
webbrowser.open_new_tab(uri)
return True
class DmfZmqPlugin(ZmqPlugin):
"""
API for adding/clearing droplet routes.
"""
def __init__(self, parent, *args, **kwargs):
self.parent = parent
super(DmfZmqPlugin, self).__init__(*args, **kwargs)
def check_sockets(self):
"""
Check for messages on command and subscription sockets and process
any messages accordingly.
.. versionchanged:: 2.25.1
Do not set actuated area according to electrode controller plugin
messages. Actuated area should be updated **_only once the DropBot
reports the channels have been actuated_**.
"""
try:
msg_frames = self.command_socket.recv_multipart(zmq.NOBLOCK)
except zmq.Again:
pass
else:
self.on_command_recv(msg_frames)
return True
def on_execute__measure_liquid_capacitance(self, request):
self.parent.on_measure_liquid_capacitance()
def on_execute__measure_filler_capacitance(self, request):
self.parent.on_measure_filler_capacitance()
def on_execute__find_liquid(self, request):
return self.parent.find_liquid()
def on_execute__identify_electrode(self, request):
data = decode_content_data(request)
self.parent.identify_electrode(data['electrode_id'])
def results_dialog(name, results, axis_count=1, parent=None):
'''
Given the name of a test and the corresponding results object, generate a
GTK dialog displaying:
- The formatted text output of the results
- The corresponding axis plot(s) (if applicable).
.. versionadded:: 0.14
Parameters
----------
name : str
Test name (e.g., ``voltage``, ``channels``, etc.).
results : dict
Results from one or more :module:`dropbot.self_test` tests.
axis_count : int, optional
The number of figure axes required for plotting.
parent : gtk.Window, optional
The parent window of the dialog.
This allows, for example, the dialog to be launched in front of the
parent window and to disable controls on the parent window until the
dialog is closed.
'''
# Resolve function for formatting results for specified test.
format_func = getattr(db.self_test, 'format_%s_results' % name)
try:
# Resolve function for plotting results for specified test (if
# available).
plot_func = getattr(db.self_test, 'plot_%s_results' % name)
except AttributeError:
plot_func = None
dialog = gtk.Dialog(parent=parent)
title = re.sub(r'^test_', '', name).replace('_', ' ').title()
dialog.set_title(title)
dialog.props.destroy_with_parent = True
dialog.props.window_position = gtk.WIN_POS_MOUSE
label = gtk.Label()
label.props.use_markup = True
message = format_func(results[name])
label.set_markup('<span face="monospace">{}</span>'.format(message))
content_area = dialog.get_content_area()
content_area.pack_start(label, fill=False, expand=False, padding=5)
# Allocate minimum of 150 pixels height for report text.
row_heights = [150]
if plot_func is not None:
# Plotting function is available.
fig = Figure()
canvas = FigureCanvas(fig)
if axis_count > 1:
# Plotting function plots to more than one axis.
axes = [fig.add_subplot(axis_count, 1, i + 1)
for i in range(axis_count)]
plot_func(results[name], axes=axes)
else:
# Plotting function plots to a single axis.
axis = fig.add_subplot(111)
plot_func(results[name], axis=axis)
# Allocate minimum of 300 pixels height for report text.
row_heights += axis_count * [300]
fig.tight_layout()
content_area.pack_start(canvas, fill=True, expand=True, padding=0)
# Allocate minimum pixels height based on the number of axes.
dialog.set_default_size(600, sum(row_heights))
content_area.show_all()
return dialog
def require_connection(log_level='error'):
'''
Decorator factory to require DropBot connection.
Parameters
----------
log_level : str
Level to log message to if DropBot is not connect.
Returns
-------
function
Decorator to require DropBot connection.
.. versionchanged:: 2.24
Convert to decorator factory to add optional log-level customization.
'''
def _require_connection(func):
@wraps(func)
def _wrapped(self, *args, **kwargs):
if not self.dropbot_connected.is_set():
logger = _L()
log_func = getattr(logger, log_level)
log_func('DropBot is not connected.')
else:
return func(self, *args, **kwargs)
return _wrapped
return _require_connection
def error_ignore(on_error=None):
'''
Generate decorator for ignoring errors.
Parameters
----------
on_error : str or callable, optional
Error message to log, or function to call if error occurs.
Returns
-------
function
Decorator function.
'''
def _decorator(func):
@wraps(func)
def _wrapped(*args, **kwargs):
try:
return func(*args, **kwargs)
except Exception as exception:
if isinstance(on_error, types.StringTypes):
print(on_error)
elif callable(on_error):
on_error(exception, func, *args, **kwargs)
else:
pass
return _wrapped
return _decorator
def require_test_board(func):
'''
Decorator to prompt user to insert DropBot test board.
.. versionchanged:: 0.16
.. versionchanged:: 2.25
Add clickable hyperlink to DropBot Test Board documentation.
.. versionchanged:: 2.35
Set default focus to OK button.
'''
@wraps(func)
def _wrapped(*args, **kwargs):
plugin_dir = ph.path(__file__).realpath().parent
images_dir = plugin_dir.joinpath('images', 'insert_test_board')
image_paths = sorted(images_dir.files('insert_test_board-*.jpg'))
dialog = animation_dialog(image_paths, loop=True,
buttons=gtk.BUTTONS_OK_CANCEL)
dialog.props.text = ('<b>Please insert the DropBot test board</b>\n\n'
'For more info, see '
'<a href="https://github.com/sci-bots/dropbot-v3/wiki/DropBot-Test-Board#loading-dropbot-test-board">'
'the DropBot Test Board documentation</a>')
# Use `activate-link` callback to manually handle action when hyperlink
# is clicked/activated.
dialog.label.connect("activate-link", gtk_on_link_clicked)
dialog.props.use_markup = True
# Set default focus to OK button.
buttons = {b.props.label: b
for b in dialog.get_action_area().get_children()}
ok_button = buttons['gtk-ok']
ok_button.props.has_focus = True
ok_button.props.has_default = True
response = dialog.run()
dialog.destroy()
if response == gtk.RESPONSE_OK:
return func(*args, **kwargs)
return _wrapped
class DropBotPlugin(Plugin, gobject.GObject, StepOptionsController,
AppDataController):
"""
This class is automatically registered with the PluginManager.
.. versionchanged:: 0.19
Inherit from ``gobject.GObject`` base class to add support for
``gsignal`` signals.
Add ``gsignal`` signals for DropBot connection status and DMF chip
status.
"""
# Without the follow line, cannot inherit from both `Plugin` and
# `gobject.GObject`. See [here][1] for more details.
#
# [1]: http://code.activestate.com/recipes/204197-solving-the-metaclass-conflict/
__metaclass__ = classmaker()
#: ..versionadded:: 0.19
implements(IPlugin)
implements(IApplicationMode)
implements(IElectrodeActuator)
implements(IWaveformGenerator)
version = __version__
plugin_name = 'dropbot_plugin'
@property
def StepFields(self):
"""
Expose StepFields as a property to avoid breaking code that accesses
the StepFields member (vs through the get_step_form_class method).
.. versionchanged:: 2.33
Deprecate all step fields _except_ ``volume_threshold`` as part of
refactoring to implement `IElectrodeActuator` interface.
"""
return Form.of(#: .. versionadded:: 0.18
Float.named('volume_threshold')
.using(default=0,
optional=True,
validators=[ValueAtLeast(minimum=0),
ValueAtMost(maximum=1.0)]))
def __init__(self):
'''
.. versionchanged:: 0.19
Add ``gsignal`` signals for DropBot connection status and DMF chip
status.
Set ``threading.Event`` when DropBot connection is established.
Set ``threading.Event`` when DMF chip is inserted.
.. versionchanged:: 2.22.4
Register update of connection status when DropBot connects or
disconnects.
.. versionchanged:: 2.22.5
Revert 2.22.4 changes since connection status is already updated
when ``chip-removed`` or ``chip-inserted`` signals are emitted, and
one of these signals is emitted whenever the DropBot either
connects or disconnects.
.. versionchanged:: 2.29
Make `_on_dropbot_connected` function reentrant, i.e., support
calling the function more than once.
.. versionchanged:: 2.30
Push changes to connection status and actuation area to statusbar.
.. versionchanged:: 2.31
Display an error message when the DropBot reports that shorts have
been detected on one or more channels.
.. versionchanged:: 2.32
Display an error message when a "halted" event is received from the
DropBot.
.. versionchanged:: 2.38.1
Show dialog if 12V power supply is not detected while attempting to
connect to a DropBot, prompting user to either plugin in 12V power
supply or unplug DropBot entirely.
.. versionchanged:: 2.38.6
Fix error dialog when DropBot reports a "halted" event.
Disable real-time mode if DropBot reports a "halted" event.
'''
# Explicitly initialize GObject base class since it is not the first
# base class listed.
gobject.GObject.__init__(self)
#: ..versionadded:: 2.33
self.executor = ThreadPoolExecutor()
#: ..versionadded:: 2.33
self._capacitance_log_lock = threading.Lock()
self.control_board = None
self.name = self.plugin_name
self.connection_status = "Not connected"
self.channel_states = pd.Series()
self.plugin = None
self.plugin_timeout_id = None
self.menu_items = []
self.menu = None
self.menu_item_root = None
self.diagnostics_results_dir = ph.path('.dropbot-diagnostics')
self.actuated_area = 0
self.monitor_task = None
#: .. versionadded:: 2.24
self.device_time_sync = {}
self._chip_inserted = threading.Event()
self._dropbot_connected = threading.Event()
self.dropbot_connected.count = 0
self.dropbot_signals = blinker.Namespace()
@asyncio.coroutine
def on_inserted(sender, **message):
self.chip_inserted.set()
self.dropbot_status.chip_inserted = True
self.dropbot_signals.signal('chip-inserted').connect(on_inserted,
weak=False)
@asyncio.coroutine
def on_removed(sender, **message):
self.chip_inserted.clear()
self.dropbot_status.chip_inserted = False
self.clear_status()
self.dropbot_signals.signal('chip-removed').connect(on_removed,
weak=False)
# Update cached device load capacitance each time the
# `'capacitance-updated'` signal is emitted from the DropBot.
def _on_capacitance_updated(sender, **message):
self.on_device_capacitance_update(message['new_value'],
message['V_a'])
# Call capacitance update callback _at most_ every 200 ms.
self.dropbot_signals.signal('capacitance-updated')\
.connect(asyncio.coroutine(Debounce(_on_capacitance_updated, 100,
max_wait=200, leading=True)),
weak=False)
@asyncio.coroutine
def _on_channels_updated(sender, **message):
'''
Message keys:
- ``"n"``: number of actuated channel
- ``"actuated"``: list of actuated channel identifiers.
- ``"start"``: ms counter before setting shift registers
- ``"end"``: ms counter after setting shift registers
'''
actuated_channels = message['actuated']
if actuated_channels:
app = get_app()
actuated_electrodes = \
(app.dmf_device.actuated_electrodes(actuated_channels)
.dropna())
actuated_areas = (app.dmf_device
.electrode_areas.ix[actuated_electrodes
.values])
self.actuated_area = actuated_areas.sum()
else:
self.actuated_area = 0
# m^2 area
area = self.actuated_area * (1e-3 ** 2)
# Approximate area in SI units.
value, pow10 = si.split(np.sqrt(area))
si_unit = si.SI_PREFIX_UNITS[len(si.SI_PREFIX_UNITS) // 2 +
pow10 // 3]
status = ('actuated electrodes: %s (%.1f %sm^2)' %
(actuated_channels, value ** 2, si_unit))
self.push_status(status, None, True)
_L().debug(status)
self.dropbot_signals.signal('channels-updated')\
.connect(_on_channels_updated, weak=False)
@gtk_threadsafe
def refresh_channels():
# XXX Reassign channel states to trigger a `channels-updated`
# message since actuated channel states may have changed based
# on the channels that were disabled.
self.control_board.turn_off_all_channels()
self.control_board.state_of_channels = self.channel_states
@asyncio.coroutine
def _on_shorts_detected(sender, **message):
'''
Example message:
```python
OrderedDict([(u'event', u'shorts-detected'), (u'values', [])])
```
'''
if message.get('values'):
status = ('Shorts detected. Disabled the following '
'channels: %s' % message['values'])
gtk_threadsafe(_L().error)\
('Shorts were detected on the following '
'channels:\n\n %s\n\n'
'You may continue using the DropBot, but the '
'affected channels have been disabled until the'
' DropBot is restarted (e.g., unplug all cables'
' and plug back in).', message['values'])
else:
status = 'No shorts detected.'
# XXX Refresh channels since some channels may have been
# disabled.
refresh_channels()
self.push_status(status)
self.dropbot_signals.signal('shorts-detected')\
.connect(_on_shorts_detected, weak=False)
@asyncio.coroutine
def _on_halted(sender, **message):
# DropBot has halted. All channels have been disabled and high
# voltage has been turned off.
reason = ''
if 'error' in message:
error = message['error']
if error.get('name') == 'output-current-exceeded':
reason = ' because output current was exceeded'
elif error.get('name') == 'chip-load-saturated':
reason = ' because chip load feedback exceeded allowable range'
status = 'DropBot has halted%s.' % reason
message = '''
All channels have been disabled and high voltage has been
turned off until the DropBot is restarted (e.g., unplug all
cables and plug back in).'''.strip()
# XXX Refresh channels since channels were disabled.
refresh_channels()
app = get_app()
# Disable real-time mode.
gtk_threadsafe(app.set_app_values)({'realtime_mode': False})
self.push_status(status)
gtk_threadsafe(_L().error)\
('\n'.join([status, '', '\n'.join(map(str.strip,
message.splitlines()))]))
self.dropbot_signals.signal('halted').connect(_on_halted, weak=False)
@asyncio.coroutine
def _on_dropbot_connected(sender, **message):
'''
.. versionchanged:: 2.24
Synchronize time between DropBot microseconds count and host
UTC time.
Update local actuation voltage with voltage sent in capacitance
update events.
.. versionchanged:: 2.25
Update local list of actuated channels and associated actuated
area from ``channels-updated`` device events.
.. versionchanged:: 2.25.1
Write the actuated channels list and actuated area to the debug
log when the DropBot reports the actuated channels.
.. versionchanged:: 2.26
Set :attr:`_state_applied` event and log actuated channels/area
to ``INFO`` level when ``channels-updated`` DropBot event is
received.
.. versionchanged:: 2.27
Connect to the ``output_enabled`` and ``output_disabled``
DropBot signals to update the chip insertion status.
Configure :attr:`control_board.state.event_mask` to enable
``channels-updated`` events.
.. versionchanged:: 2.29
Make function reentrant, i.e., support calling this function
more than once. This may be useful, e.g., for supporting
DropBot reconnects.
.. versionchanged:: 2.32.1
Enable DropBot ``shorts-detected`` events by setting the
respective bit in the event mask. Fixes bug introduced in
2.31.
.. versionchanged:: 2.39.1
Disable ``chip-load-saturated`` events if the value of
C16 < 300nF.
'''
dropbot_ = message['dropbot']
map(_L().info, str(dropbot_.properties).splitlines())
self.control_board = dropbot_
if self.dropbot_connected.count < 1:
status = 'Initial connection to DropBot established.'
_L().debug(status)
self.push_status(status)
else:
# DropBot signal callbacks have already been connected.
status = ('DropBot connection re-established.')
_L().debug(status)
self.push_status(status)
self.dropbot_connected.count += 1
# Set event indicating DropBot has been connected.
self.dropbot_connected.set()
# Request for the DropBot to measure the device load capacitance
# every 100 ms.
app_values = self.get_app_values()
self.control_board.update_state(capacitance_update_interval_ms=
app_values['c_update_ms'],
event_mask=EVENT_CHANNELS_UPDATED |
EVENT_SHORTS_DETECTED |
EVENT_ENABLE)
self.device_time_sync = {'host': dt.datetime.utcnow(),
'device_us':
self.control_board.microseconds()}
self.dropbot_status.on_connected(dropbot_)
# If the feedback capacitor is < 300nF, disable the chip load
# saturation check to prevent false positive triggers.
if self.control_board.config.C16 < 0.3e-6:
self.control_board.update_state(chip_load_range_margin=-1)
self.dropbot_signals.signal('connected').connect(_on_dropbot_connected,
weak=False)
@asyncio.coroutine
def _on_dropbot_disconnected(sender, **message):
self.push_status('DropBot connection lost.')
# Clear event indicating DropBot has been disconnected.
self.dropbot_connected.clear()
self.control_board = None
self.dropbot_status.on_disconnected()
self.dropbot_signals.signal('disconnected')\
.connect(_on_dropbot_disconnected, weak=False)
@asyncio.coroutine
def on_version_mismatch(*args, **kwargs):
message = ('**DropBot driver** version `%(driver_version)s` does '
'not match **firmware** version `%(firmware_version)s`.'
% kwargs)
@sync(gtk_threadsafe)
def version_mismatch_dialog():
parent = get_app().main_window_controller.view
dialog = gtk.Dialog('DropBot driver version mismatch',
buttons=('_Update', gtk.RESPONSE_ACCEPT,
'_Ignore', gtk.RESPONSE_OK,
'_Skip', gtk.RESPONSE_NO),
flags=gtk.DIALOG_MODAL |
gtk.DIALOG_DESTROY_WITH_PARENT,
parent=parent)
# Disable dialog window close "X" button.
dialog.props.deletable = False
# Do not close window when <Escape> key is pressed.
#
# See: http://www.async.com.br/faq/pygtk/index.py?req=show&file=faq10.013.htp
def on_response(dialog, response):
if response in (gtk.RESPONSE_DELETE_EVENT, gtk.RESPONSE_CLOSE):
dialog.emit_stop_by_name('response')
dialog.connect('delete_event', lambda *args: True)
dialog.connect('response', on_response)
dialog.connect('close', lambda *args: True)
buttons = {'ignore':
dialog.get_widget_for_response(gtk.RESPONSE_OK),
'update':
dialog.get_widget_for_response(gtk.RESPONSE_ACCEPT),
'skip':
dialog.get_widget_for_response(gtk.RESPONSE_NO)}
buttons['ignore']\
.set_tooltip_text('Ignore driver version mismatch and '
'connect anyway.')
buttons['update']\
.set_tooltip_text('Upload DropBot firmware to match driver'
' version.')
buttons['skip'].set_tooltip_text('Do not connect to DropBot. '
'Unplug DropBot to avoid this'
' dialog.')
content_area = dialog.get_content_area()
label = gtk.Label(markdown2pango.markdown2pango(message))
label.props.use_markup = True
label.props.wrap = True
label.props.xpad = 20
label.props.ypad = 20
label.props.height_request = 200
label.props.height_request = 100
# Align text to top left of dialog.
label.set_alignment(0, 0)
content_area.pack_start(label, expand=True, fill=True)
content_area.show_all()
try:
response = dialog.run()
response_button = dialog.get_widget_for_response(response)
response_str = [b[0] for b in buttons.items()
if b[1] == response_button][0]
return response_str
finally:
dialog.destroy()
response = yield asyncio.From(version_mismatch_dialog())
if response == 'skip':
raise IOError(message)
raise asyncio.Return(response)
self.dropbot_signals.signal('version-mismatch')\
.connect(on_version_mismatch, weak=False)
@asyncio.coroutine
def on_no_power(*args, **kwargs):
message = ('Please connect **DropBot 12V power supply** _or_ '
'**disconnect DropBot entirely**.\n'
'\n'
'See [DropBot user guide][1] for more information.\n'
'\n'
'[1]: https://github.com/sci-bots/dropbot-v3/wiki/UserGuide#connect-dropbot')
@sync(gtk_threadsafe)
def no_power_dialog():
parent = get_app().main_window_controller.view
dialog = gtk.Dialog('No DropBot 12V power supply detected',
buttons=('_Retry', gtk.RESPONSE_OK),
flags=gtk.DIALOG_MODAL |
gtk.DIALOG_DESTROY_WITH_PARENT,
parent=parent)
# Disable dialog window close "X" button.
dialog.props.deletable = False
# Do not close window when <Escape> key is pressed.
#
# See: http://www.async.com.br/faq/pygtk/index.py?req=show&file=faq10.013.htp
def on_response(dialog, response):
if response in (gtk.RESPONSE_DELETE_EVENT, gtk.RESPONSE_CLOSE):
dialog.emit_stop_by_name('response')
dialog.connect('delete_event', lambda *args: True)
dialog.connect('response', on_response)
dialog.connect('close', lambda *args: True)
buttons = {'skip':
dialog.get_widget_for_response(gtk.RESPONSE_OK)}
content_area = dialog.get_content_area()
label = gtk.Label(markdown2pango.markdown2pango(message))
label.props.use_markup = True
label.props.wrap = True
label.props.xpad = 20
label.props.ypad = 20
label.props.height_request = 200
label.props.height_request = 100
# Align text to top left of dialog.
label.set_alignment(0, 0)
# Use `activate-link` callback to manually handle action when
# hyperlink is clicked/activated.
label.connect("activate-link", gtk_on_link_clicked)
hbox = gtk.HBox()
image = gtk.Image()
image_size = 150
image_path = ph.path(__file__).parent\
.joinpath('dropbot-power.png')
pixbuf = gtk.gdk.pixbuf_new_from_file(image_path)
if pixbuf.props.width > pixbuf.props.height:
scale = image_size / pixbuf.props.width
else:
scale = image_size / pixbuf.props.height
width = int(scale * pixbuf.props.width)
height = int(scale * pixbuf.props.height)
pixbuf = pixbuf.scale_simple(width, height,
gtk.gdk.INTERP_BILINEAR)
image.set_from_pixbuf(pixbuf)
hbox.pack_start(image, expand=False, fill=False)
hbox.pack_start(label, expand=True, fill=True)
content_area.pack_start(hbox, expand=True, fill=True)
content_area.show_all()
try:
response = dialog.run()
response_button = dialog.get_widget_for_response(response)
response_str = [b[0] for b in buttons.items()
if b[1] == response_button][0]
return response_str
finally:
dialog.destroy()
response = yield asyncio.From(no_power_dialog())
if response == 'skip':
raise IOError(message)
raise asyncio.Return(response)
self.dropbot_signals.signal('no-power').connect(on_no_power,
weak=False)
@gtk_threadsafe
def clear_status(self):
'''
Clear statusbar context for this plugin.
.. versionadded:: 2.30
'''
app = get_app()
statusbar = app.builder.get_object('statusbar')
context_id = statusbar.get_context_id(self.name)
statusbar.remove_all(context_id)
@gtk_threadsafe
def push_status(self, status, hide_timeout_s=3, clear=True):
'''
Push status message to statusbar context for this plugin.
Parameters
----------
status : str
Status message.
hide_timeout_s : float, optional
Number of seconds to display message before hiding. If `None`, do
not hide.
clear : bool, optional
Clear existing statusbar stack before pushing new status.
.. versionadded:: 2.30
'''
app = get_app()
statusbar = app.builder.get_object('statusbar')
context_id = statusbar.get_context_id(self.name)
if clear:
statusbar.remove_all(context_id)
message_id = statusbar.push(context_id, '[%s] %s' % (self.name, status))
if hide_timeout_s is not None:
# Hide status message after specified timeout.
gobject.timeout_add(int(hide_timeout_s * 1e3),
statusbar.remove_message, context_id,
message_id)
@property
def chip_inserted(self):
'''
Event set when a DMF chip is inserted into DropBot (and cleared when
DMF chip is removed).
.. versionadded:: 0.19
'''
return self._chip_inserted
@property
def dropbot_connected(self):
'''
Event set when DropBot is connected (and cleared when DropBot is
disconnected).
.. versionadded:: 0.19
'''
return self._dropbot_connected
@gtk_threadsafe # Execute in GTK main thread
@error_ignore(lambda exception, func, self, test_name, *args:
_L().error('Error executing: "%s"', test_name,
exc_info=True))
@require_connection() # Display error dialog if DropBot is not connected.
def execute_test(self, test_name, axis_count=1):
'''
Run single DropBot on-board self-diagnostic test.
Record test results as JSON and display dialog to show text summary
(and plot, where applicable).
.. versionadded:: 0.14
'''
test_func = getattr(db.hardware_test, test_name)
results = {test_name: test_func(self.control_board)}
db.hardware_test.log_results(results, self.diagnostics_results_dir)
format_func = getattr(db.self_test, 'format_%s_results' % test_name)
message = format_func(results[test_name])
map(_L().info, map(unicode.rstrip, unicode(message).splitlines()))
app = get_app()
parent = app.main_window_controller.view
dialog = results_dialog(test_name, results, parent=parent,
axis_count=axis_count)
dialog.run()
dialog.destroy()
@error_ignore(lambda *args:
_L().error('Error executing DropBot self tests.',
exc_info=True))
@require_connection() # Display error dialog if DropBot is not connected.
@require_test_board # Prompt user to insert DropBot test board
def run_all_tests(self):
'''
Run all DropBot on-board self-diagnostic tests.
Record test results as JSON and results summary as a Word document.
.. versionadded:: 0.14
.. versionchanged:: 0.16
Prompt user to insert DropBot test board.
.. versionchanged:: 2.28
Generate a self-contained HTML report with JSON report results
included in a ``<script id="results">...</script>`` tag.
.. versionchanged:: 2.28
Display a progress dialog while the test is running.
.. versionchanged:: 2.35
Use :meth:`run_tests()` to execute tests while displaying a
progress dialog indicating which test is currently running.
'''
# XXX Wait for test results in background thread to allow UI to display
# and update a progress dialog.
future = self.executor.submit(self.run_tests().result)
def _on_tests_completed(future):
results = future.result()
# Tests have completed, now:
#
# 1. Write HTML report to `.dropbot-diagnostics` directory in current
# working directory (with raw JSON results embedded in a
# `<script id="results" .../> tag).
# 2. Launch HTML report in web browser.
results_dir = ph.path(self.diagnostics_results_dir)
results_dir.makedirs_p()
# Create unique output filenames based on current timestamp.
timestamp = dt.datetime.utcnow().isoformat().replace(':', '_')
report_path = results_dir.joinpath('results-%s.html' % timestamp)
# Generate test result summary report as HTML document.
db.self_test.generate_report(results, output_path=report_path,
force=True)
# Launch HTML report.
report_path.launch()
future.add_done_callback(_on_tests_completed)
def create_ui(self):
'''
Create user interface elements (e.g., menu items).
.. versionchanged:: 0.14
Add "Run all on-board self-tests..." menu item.
Add "On-board self-tests" menu.
.. versionchanged:: 0.15
Add "Help" menu item.
.. versionchanged:: 0.16
Prompt user to insert DropBot test board before running channels
test.
.. versionchanged:: 2.34.1
Add ``_DropBot help...`` entry to main window ``Help`` menu.
.. versionchanged:: 2.37
Replace original MicroDrop help menu item.
'''
app = get_app()
# Get reference to status labels container in main window.
hbox = app.builder.get_object('status_labels')
self.dropbot_status = DropBotStatusView()