# -*- coding: utf-8 -*-
"""The baseline for loading a scanning instrument
Each instrument will have its own module that declares a class
inheriting from ScanningInstrument. The abstract base class is used
to ensure that the derived classes define the necessary methods to run
any generic scripts.
"""
from abc import ABCMeta, abstractmethod, abstractproperty
from logging import info, warning
from six import add_metaclass
from .genie import gen
[docs]@add_metaclass(ABCMeta) # pylint: disable=too-many-public-methods
class ScanningInstrument(object):
"""The base class for scanning measurement instruments."""
_dae_mode = None
title_footer = ""
_TIMINGS = ["uamps", "frames", "seconds", "minutes", "hours"]
def __init__(self):
self.setup_sans = self.setup_dae_event
self.setup_trans = self.setup_dae_transmission
[docs] def set_default_dae(self, mode=None, trans=False):
"""Set the default DAE mode for SANS or TRANS measuremnts.
Parameters
----------
mode : str or function
If the mode is a function, call that function to set the DAE
mode. If the mode is a string, call the function whose name
is "setup_dae_" followed by that string.
trans : bool
If true, set the default transmission instead of the default
SANS mode.
"""
if mode is None:
pass
elif isinstance(mode, str):
self.set_default_dae(
getattr(self, "setup_dae_"+mode))
else:
if trans:
self.setup_trans = mode
else:
self.setup_sans = mode
@property
def TIMINGS(self): # pylint: disable=invalid-name
"""The list of valid waitfor keywords."""
return self._TIMINGS
[docs] def sanitised_timings(self, kwargs):
"""Include only the keyword arguments for run timings.
Parameters
----------
kwargs : dict
A dictionary of keyword arguments
Returns
-------
dict
Keyword arguments accepted by gen.waitfor
"""
result = {}
for k in self.TIMINGS:
if k in kwargs:
result[k] = kwargs[k]
return result
@staticmethod
def _generic_scan(detector, spectra, wiring, tcbs):
"""A utility class for setting up dae states
On its own, it's not particularly useful, but
letting subclasses provide default parameters
simplifies creating new dae states.
"""
gen.change(nperiods=1)
gen.change_start()
gen.change_tables(detector=detector)
gen.change_tables(spectra=spectra)
gen.change_tables(wiring=wiring)
for tcb in tcbs:
gen.change_tcb(**tcb)
gen.change_finish()
@abstractproperty
def _poslist(self):
"""The list of named positions that the instrument can run through in
the sample changer"""
return []
@staticmethod
def _needs_setup():
if gen.get_runstate() != "SETUP":
raise RuntimeError("Cannot start a measurement in a measurement")
[docs] @abstractmethod
def setup_dae_scanning(self):
"""Set the wiring tables for a scan"""
pass
[docs] @abstractmethod
def setup_dae_nr(self):
"""Set the wiring tables for a neutron
reflectivity measurement"""
pass
[docs] @abstractmethod
def setup_dae_nrscanning(self):
"""Set the wiring tables for performing
scans during neutron reflectivity"""
pass
[docs] @abstractmethod
def setup_dae_event(self):
"""Set the wiring tables for event mode"""
pass
[docs] @abstractmethod
def setup_dae_histogram(self):
"""Set the wiring tables for histogram mode"""
pass
[docs] @abstractmethod
def setup_dae_event_fastsave(self):
"""Event mode with reduced detector histogram binning to decrease
filesize."""
pass
[docs] @abstractmethod
def setup_dae_transmission(self):
"""Set the wiring tables for a transmission measurement"""
pass
[docs] @abstractmethod
def setup_dae_polarised(self):
"""Set the wiring tables for a polarisation measurement"""
pass
[docs] @abstractmethod
def setup_dae_bsalignment(self):
"""Configure wiring tables for beamstop alignment."""
pass
[docs] @abstractmethod
def setup_dae_monitorsonly(self):
"""Set the wiring tables to record only the monitors"""
pass
def _configure_sans_custom(self, size):
"""The specific actions required by the instrument
to run a SANS measurement (e.g. remove the monitor
from the beam).
This is a no-op for the default instrument but can be
overwritten by other instruments to perform any actions they
need to put the instrument into SANS mode.
Parameters
----------
size : str
The aperature size (e.g. "Small" or "Medium"). A blank
string results in the aperature not being changed.
"""
pass
def _configure_trans_custom(self, size):
"""The specific actions required by the instrument
to run a SANS measurement (e.g. remove the monitor
from the beam).
This is a no-op for the default instrument but can be
overwritten by other instruments to perform any actions they
need to put the instrument into SANS mode.
Parameters
----------
size : str
The aperature size (e.g. "Small" or "Medium"). A blank
string results in the aperature not being changed.
"""
pass
[docs] def begin(self, *args, **kwargs):
"""Start a measurement."""
if hasattr(self, "_begin_"+self._dae_mode):
getattr(self, "_begin_"+self._dae_mode)(*args, **kwargs)
else:
gen.begin(*args, **kwargs)
[docs] def end(self):
"""End a measurement."""
if hasattr(self, "_end_"+self._dae_mode):
getattr(self, "_end_"+self._dae_mode)()
else:
gen.end()
[docs] def waitfor(self, **kwargs):
"""Await the user's desired statistics."""
if hasattr(self, "_waitfor_"+self._dae_mode):
getattr(self, "_waitfor_"+self._dae_mode)(**kwargs)
else:
gen.waitfor(**kwargs)
[docs] @staticmethod
@abstractmethod
def set_aperature(size):
"""Set the beam aperature to the desired size
Parameters
----------
size : str
The aperature size. e.g. "Small" or "Medium"
A blank string (the default value) results in
the aperature not being changed."""
pass
[docs] def detector_on(self, powered=None, delay=True):
"""Query and set the detector's electrical state.
Parameters
----------
on : bool or None
If None, then return the detector's current state. If True,
turn the detector on. If False, turn the detector off.
delay : bool
If changing the detector state, whether to wait for the
detector to finish warming up or powering down before
continuing the script.
Returns :
bool
If the detector is currently on
"""
if powered is not None:
if powered is True:
self._detector_turn_on(delay=delay)
else:
self._detector_turn_off(delay=delay)
return self._detector_is_on()
@staticmethod
@abstractmethod
def _detector_is_on():
"""Determine the current state of the detector.
Returns
-------
bool
True if the detector is powered up.
"""
return False
@staticmethod
@abstractmethod
def _detector_turn_on(delay=True):
"""Power on the detector
Parameters
==========
delay : bool
Wait for the detector to warm up before continuing
"""
return False
@staticmethod
@abstractmethod
def _detector_turn_off(delay=True):
"""Remove detector power
Parameters
==========
delay : bool
Wait for the detector to cool down before continuing
"""
return False
[docs] def check_move_pos(self, pos):
"""Check whether the position is valid and return True or False
Parameters
----------
pos : str
The sample changer position
"""
if pos.upper() not in self._poslist:
warning("Error in script, position {} does not exist".format(pos))
return False
return True
[docs] def measure(self, title, pos=None, thickness=1.0, trans=False,
dae=None, aperature="", **kwargs):
"""Take a sample measurement.
Parameters
==========
title : str
The title for the measurement. This is the only required parameter.
pos
The sample position. This can be a string with the name of
a sample position or it can be a function which moves the
detector into the desired position. If left undefined, the
instrument will take the measurement in its current
position.
thickness : float
The thickness of the sample in millimeters. The default is 1mm.
trans : bool
Whether to perform a transmission run instead of a sans run.
dae : str or func
This option allows setting the default dae mode. It takes a
string that contains the name of the DAE mode to be used as
the new default. For example,
>>> measure("Test", frames=10, dae="event")
Is equivalent to
>>> set_default_dae(setup_dae_event)
>>> measure("Test", frames=10)
If dae is a function, then the function is set to the default
>>> measure("Test", frames=10, dae=foo)
Is equivalent to
>>> set_default_dae(foo)
>>> measure("Test", frames=10)
aperature : str
The aperature size. e.g. "Small" or "Medium" A blank string
(the default value) results in the aperature not being
changed.
**kwargs
This function takes two kinds of keyword arguments. If
given a block name, it will move that block to the given
position. If given a time duration, then that will be the
duration of the run.
Examples
========
>>> measure("H2O", frames=900)
Perform a SANS measurment in the current position on a 1 mm
thick water sample until the proton beam has released 900
proton pulses (approx 15 minutes).
>>> measure("D2O", "LT", thickness=2.0, trans=True, Phi=3, uamps=10)
Move to sample changer position LT, then adjust the CoarseZ
motor to 38 mm. Finally, take a transmission measurement on a
2 mm thick deuterium sample for 10 µA hours of proton
current. (approx 15 minutes).
"""
self._needs_setup()
if not self.detector_on() and not trans:
warning("The detector was off. Turning on the detector")
self.detector_on(True)
self.set_default_dae(dae, trans)
if trans:
self.configure_trans(size=aperature)
else:
self.configure_sans(size=aperature)
if pos:
if isinstance(pos, str):
if self.check_move_pos(pos=pos):
info("Moving to sample changer position {}".format(pos))
gen.cset(SamplePos=pos)
else:
raise RuntimeError(
"Position {} does not exist".format(pos))
elif callable(pos):
info("Moving to position {}".format(pos.__name__))
pos()
else:
raise TypeError("Cannot understand position {}".format(pos))
for arg in kwargs:
if arg in self.TIMINGS:
continue
info("Moving {} to {}".format(arg, kwargs[arg]))
gen.cset(arg, kwargs[arg])
times = self.sanitised_timings(kwargs)
gen.waitfor_move()
gen.change_sample_par("Thick", thickness)
info("Using the following Sample Parameters")
self.printsamplepars()
gen.change(title=title+self.title_footer)
self.begin()
info("Measuring {title:} for {time:} {units:}".format(
title=title+self.title_footer,
units=list(times.keys())[0],
time=times[list(times.keys())[0]]))
self.waitfor(**times)
self.end()
[docs] def measure_file(self, file_path, forever=False):
"""Perform a series of measurements based on a spreadsheet
The file should contain comma separated values. Excel can
easily produce files of this sort. The first line of the file
is the header with each field giving the name of a parameter
to the `measure` function. As always, the ``title`` parameter
is mandatory. Each subsequent line of the file represents a
single measurement with the fields indicating that values to
pass to their corresponding keywords. If a cell is blank, the
keyword's default parameter it used. Boolean values are
represented by `True` and `False` and are not case sensitive.
The script is run through the simulator to check for errors
before attempting a real run.
Parameters
----------
file_path : str
The location of the script file
forever : bool
If set to True, the instrument will repeatedly run the
script manually stopped. This can be useful for an
overnight run where you want to keep measureing until the
users return.
"""
from .Util import user_script
@user_script
def inner():
"""Actually load and run the script"""
import csv
import ast
with open(file_path, "rb") as csvfile:
reader = csv.DictReader(csvfile)
for row in reader:
for k in row.keys():
if row[k].strip() == "":
del row[k]
elif row[k].upper() == "TRUE":
row[k] = True
elif row[k].upper() == "FALSE":
row[k] = True
else:
try:
row[k] = ast.literal_eval(row[k])
except ValueError:
continue
self.measure(**row)
if forever:
while True:
inner()
else:
inner()
[docs] @staticmethod
def printsamplepars():
"""Display the basic sample parameters on the console."""
pars = gen.get_sample_pars()
for par in ["Geometry", "Width", "Height", "Thickness"]:
info("{}={}".format(par, pars[par.upper()]))