Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
lut_data_mode = Trait('auto',
TraitPrefixList(LUT_DATA_MODE_TYPES),
desc='specify the data type used by the lookup tables',
)
# The scalar lookup table manager.
scalar_lut_manager = Instance(LUTManager, args=(), record=True)
# The vector lookup table manager.
vector_lut_manager = Instance(LUTManager, args=(), record=True)
# The name of the ModuleManager.
name = Str('Colors and legends')
# The icon
icon = Str('modulemanager.ico')
# The human-readable type for this object
type = Str(' colors and legends')
# Information about what this object can consume.
input_info = PipelineInfo(datasets=['any'])
# Information about what this object can produce.
output_info = PipelineInfo(datasets=['any'])
######################################################################
# `object` interface
######################################################################
def __get_pure_state__(self):
d = super(ModuleManager, self).__get_pure_state__()
The only instance methods that should be called in ordinary use are
'init' and 'main'. Communication should be done through the methods on
'Client' objects.
"""
spawn_commands = Dict(Str, Str)
_port = Int
_sock = Instance(socket.socket)
_port_map = Dict(Int, Instance(PortInfo))
_orphans = List(Instance(PortInfo))
_pairs = Dict(Instance(PortInfo), Instance(PortInfo))
# Commands that have been queued for spawned process
# desired_type -> list of (command, arguments)
_queue = Dict(Str, List(Tuple(Str, Str)))
def init(self, pref_path='', pref_node=''):
""" Read a configuration file and attempt to bind the server to the
specified port.
"""
# Bind to port and write port to lock file
self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self._sock.bind(('localhost', 0))
self._port = self._sock.getsockname()[1]
f = open(LOCK_PATH, 'w')
f.write(str(self._port))
f.close()
# Read configuration file
if not pref_path:
return
name = Str
enabled = Bool
deflection = Float
def __init__(self, obj):
self.name = obj.name
self.enabled = False
self.deflection = obj.deflection
class MFTableConfig(HasTraits, PersistenceMixin):
peak_center_config = Any
detectors = List
available_detector_names = List
finish_detector = Str(dump=True)
finish_isotope = Str(dump=True)
isotopes = List
isotope = Str(dump=True)
pdetectors = List(dump=True)
def dump(self, verbose=False):
self.pdetectors = [(d.name, d.enabled, d.deflection) for d in self.detectors if d.enabled]
super(MFTableConfig, self).dump(verbose=verbose)
def get_finish_position(self):
return self.finish_isotope, self.finish_detector
def set_detectors(self, dets):
self.detectors = [Detector(d) for d in dets]
self.available_detector_names = [di.name for di in self.detectors]
threshold = Float
inverted_logic = Bool
def get_value(self, v):
o = v > self.threshold
if self.inverted_logic:
o = not o
return 'ON' if o else 'OFF'
class Channel(HasTraits):
address = Str
name = Str
value = Float
process_value = Property(depends_on='value')
kind = Str('DC')
equation = Instance(Equation, ())
def traits_view(self):
v = View(HGroup(Item('name', show_label=False, style='readonly', width=200),
Item('address', show_label=False, style='readonly', width=75),
Item('value', show_label=False, style='readonly', width=100),
Item('process_value', show_label=False, style='readonly', width=100)))
return v
def _get_process_value(self):
return self.equation.get_value(self.value)
# value = self.value
# if self.coefficients:
# value = polyval(self.coefficients, value)
"""
t = getattr(object, name)
# Get the list of inner traits. Only a single inner trait is allowed.
it_list = t.trait.inner_traits()
if len(it_list) > 1:
raise TraitError("Only one inner trait may be specified when "
"using a CSVListEditor.")
# `it` is the single inner trait. This will be an instance of
# traits.traits.CTrait.
it = it_list[0]
# The following 'if' statement figures out the appropriate evaluation
# function (evaluate) and formatting function (fmt_func) for the
# given inner trait.
if it.is_trait_type(Int) or it.is_trait_type(Float) or \
it.is_trait_type(Str):
evaluate = lambda s: _eval_list_str(
s,
sep=self.sep,
item_eval=it.trait_type.evaluate,
ignore_trailing_sep=self.ignore_trailing_sep)
fmt_func = lambda vals: _format_list_str(vals,
sep=self.sep)
elif it.is_trait_type(Enum):
values, mapping, inverse_mapping = enum_values_changed(it)
evaluate = lambda s: _eval_list_str(
s,
sep=self.sep,
item_eval=mapping.__getitem__,
ignore_trailing_sep=self.ignore_trailing_sep)
fmt_func = \
lambda vals: \
# ------------------------------------------------------------------------
# 'Action' interface.
# ------------------------------------------------------------------------
def perform(self, event):
""" Activates the previous window. """
self.window.control.ActivatePrevious()
class Close(WindowAction):
""" Closes the current window. """
# 'Action' interface ---------------------------------------------------
name = Str("&Close")
tooltip = Str("Close the current window")
# ------------------------------------------------------------------------
# 'Action' interface.
# ------------------------------------------------------------------------
def perform(self, event):
""" Closes the current window. """
page = self.window.control.GetActiveChild()
if page is not None:
page.Close()
class CloseAll(WindowAction):
""" Closes all of the child windows. """
# `Explorer3D` class.
######################################################################
class Explorer3D(HasTraits):
"""This class basically allows you to create a 3D cube of data (a
numpy array), specify an equation for the scalars and view it
using the mayavi plugin.
"""
########################################
# Traits.
# Set by envisage when this is offered as a service offer.
window = Instance('pyface.workbench.api.WorkbenchWindow')
# The equation that generates the scalar field.
equation = Str('sin(x*y*z)/(x*y*z)',
desc='equation to evaluate (enter to set)',
auto_set=False,
enter_set=True)
# Dimensions of the cube of data.
dimensions = Array(value=(128, 128, 128),
dtype=int,
shape=(3,),
cols=1,
labels=['nx', 'ny', 'nz'],
desc='the array dimensions')
# The volume of interest (VOI).
volume = Array(dtype=float,
value=(-5,5,-5,5,-5,5),
shape=(6,),
CategorySubset.__repr__ = traits_repr
@camel_registry.dumper(CategorySubset, 'category-subset', 1)
def _dump_category_subset(cs):
return dict(name = cs.name,
values = cs.values,
selected = cs.selected)
@camel_registry.loader('category-subset', 1)
def _load_category_subset(data, version):
return CategorySubset(**data)
@provides(ISubset)
class RangeSubset(HasStrictTraits):
name = Str
values = List
high = CFloat(Undefined)
low = CFloat(Undefined)
str = Property(Str, depends_on = "name, values, high, low")
def default_traits_view(self):
return View(Item('high',
label = self.name,
editor = ValuesBoundsEditor(
name = 'values',
low_name = 'low',
high_name = 'high',
format = '%g',
auto_set = False)))
cache_dir = Str(
"../cache",
desc=dedent("""
The location where lyman workflows will write intermediate files during
execution. Should be defined relative to the ``lyman_dir``.
"""),
)
remove_cache = Bool(
True,
desc=dedent("""
If True, delete the cache directory containing intermediate files after
successful execution of the workflow. This behavior can be overridden
at runtime by command-line arguments.
"""),
)
fm_template = Str(
"{session}_fieldmap_{encoding}.nii.gz",
desc=dedent("""
A template string to identify session-specific fieldmap files.
"""),
)
ts_template = Str(
"{session}_{experiment}_{run}.nii.gz",
desc=dedent("""
A template string to identify time series data files.
"""),
)
sb_template = Str(
"{session}_{experiment}_{run}_ref.nii.gz",
desc=dedent("""
A template string to identify reference volumes corresponding to each
run of time series data.
if isok:
info.object.dump()
return isok
class PeakCenterConfig(HasTraits):
name = Str
detectors = List(transient=True)
detector = Str
# detector_name = Str
additional_detectors = List
available_detectors = List
isotope = Str('Ar40')
isotopes = List(transient=True)
dac = Float
use_current_dac = Bool(True)
# integration_time = Enum(QTEGRA_INTEGRATION_TIMES)
integration_time = Either(Float, Int)
integration_times = List(transient=True)
directions = Enum('Increase', 'Decrease', 'Oscillate')
dataspace = Enum('dac', 'mass', 'av')
window = Float(0.015)
step_width = Float(0.0005)
min_peak_height = Float(1.0)
percent = Int(80)