# "$Name: $";
# "$Header: $";
# =============================================================================
#
# file : NXSDataWriter.py
#
# description : Python source for the NXSDataWriter and its commands.
# The class is derived from Device. It represents the
# CORBA servant object which will be accessed from the
# network. All commands which can be executed on the
# NXSDataWriter are implemented in this file.
#
# project : TANGO Device Server
#
# $Author: $
#
# $Revision: $
#
# $Log: $
#
# copyleft : European Synchrotron Radiation Facility
# BP 220, Grenoble 38043
# FRANCE
#
# =============================================================================
# This file is generated by POGO
# (Program Obviously used to Generate tango Object)
#
# (c) - Software Engineering Group - ESRF
# =============================================================================
#
""" Nexus Data Writer - Tango Server """
try:
import tango
except Exception:
import PyTango as tango
import sys
from threading import Thread, Lock
from datetime import datetime
from .TangoDataWriter import TangoDataWriter as TDW
[docs]class CommandThread(Thread):
""" thread with server command
"""
def __init__(self, server, command, finalState, args=None):
"""constructor
:param server: Tango server implementation
:type server: :class:`tango.LatestDeviceImpl`
:param command: Thread command
:type command: :obj:`str`
:param finalState: Final State Code
:type finalState: :class:`tango.DevState`
:param args: List of command arguments
:type args: :obj:`list` <:obj:`str`>
"""
Thread.__init__(self)
#: (:class:`tango.LatestDeviceImpl`) tango server
self.server = server
#: (:obj:`__callable__`) command
self.command = getattr(server.tdw, command)
#: (:class:`tango.DevState`) final state
self.fstate = finalState
#: (:class:`tango.DevState`) error state
self.estate = tango.DevState.FAULT
#: (:obj:`list` <:obj:`str`>) command arguments
self.args = args if isinstance(args, list) else []
self.dp = tango.DeviceProxy(self.server.get_name())
self.dp.set_source(tango.DevSource.DEV)
[docs] def run(self):
""" runs the given command on the server and changes the state on exit
"""
try:
self.command(*self.args)
with self.server.lock:
self.server.state_flag = self.fstate
self.dp.state()
except (tango.DevFailed, BaseException):
self.__failed()
raise
except Exception:
self.__failed()
tango.Except.throw_exception(
str(sys.exc_info()[0]),
str(sys.exc_info()[1]),
str(sys.exc_info()[2])
)
def __failed(self):
with self.server.lock:
self.server.state_flag = self.estate
self.server.errors.append(
str(datetime.now()) + ":\n" + str(sys.exc_info()[1]))
self.dp.state()
[docs]class NXSDataWriter(tango.LatestDeviceImpl):
""" Tango Server to store data in H5 files
:brief: Device States Description:
DevState.ON : NeXuS Data Server is switch on
DevState.OFF : NeXuS Data Writer is switch off
DevState.EXTRACT : H5 file is open
DevState.OPEN : XML configuration is initialized
DevState.RUNNING : NeXus Data Server is writing
DevState.FAULT : Error state
"""
def __init__(self, cl, name):
""" Device constructor
:param cl: class name
:type cl: :obj:`str`
:param name: device name
:type name: :obj:`str`
"""
tango.LatestDeviceImpl.__init__(self, cl, name)
self.debug_stream("In __init__()")
if not hasattr(self, "lock"):
#: (:class:`threading.Lock`) thread lock
self.lock = Lock()
#: (:class:`tango.DevState`) state flag
self.state_flag = tango.DevState.OFF
#: (:class:`CommandThread`) openentry thread
self.othread = None
#: (:class:`CommandThread`) record thread
self.rthread = None
#: (:class:`CommandThread`) closentry thread
self.cthread = None
#: (:class:`nxswriter.TangoDataWriter.TangoDataWriter`) \
#: Tango Data Writer
self.tdw = TDW(self)
#: (:obj:`list`<:obj:`str`>) list with errors
self.errors = []
#: (:obj:`dict` < :class:`tango.DevState`, :obj:`str`> ) \
#: status messages
self.__status = {
tango.DevState.OFF: "Not Initialized",
tango.DevState.ON: "Ready",
tango.DevState.OPEN: "File Open",
tango.DevState.EXTRACT: "Entry Open",
tango.DevState.RUNNING: "Writing ...",
tango.DevState.FAULT: "Error",
}
NXSDataWriter.init_device(self)
[docs] def delete_device(self):
""" Device destructor
"""
self.debug_stream("In delete_device()")
if hasattr(self, 'tdw') and self.tdw:
if hasattr(self.tdw, 'closeFile'):
try:
self.tdw.closeFile()
del self.tdw
except Exception:
pass
self.tdw = None
self.set_state(tango.DevState.OFF)
[docs] def init_device(self):
""" Device initialization
"""
self.debug_stream("In init_device()")
if not hasattr(self, "lock"):
self.lock = Lock()
try:
oldstate = self.get_state()
self.set_state(tango.DevState.RUNNING)
with self.lock:
self.errors = []
if hasattr(self, 'tdw') and self.tdw:
if hasattr(self.tdw, 'closeFile'):
try:
if oldstate not in [
tango.DevState.ON,
tango.DevState.OFF
]:
self.tdw.closeFile()
del self.tdw
except Exception:
pass
self.tdw = None
self.tdw = TDW(self)
self.set_state(tango.DevState.ON)
except (tango.DevFailed, BaseException):
self.__failed()
raise
except Exception:
self.__failed()
tango.Except.throw_exception(
str(sys.exc_info()[0]),
str(sys.exc_info()[1]),
str(sys.exc_info()[2])
)
self.get_device_properties(self.get_device_class())
self.tdw.defaultCanFail = bool(self.DefaultCanFail)
self.tdw.addingLogs = bool(self.AddingLogs)
[docs] def set_state(self, state):
"""set_state method
:param state: State Code
:type state: :class:`tango.DevState`
"""
with self.lock:
if state is not None:
self.state_flag = state
tango.LatestDeviceImpl.set_state(self, self.state_flag)
[docs] def dev_state(self):
""" dev_state method
:returns: State Code
:rtype: :class:`tango.DevState`
"""
with self.lock:
tango.LatestDeviceImpl.set_state(self, self.state_flag)
if self.state_flag != tango.DevState.ALARM:
tango.LatestDeviceImpl.dev_state(self)
return tango.LatestDeviceImpl.get_state(self)
[docs] def always_executed_hook(self):
""" Always excuted hook method
"""
self.debug_stream("In always_excuted_hook()")
# ==================================================================
#
# NXSDataWriter read/write attribute methods
#
# ==================================================================
[docs] def read_attr_hardware(self, _):
""" Read Attribute Hardware
"""
self.debug_stream("In read_attr_hardware()")
def __failed(self):
""" on error
"""
self.set_state(tango.DevState.FAULT)
with self.lock:
self.errors.append(
str(datetime.now()) + ":\n" + str(sys.exc_info()[1]))
[docs] def read_CurrentFileId(self, attr):
""" Read CurrentFileId
:param attr: attribute object
:type attr: :class:`tango.Attribute`
"""
self.debug_stream("In read_CurrentFileId()")
attr.set_value(self.tdw.currentfileid)
[docs] def read_XMLSettings(self, attr):
""" Read XMLSettings attribute
:param attr: attribute object
:type attr: :class:`tango.Attribute`
"""
self.debug_stream("In read_XMLSettings()")
attr.set_value(self.tdw.xmlsettings)
[docs] def write_XMLSettings(self, attr):
""" Write XMLSettings attribute
:param attr: attribute object
:type attr: :class:`tango.Attribute`
"""
self.debug_stream("In write_XMLSettings()")
self.tdw.xmlsettings = attr.get_write_value()
[docs] def is_XMLSettings_allowed(self, _):
"""XMLSettings attribute State Machine
:returns: True if the operation allowed
:rtype: :obj:`bool`
"""
if self.get_state() in [tango.DevState.OFF,
tango.DevState.EXTRACT,
tango.DevState.RUNNING]:
return False
return True
[docs] def read_JSONRecord(self, attr):
""" Read JSONRecord attribute
:param attr: attribute object
:type attr: :class:`tango.Attribute`
"""
self.debug_stream("In read_JSONRecord()")
attr.set_value(self.tdw.jsonrecord)
[docs] def write_JSONRecord(self, attr):
""" Write JSONRecord attribute
:param attr: attribute object
:type attr: :class:`tango.Attribute`
"""
self.debug_stream("In write_JSONRecord()")
self.tdw.jsonrecord = attr.get_write_value()
[docs] def is_JSONRecord_allowed(self, _):
""" JSONRecord attribute State Machine
:returns: True if the operation allowed
:rtype: :obj:`bool`
"""
if self.get_state() in [tango.DevState.OFF,
tango.DevState.RUNNING]:
return False
return True
[docs] def read_FileName(self, attr):
""" Read FileName attribute
:param attr: attribute object
:type attr: :class:`tango.Attribute`
"""
self.debug_stream("In read_FileName()")
attr.set_value(self.tdw.fileName)
[docs] def write_FileName(self, attr):
""" Write FileName attribute
:param attr: attribute object
:type attr: :class:`tango.Attribute`
"""
self.debug_stream("In write_FileName()")
if self.is_FileName_write_allowed():
self.tdw.fileName = attr.get_write_value()
else:
self.warn_stream("To change the file name please close the file.")
raise Exception(
"To change the file name please close the file.")
[docs] def is_FileName_write_allowed(self):
""" FileName attribute Write State Machine
:returns: True if the operation allowed
:rtype: :obj:`bool`
"""
if self.get_state() in [tango.DevState.OFF,
tango.DevState.EXTRACT,
tango.DevState.OPEN,
tango.DevState.RUNNING]:
return False
return True
[docs] def is_FileName_allowed(self, _):
"""FileName attribute State Machine
:returns: True if the operation allowed
:rtype: :obj:`bool`
"""
if self.get_state() in [tango.DevState.OFF]:
return False
return True
[docs] def read_CanFail(self, attr):
""" Read CanFail attribute
:param attr: attribute object
:type attr: :class:`tango.Attribute`
"""
self.debug_stream("In read_CanFail()")
attr.set_value(self.tdw.canfail)
[docs] def write_CanFail(self, attr):
""" Write CanFail attribute
:param attr: attribute object
:type attr: :class:`tango.Attribute`
"""
self.debug_stream("In write_CanFail()")
if self.is_CanFail_write_allowed():
self.tdw.canfail = attr.get_write_value()
else:
self.warn_stream("To change the can fail flag please"
" close the file.")
raise Exception(
"To change the can fail flag please close the file.")
[docs] def is_CanFail_write_allowed(self):
""" CanFail attribute Write State Machine
:returns: True if the operation allowed
:rtype: :obj:`bool`
"""
if self.get_state() in [tango.DevState.OFF,
tango.DevState.EXTRACT,
tango.DevState.RUNNING]:
return False
return True
[docs] def is_CanFail_allowed(self, _):
"""CanFail attribute State Machine
:returns: True if the operation allowed
:rtype: :obj:`bool`
"""
if self.get_state() in [tango.DevState.OFF]:
return False
return True
[docs] def read_SkipAcquisition(self, attr):
""" Read SkipAcquisition attribute
:param attr: attribute object
:type attr: :class:`tango.Attribute`
"""
self.debug_stream("In read_SkipAcquisition()")
attr.set_value(self.tdw.skipacquisition)
[docs] def write_SkipAcquisition(self, attr):
""" Write SkipAcquisition attribute
:param attr: attribute object
:type attr: :class:`tango.Attribute`
"""
self.debug_stream("In write_SkipAcquisition()")
if self.is_SkipAcquisition_write_allowed():
self.tdw.skipacquisition = attr.get_write_value()
else:
self.warn_stream("To change the file name please close the file.")
raise Exception(
"To change the file name please close the file.")
[docs] def is_SkipAcquisition_write_allowed(self):
""" SkipAcquisition attribute Write State Machine
:returns: True if the operation allowed
:rtype: :obj:`bool`
"""
if self.get_state() in [tango.DevState.OFF,
tango.DevState.RUNNING]:
return False
return True
[docs] def is_SkipAcquisition_allowed(self, _):
"""SkipAcquisition attribute State Machine
:returns: True if the operation allowed
:rtype: :obj:`bool`
"""
if self.get_state() in [tango.DevState.OFF]:
return False
return True
[docs] def read_StepsPerFile(self, attr):
""" Read StepsPerFile attribute
:param attr: attribute object
:type attr: :class:`tango.Attribute`
"""
self.debug_stream("In read_StepsPerFile()")
attr.set_value(self.tdw.stepsperfile)
[docs] def write_StepsPerFile(self, attr):
""" Write StepsPerFile attribute
:param attr: attribute object
:type attr: :class:`tango.Attribute`
"""
self.debug_stream("In write_StepsPerFile()")
if self.is_StepsPerFile_write_allowed():
self.tdw.stepsperfile = attr.get_write_value()
else:
self.warn_stream("To change the file name please close the file.")
raise Exception(
"To change the file name please close the file.")
[docs] def is_StepsPerFile_write_allowed(self):
""" StepsPerFile attribute Write State Machine
:returns: True if the operation allowed
:rtype: :obj:`bool`
"""
if self.get_state() in [tango.DevState.OFF,
tango.DevState.EXTRACT,
tango.DevState.OPEN,
tango.DevState.RUNNING]:
return False
return True
[docs] def is_StepsPerFile_allowed(self, _):
"""StepsPerFile attribute State Machine
:returns: True if the operation allowed
:rtype: :obj:`bool`
"""
if self.get_state() in [tango.DevState.OFF]:
return False
return True
[docs] def read_Errors(self, attr):
""" Read Errors attribute
:param attr: attribute object
:type attr: :class:`tango.Attribute`
"""
self.debug_stream("In read_Errors()")
with self.lock:
attr.set_value(self.errors)
# ==================================================================
#
# NXSDataWriter command methods
#
# ==================================================================
[docs] def dev_status(self):
""" Status command
:brief: This command gets the device status
(stored in its <i>device_status</i> data member)
and returns it to the caller.
:returns: Status description
:rtype: :obj:`str`
"""
self.debug_stream("In dev_status()")
self.set_state(None)
with self.lock:
state = self.state_flag
self.set_status(self.__status[state])
return self.__status[state]
[docs] def OpenFile(self):
"""OpenFile command
:brief: Opens the H5 file
"""
self.debug_stream("In OpenFile()")
state = self.get_state()
if state in [tango.DevState.OPEN]:
try:
self.CloseFile()
except Exception:
pass
self.set_state(tango.DevState.RUNNING)
with self.lock:
self.errors = []
try:
self.tdw.writer = self.Writer
self.tdw.metadataOutput = self.MetadataOutput
self.tdw.openFile()
self.set_state(tango.DevState.OPEN)
except (tango.DevFailed, BaseException):
self.__failed()
raise
except Exception:
self.__failed()
tango.Except.throw_exception(
str(sys.exc_info()[0]),
str(sys.exc_info()[1]),
str(sys.exc_info()[2])
)
[docs] def is_OpenFile_allowed(self):
""" OpenFile command State Machine
:returns: True if the operation allowed
:rtype: :obj:`bool`
"""
if self.get_state() in [tango.DevState.OFF,
tango.DevState.EXTRACT,
tango.DevState.RUNNING]:
return False
return True
[docs] def OpenEntry(self):
""" OpenEntry command
:brief: Creates the new entry
"""
self.debug_stream("In OpenEntry()")
self.set_state(tango.DevState.RUNNING)
try:
self.get_device_properties(self.get_device_class())
self.tdw.numberOfThreads = self.NumberOfThreads
self.tdw.maxRecordRuntime = self.MaxRecordRuntime
self.tdw.maxElementRuntime = self.MaxElementRuntime
self.tdw.openEntry()
self.set_state(tango.DevState.EXTRACT)
except (tango.DevFailed, BaseException):
self.__failed()
raise
except Exception:
self.__failed()
tango.Except.throw_exception(
str(sys.exc_info()[0]),
str(sys.exc_info()[1]),
str(sys.exc_info()[2])
)
[docs] def is_OpenEntry_allowed(self):
""" OpenEntry command State Machine
:returns: True if the operation allowed
:rtype: :obj:`bool`
"""
if self.get_state() in [tango.DevState.ON,
tango.DevState.OFF,
tango.DevState.EXTRACT,
tango.DevState.FAULT,
tango.DevState.RUNNING]:
return False
return True
[docs] def Record(self, argin):
""" Record command
:brief: Records data for one scan step
:param argin: JSON string with data
:type argin: :obj:`str`
"""
self.debug_stream("In Record()")
self.set_state(tango.DevState.RUNNING)
try:
self.tdw.record(argin)
self.set_state(tango.DevState.EXTRACT)
except (tango.DevFailed, BaseException):
self.__failed()
raise
except Exception:
self.__failed()
tango.Except.throw_exception(
str(sys.exc_info()[0]),
str(sys.exc_info()[1]),
str(sys.exc_info()[2])
)
[docs] def is_Record_allowed(self):
""" Record command State Machine
:returns: True if the operation allowed
:rtype: :obj:`bool`
"""
if self.get_state() in [tango.DevState.ON,
tango.DevState.OFF,
tango.DevState.OPEN,
tango.DevState.FAULT,
tango.DevState.RUNNING]:
return False
return True
[docs] def CloseEntry(self):
""" CloseEntry command
:brief: Closes the entry
"""
self.debug_stream("In CloseEntry()")
state = self.get_state()
if state != tango.DevState.FAULT:
state = tango.DevState.OPEN
self.set_state(tango.DevState.RUNNING)
try:
self.tdw.closeEntry()
self.set_state(state)
except (tango.DevFailed, BaseException):
self.__failed()
raise
except Exception:
self.__failed()
tango.Except.throw_exception(
str(sys.exc_info()[0]),
str(sys.exc_info()[1]),
str(sys.exc_info()[2])
)
[docs] def is_CloseEntry_allowed(self):
""" CloseEntry command State Machine
:returns: True if the operation allowed
:rtype: :obj:`bool`
"""
if self.get_state() in [tango.DevState.ON,
tango.DevState.OFF,
tango.DevState.OPEN,
tango.DevState.RUNNING]:
return False
return True
[docs] def OpenEntryAsynch(self):
""" OpenEntryAsynch command
:brief: Creates the new entry in asynchronous mode
"""
self.debug_stream("In OpenEntryAsynch()")
self.set_state(tango.DevState.RUNNING)
self.get_device_properties(self.get_device_class())
self.tdw.numberOfThreads = self.NumberOfThreads
self.tdw.maxRecordRuntime = self.MaxRecordRuntime
self.tdw.maxElementRuntime = self.MaxElementRuntime
self.tdw.writer = self.Writer
self.tdw.metadataOutput = self.MetadataOutput
self.othread = CommandThread(
self, "openEntry", tango.DevState.EXTRACT)
self.othread.start()
[docs] def is_OpenEntryAsynch_allowed(self):
""" OpenEntryAsynch command State Machine
:returns: True if the operation allowed
:rtype: :obj:`bool`
"""
if self.get_state() in [tango.DevState.ON,
tango.DevState.OFF,
tango.DevState.FAULT,
tango.DevState.EXTRACT,
tango.DevState.RUNNING]:
return False
return True
[docs] def RecordAsynch(self, argin):
""" RecordAsynch command
:brief: Records data for one scan step in asynchronous mode
:param argin: DevString JSON string with data
:type argin: :obj:`str`
"""
self.debug_stream("In RecordAsynch()")
self.set_state(tango.DevState.RUNNING)
self.rthread = CommandThread(
self, "record", tango.DevState.EXTRACT, [argin])
self.rthread.start()
[docs] def is_RecordAsynch_allowed(self):
""" RecordAsynch command State Machine
:returns: True if the operation allowed
:rtype: :obj:`bool`
"""
if self.get_state() in [tango.DevState.ON,
tango.DevState.OFF,
tango.DevState.FAULT,
tango.DevState.OPEN,
tango.DevState.RUNNING]:
return False
return True
[docs] def CloseEntryAsynch(self):
"""CloseEntryAsynch command
:brief: Closes the entry is asynchronous mode
"""
self.debug_stream("In CloseEntryAsynch()")
state = self.get_state()
if state != tango.DevState.FAULT:
state = tango.DevState.OPEN
self.set_state(tango.DevState.RUNNING)
self.cthread = CommandThread(
self, "closeEntry", state)
self.cthread.start()
[docs] def is_CloseEntryAsynch_allowed(self):
""" CloseEntryAsynch command State Machine
:returns: True if the operation allowed
:rtype: :obj:`bool`
"""
if self.get_state() in [tango.DevState.ON,
tango.DevState.OFF,
tango.DevState.OPEN,
tango.DevState.RUNNING]:
return False
return True
[docs] def CloseFile(self):
"""CloseFile command
:brief: Close the H5 file
"""
self.debug_stream("In CloseFile()")
state = self.get_state()
if state in [tango.DevState.EXTRACT]:
self.CloseEntry()
if state != tango.DevState.FAULT:
state = tango.DevState.ON
self.set_state(tango.DevState.RUNNING)
try:
self.tdw.closeFile()
self.set_state(state)
except (tango.DevFailed, BaseException):
self.__failed()
raise
except Exception:
self.__failed()
tango.Except.throw_exception(
str(sys.exc_info()[0]),
str(sys.exc_info()[1]),
str(sys.exc_info()[2])
)
[docs] def is_CloseFile_allowed(self):
""" CloseFile command State Machine
:returns: True if the operation allowed
:rtype: :obj:`bool`
"""
if self.get_state() in [tango.DevState.ON,
tango.DevState.OFF,
tango.DevState.RUNNING]:
return False
return True
[docs]class NXSDataWriterClass(tango.DeviceClass):
""" NXSDataWriterClass class definition
"""
#: (:obj:`dict` <:obj:`str`, \
#: [ :obj:`str`, :class:`tango.CmdArgType`, \
#: [ :obj:`list` <:obj:`int`> ] ] > ) Class Properties
class_property_list = {
}
#: (:obj:`dict` <:obj:`str`, \
#: [ :obj:`str`, :class:`tango.CmdArgType`, \
#: [ :obj:`list` <:obj:`int`> ] ] > ) Device Properties
device_property_list = {
'NumberOfThreads':
[tango.DevLong,
"maximal number of threads",
[100]],
'MaxRecordRuntime':
[tango.DevDouble,
"maximal runtime for a record command in seconds",
[0.0]],
'MaxElementRuntime':
[tango.DevDouble,
"maximal runtime for a thread element in seconds",
[0.0]],
'Writer':
[tango.DevString,
"writer module",
[""]],
'DefaultCanFail':
[tango.DevBoolean,
"Default value of CanFail attribute",
[True]],
'AddingLogs':
[tango.DevBoolean,
"Add XML logs",
[True]],
'MetadataOutput':
[tango.DevString,
"metadata output",
[""]],
}
#: (:obj:`dict` <:obj:`str`, \
#: [[ :class:`tango.CmdArgType`, :obj:`str`]] >)
#: Command definitions
cmd_list = {
'OpenFile':
[[tango.DevVoid, ""],
[tango.DevVoid, ""]],
'OpenEntry':
[[tango.DevVoid, ""],
[tango.DevVoid, ""]],
'Record':
[[tango.DevString, "JSON string with data"],
[tango.DevVoid, ""]],
'CloseEntry':
[[tango.DevVoid, ""],
[tango.DevVoid, ""]],
'OpenEntryAsynch':
[[tango.DevVoid, ""],
[tango.DevVoid, ""]],
'RecordAsynch':
[[tango.DevString, "JSON string with data"],
[tango.DevVoid, ""]],
'CloseEntryAsynch':
[[tango.DevVoid, ""],
[tango.DevVoid, ""]],
'CloseFile':
[[tango.DevVoid, ""],
[tango.DevVoid, ""]],
}
#: (:obj:`dict` <:obj:`str`, \
#: [[ :class:`tango.CmdArgType`, :class:`tango.AttrDataFormat`, \
#: :class:`tango.AttrWriteType`], :obj:`dict` <:obj:`str`, any> ] > ) \
#: Attribute definitions
attr_list = {
'XMLSettings':
[[tango.DevString,
tango.SCALAR,
tango.READ_WRITE],
{
'label': "XML Configuration",
'description': "An XML string with Nexus configuration.",
'Display level': tango.DispLevel.EXPERT,
}],
'JSONRecord':
[[tango.DevString,
tango.SCALAR,
tango.READ_WRITE],
{
'label': "JSON string with client data",
'description': "A JSON string with global client data.",
'Display level': tango.DispLevel.EXPERT,
}],
'FileName':
[[tango.DevString,
tango.SCALAR,
tango.READ_WRITE],
{
'label': "Output file with its path",
'description': "A name of H5 output file with its full path",
}],
'CanFail':
[[tango.DevBoolean,
tango.SCALAR,
tango.READ_WRITE],
{
'label': "Can fail",
'description': "Global can fail flag. By default it is False",
}],
'SkipAcquisition':
[[tango.DevBoolean,
tango.SCALAR,
tango.READ_WRITE],
{
'label': "Skip acquisition",
'description': "Skip acquisition. "
"It is set to default False value "
"after excuting the OpenEntry, Record or CloseEntry",
}],
'Errors':
[[tango.DevString,
tango.SPECTRUM,
tango.READ, 1000],
{
'label': "List of errors",
'description': "list of errors",
}],
'CurrentFileId':
[[tango.DevLong,
tango.SCALAR,
tango.READ],
{
'label': "Current file id",
'description': "current file id",
}],
'StepsPerFile':
[[tango.DevLong,
tango.SCALAR,
tango.READ_WRITE],
{
'label': "Steps per file",
'description': "Number of steps per file",
'Memorized': "true"
}],
}
def __init__(self, name):
""" NXSDataWriterClass Constructor
"""
tango.DeviceClass.__init__(self, name)
self.set_type(name)
print("In NXSDataWriterClass constructor")
# NXSDataWriter class main method
if __name__ == '__main__':
pass