Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
from pychron.envisage.tasks.base_task_plugin import BaseTaskPlugin
from pychron.image.tasks.video_task import VideoTask
# ============= standard library imports ========================
# ============= local library imports ==========================
class VideoPlugin(BaseTaskPlugin):
'''
a list of name, url (file:///abs/path or pvs://host[:port=8080]) tuples
if file then should be a path to an image
if pvs than should be address to a Pychron Video Server
'''
id = 'pychron.video'
sources = ExtensionPoint(List,
id='pychron.video.sources'
)
def _tasks_default(self):
ts = [TaskFactory(id='pychron.video',
name='Video Display',
factory=self._video_task_factory,
task_group='hardware'
)]
return ts
def _video_task_factory(self):
t = VideoTask(
available_connections=self.sources
)
NewPatternAction, OpenPatternAction, ResetQueuesAction, OpenLastExperimentQueueAction, UndoAction, \
QueueConditionalsAction, ConfigureEditorTableAction, SystemConditionalsAction, \
OpenExperimentHistoryAction, LastAnalysisRecoveryAction, OpenCurrentExperimentQueueAction, \
SaveAsCurrentExperimentAction, SyncQueueAction, AcquireSpectrometerAction, ReleaseSpectrometerAction, \
RunHistoryAction, MeltingPointCalibrationAction
from pychron.experiment.tasks.experiment_preferences import ExperimentPreferencesPane, ConsolePreferencesPane, \
UserNotifierPreferencesPane, HumanErrorCheckerPreferencesPane
from pychron.experiment.tasks.experiment_task import ExperimentEditorTask
class ExperimentPlugin(BaseTaskPlugin):
id = 'pychron.experiment.plugin'
events = ExtensionPoint(List(ExperimentEventAddition), id='pychron.experiment.events')
dock_pane_factories = ExtensionPoint(List, id='pychron.experiment.dock_pane_factories')
activations = ExtensionPoint(List(Callable), id='pychron.experiment.activations')
deactivations = ExtensionPoint(List(Callable), id='pychron.experiment.deactivations')
def _signal_calculator_factory(self, *args, **kw):
return SignalCalculator()
def _sens_selector_factory(self, *args, **kw):
return SensitivitySelector()
def _run_history_factory(self, *args, **kw):
dvc = self.application.get_service('pychron.dvc.dvc.DVC')
rhm = RunHistoryModel(dvc=dvc)
rhm.load()
rh = RunHistoryView(model=rhm)
return rh
def _extension_registry_default(self):
""" Trait initializer. """
# fixme: Sneaky global!!!!!
from .extension_point import ExtensionPoint
return ExtensionPoint.extension_registry
OpenExperimentQueueAction, SignalCalculatorAction, \
DeselectAction, \
NewPatternAction, OpenPatternAction, ResetQueuesAction, OpenLastExperimentQueueAction, UndoAction, \
QueueConditionalsAction, ConfigureEditorTableAction, SystemConditionalsAction, \
OpenExperimentHistoryAction, LastAnalysisRecoveryAction, OpenCurrentExperimentQueueAction, \
SaveAsCurrentExperimentAction, SyncQueueAction, AcquireSpectrometerAction, ReleaseSpectrometerAction, \
RunHistoryAction, MeltingPointCalibrationAction
from pychron.experiment.tasks.experiment_preferences import ExperimentPreferencesPane, ConsolePreferencesPane, \
UserNotifierPreferencesPane, HumanErrorCheckerPreferencesPane
from pychron.experiment.tasks.experiment_task import ExperimentEditorTask
class ExperimentPlugin(BaseTaskPlugin):
id = 'pychron.experiment.plugin'
events = ExtensionPoint(List(ExperimentEventAddition), id='pychron.experiment.events')
dock_pane_factories = ExtensionPoint(List, id='pychron.experiment.dock_pane_factories')
activations = ExtensionPoint(List(Callable), id='pychron.experiment.activations')
deactivations = ExtensionPoint(List(Callable), id='pychron.experiment.deactivations')
def _signal_calculator_factory(self, *args, **kw):
return SignalCalculator()
def _sens_selector_factory(self, *args, **kw):
return SensitivitySelector()
def _run_history_factory(self, *args, **kw):
dvc = self.application.get_service('pychron.dvc.dvc.DVC')
rhm = RunHistoryModel(dvc=dvc)
rhm.load()
rh = RunHistoryView(model=rhm)
from pychron.core.helpers.strtools import to_bool
from pychron.core.yaml import yload
from pychron.envisage.view_util import open_view, close_views, report_view_stats
from pychron.globals import globalv
from pychron.hardware.core.i_core_device import ICoreDevice
from pychron.loggable import Loggable
from pychron.paths import paths
from pychron.startup_test.results_view import ResultsView
from pychron.startup_test.tester import StartupTester
class BaseTasksApplication(TasksApplication, Loggable):
about_dialog = Instance(Dialog)
startup_tester = Instance(StartupTester)
uis = List
available_task_extensions = ExtensionPoint(id='pychron.available_task_extensions')
def __init__(self, *args, **kw):
super().__init__(*args, **kw)
self.init_logger()
def _application_initialized_fired(self):
if globalv.use_startup_tests:
self.do_startup_tests()
if globalv.use_testbot:
from pychron.testbot.testbot import TestBot
testbot = TestBot(application=self)
testbot.run()
def do_startup_tests(self, force_show_results=False, **kw):
FreezeProductionRatios, InverseIsochronAction, IsoEvolutionAction, ExtractionAction, RecallAction, \
AnalysisTableAction, ClearAnalysisSetsAction, SubgroupIdeogramAction, HistoryIdeogramAction, HybridIdeogramAction, \
MassSpecReducedAction, InterpretedAgeRecallAction, IdentifyPeaksDemoAction
from pychron.pipeline.tasks.preferences import PipelinePreferencesPane
# ============= enthought library imports =======================
class PipelinePlugin(BaseTaskPlugin):
name = 'Pipeline'
id = 'pychron.pipeline.plugin'
nodes = ExtensionPoint(List, id='pychron.pipeline.nodes')
node_factories = ExtensionPoint(List, id='pychron.pipeline.node_factories')
predefined_templates = ExtensionPoint(List, id='pychron.pipeline.predefined_templates')
pipeline_group_icon_map = ExtensionPoint(List, id='pychron.pipeline.pipeline_group_icon_map')
def _help_tips_default(self):
return []
def _file_defaults_default(self):
files = [('flux_constants', 'FLUX_CONSTANTS_DEFAULT', False)]
return files
def _pipeline_factory(self):
model = self.application.get_service(SampleBrowserModel)
iamodel = self.application.get_service(InterpretedAgeBrowserModel)
dvc = self.application.get_service(DVC)
from pychron.pipeline.tasks.task import PipelineTask
t = PipelineTask(browser_model=model,
FluxAction, \
FreezeProductionRatios, InverseIsochronAction, IsoEvolutionAction, ExtractionAction, RecallAction, \
AnalysisTableAction, ClearAnalysisSetsAction, SubgroupIdeogramAction, HistoryIdeogramAction, HybridIdeogramAction, \
MassSpecReducedAction, InterpretedAgeRecallAction, IdentifyPeaksDemoAction
from pychron.pipeline.tasks.preferences import PipelinePreferencesPane
# ============= enthought library imports =======================
class PipelinePlugin(BaseTaskPlugin):
name = 'Pipeline'
id = 'pychron.pipeline.plugin'
nodes = ExtensionPoint(List, id='pychron.pipeline.nodes')
node_factories = ExtensionPoint(List, id='pychron.pipeline.node_factories')
predefined_templates = ExtensionPoint(List, id='pychron.pipeline.predefined_templates')
pipeline_group_icon_map = ExtensionPoint(List, id='pychron.pipeline.pipeline_group_icon_map')
def _help_tips_default(self):
return []
def _file_defaults_default(self):
files = [('flux_constants', 'FLUX_CONSTANTS_DEFAULT', False)]
return files
def _pipeline_factory(self):
model = self.application.get_service(SampleBrowserModel)
iamodel = self.application.get_service(InterpretedAgeBrowserModel)
dvc = self.application.get_service(DVC)
from pychron.pipeline.tasks.task import PipelineTask
return SerialPreference()
class OpenFlagManagerAction(Action):
name = 'Flag Manager'
def perform(self, event):
app = event.task.window.application
man = app.get_service('pychron.hardware.flag_manager.FlagManager')
app.open_view(man)
class HardwarePlugin(BaseTaskPlugin):
id = 'pychron.hardware.plugin'
managers = ExtensionPoint(List(Dict),
id='pychron.hardware.managers')
# my_managers = List(contributes_to='pychron.hardware.managers')
sources = List(contributes_to='pychron.video.sources')
# def _my_managers_default(self):
# return [dict(name='hardware', manager=self._hardware_manager_factory())]
# def _system_lock_manager_factory(self):
# return SystemLockManager(application=self.application)
enable_hardware_server = Bool
_remote_hardware_manager = Instance('pychron.remote_hardware.remote_hardware_manager.RemoteHardwareManager')
# _hardware_manager = Instance('pychron.managers.hardware_manager.HardwareManager')
def start(self):
from pychron.envisage.tasks.preferences import GeneralPreferencesPane
from pychron.globals import globalv
# logger = new_logger('PychronTasksPlugin')
class PychronTasksPlugin(BasePlugin):
id = 'pychron.tasks.plugin'
name = 'Tasks'
preferences_panes = List(
contributes_to='envisage.ui.tasks.preferences_panes')
task_extensions = List(contributes_to='envisage.ui.tasks.task_extensions')
actions = ExtensionPoint(List, id='pychron.actions')
file_defaults = ExtensionPoint(List(Tuple), id='pychron.plugin.file_defaults')
help_tips = ExtensionPoint(List, id='pychron.plugin.help_tips')
available_task_extensions = ExtensionPoint(List, id='pychron.available_task_extensions')
my_tips = List(contributes_to='pychron.plugin.help_tips')
def _application_changed(self):
# defaults = (('use_advanced_ui', False), ('show_random_tip', True))
defaults = (('show_random_tip', True),)
try:
self._set_preference_defaults(defaults, 'pychron.general')
except AttributeError, e:
print 'exception', e
def start(self):
self.info('Writing plugin file defaults')
for p, d, o in self.file_defaults:
try:
def disconnect_extension_point_traits(self):
""" Disconnect all of the plugin's extension points."""
ExtensionPoint.disconnect_extension_point_traits(self)
return