123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343 |
- """
- This is a custom logfile creation module
- """
- import datetime
- import os
- from PyQt4 import QtGui, QtCore
- # from backend import board
- from backend.board import available_boards
- import backendinterface as bif
- import numpy as np
- import codecs
- from .. import config
- import storage
- import logging
- epics_reachable = True
- try: # try to import epics
- import epics
- no_epics = False
- os.environ["EPICS_BASE"] = config.epics_base_path
- try: # if import was successful, try to find libca and test for connectivity with config.epics_test_pv
- # sys.stderr = os.devnull # epics.ca.find_libca() prints a useless None to stderr if libca is not found
- epics.ca.find_libca()
- # sys.stderr = sys.__stderr__
- if epics.caget(config.epics_test_pv) == None:
- no_epics = True
- epics_reachable = False
- logging.error("Epics is not accessible (possible Timeout)")
- except epics.ca.ChannelAccessException as e:
- if str(e) == "cannot find Epics CA DLL":
- no_epics = True
- logging.error("Epics CA DLL not found")
- except ImportError:
- no_epics = True
- # after = datetime.datetime.now()
- # if (after-before).total_seconds() > 3.0:
- # no_epics = True
- # logging.error("Epics is not accessible (Timeout)")
- # tr = kcgw.tr
- tr = lambda _, x: x # log entries will not be translated
- class LogLevels: # Keep every new value an integer. Every level with higher value will include those with lower level
- NONE = -1
- INFO = 0
- DEBUG = 1
- TWO_COLUMNS = False
- BOARDID = "BOARD_ID"
- class MeasurementLogger(object):
- """
- Logfile creator class
- It will automatically get the info needed to create a logfile entry from registered functions.
- """
- def __init__(self, filename=None, level=LogLevels.INFO, oneline=False):
- """
- Initialise the logfile creator
- :param filename: (str) filename of the logfile THIS IS IGNORED AS OF NOW
- :param level: (int) valid log level (see LogLevels)
- :param oneline: (bool) whether to format the logfile entries in multilines or one line
- :return: -
- """
- self.level = level
- self.filename = filename
- self.parameter_functions = []
- self.oneline = oneline
- self.dumper = None
- self.predefined_parameters = [ # format: [logstring, [function (args,)]]
- ["Number of Orbits", [bif.bk_get_config, (BOARDID, 'orbits_observe')]],
- ["Number of Skipped Orbits", [bif.bk_get_config, (BOARDID, 'orbits_skip')]],
- ["Number of Acquisitions", [bif.bk_get_config, (BOARDID, 'acquisition_count')]],
- ["Time between Acquisitions", [bif.bk_get_config, (BOARDID, 'orbits_wait_time')]],
- ["Pilot Bunch Simulator", [bif.bk_get_config, (BOARDID, 'pilot_bunch')]],
- ["Header saved", [bif.bk_get_config, (BOARDID, 'header')]],
- ["T/H Delay", [bif.bk_get_config, (BOARDID, 'th_delay')]],
- ["ADC 1 Delay", [bif.bk_get_config, (BOARDID, 'chip_1_delay')]],
- ["ADC 2 Delay", [bif.bk_get_config, (BOARDID, 'chip_2_delay')]],
- ["ADC 3 Delay", [bif.bk_get_config, (BOARDID, 'chip_3_delay')]],
- ["ADC 4 Delay", [bif.bk_get_config, (BOARDID, 'chip_4_delay')]]
- ]
- self.gui_values_number = len(self.predefined_parameters)
- if not no_epics:
- for entr in config.epics_log_entry_pvs:
- self.predefined_parameters.append([entr[0], [epics.caget, entr[1]]])
- def __now(self):
- """
- Get the current time in %Y-%m-%d %H:%M:%S format
- :return: the time
- """
- return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
- def _logging_formatter(self, info):
- """
- Formatter for the logfile entries
- Override this to implement custom formatters
- """
- if self.oneline:
- return self.__now() + " " + ", ".join(info)
- return self.__now() + "\n\t" + "\n\t".join(info)
- def _log(self, board_id, stuff):
- """
- Write entries to logfile
- This will also make shure the directory the logfile is located is available and if not creates it
- :param stuff: (str) the entrie to write
- :return: -
- """
- # if not self.filename:
- # filename = storage.storage.save_location + '/' + storage.storage.subdirname + "/Measurement.log"
- # else:
- # filename = self.filename
- filename = storage.storage.save_location + '/' + \
- storage.storage.subdirname + '/' + "/Measurement_board_" + \
- available_boards.get_board_name_from_id(board_id).replace(' ', '_') + ".log"
- if not os.path.isdir(str(storage.storage.save_location + '/' + storage.storage.subdirname)):
- os.makedirs(str(storage.storage.save_location + '/' + storage.storage.subdirname))
- f = codecs.open(filename, 'a', encoding='utf-8')
- f.write(stuff+'\n\n')
- f.close()
- def register_parameter(self, logstring, func, args):
- """
- This function is used to register parameters for the logfile entries
- :param logstring: (str) the string to describe the value in the logfile
- :param func: (callable) the function used to get the value
- :param args: arguments to pass to func upon gathering of information
- :return: -
- """
- if not isinstance(args, tuple):
- args = (args,)
- self.parameter_functions.append([func, args, logstring])
- # self.parameter_functions[logstring] = [func, args, logstring]
- def reset_parameters(self):
- """
- Resets all parameters (unregisteres all parameters)
- """
- self.parameter_functions = []
- def register_dumper(self, func):
- """
- Register a dumper that is used to dump a log of data into the logfile
- :param func: (callable) Function that is used as dumper
- :return: -
- """
- self.dumper = func
- def dump(self, board_id):
- """
- Perform a dump (see register_dumper)
- :return: -
- """
- self.do_log(board_id=board_id, c=self.__now() + " " + self.dumper())
- def do_log(self, c=None, additional=None, board_id=None, level=LogLevels.INFO):
- """
- Perform a log. This creates a log entry in the logfile
- :param c: (str) if this is not None this is the only text written to the logfile (in addition to the current time)
- :param additional: (str) This text is written below the current time to customize log entries
- :param board_id: the id of the board to log for, if this is not given, all boards are logged
- :return: -
- """
- if level > self.level:
- return
- if c:
- if board_id is not None:
- self._log(board_id, c)
- else:
- for bid in available_boards:
- self._log(bid, c)
- else:
- def do_the_log(bid):
- st = []
- if additional:
- s = additional.split('\n')[0]
- for line in additional.split('\n')[1:]:
- s += "\n\t"+line
- st.append(s)
- for par in self.parameter_functions:
- par[1] = list(par[1])
- for i, p in enumerate(par[1]):
- if p == BOARDID:
- par[1][i] = bid
- st.append(par[2] + ": " + str(par[0](*par[1])))
- return st
- if board_id is not None:
- st = do_the_log(board_id)
- self._log(board_id, self._logging_formatter(st))
- else:
- for bid in available_boards:
- st = do_the_log(bid)
- self._log(bid, self._logging_formatter(st))
- logger = None
- dump_general = False
- def log(c=None, additional=None, dump=False, board_id=None, level=LogLevels.INFO):
- """
- Execute logging if logger was registered
- :param (str) c: if this is given c is the only entry that is created (additional is ignored)
- :param (str) additional: the additional text that is written below the current time
- :param (bool) dump: if this is true, a dump of every log value is performed
- :param board_id: the board id to log for, if this is not given, all boards are logged
- :return: -
- """
- global logger, dump_general
- if logger:
- if dump or dump_general:
- logger.dump(board_id=board_id)
- else:
- logger.do_log(c, additional, board_id=board_id, level=level)
- else:
- print "No Logger registered"
- class ConfigureLog(QtGui.QDialog):
- """
- Class that is basically a QDialog to configure the logger
- """
- def __init__(self, parent=None): # parameter parent does nothing when adding parent to the folowing __init__
- # no entry in taskbar is made
- super(ConfigureLog, self).__init__()
- self.setWindowTitle("Configure Measurement Log Entries")
- self.setWindowIcon(QtGui.QIcon(config.install_path + config.guiIcon))
- self.sl = QtGui.QVBoxLayout()
- self.setLayout(self.sl)
- self.scrollArea = QtGui.QScrollArea()
- if TWO_COLUMNS:
- self.scrollArea.setMinimumSize(410, 300)
- else:
- self.scrollArea.setMinimumSize(250, 520)
- self.sl.addWidget(self.scrollArea)
- self.widget = QtGui.QWidget()
- if TWO_COLUMNS:
- self.layout = QtGui.QGridLayout()
- else:
- self.layout = QtGui.QVBoxLayout()
- self.widget.setLayout(self.layout)
- self.parameter = []
- for par in logger.predefined_parameters:
- self.parameter.append((QtGui.QCheckBox(par[0]), par[1]))
- self.i = 0
- def add(w):
- if w == 'sep':
- sep = QtGui.QFrame()
- sep.setFrameShape(QtGui.QFrame.HLine)
- w = sep
- self.layout.addWidget(w, self.i//2, self.i%2)
- self.i += 1
- for i, (cb, _) in enumerate(self.parameter):
- if TWO_COLUMNS:
- if i == logger.gui_values_number:
- # sep1 = QtGui.QFrame()
- # sep1.setFrameShape(QtGui.QFrame.HLine)
- # sep2 = QtGui.QFrame()
- # sep2.setFrameShape(QtGui.QFrame.HLine)
- if logger.gui_values_number%2 == 1:
- self.i += 1
- # add(sep1)
- # add(sep2)
- add('sep')
- add('sep')
- add(cb)
- else:
- if i == logger.gui_values_number:
- sep = QtGui.QFrame()
- sep.setFrameShape(QtGui.QFrame.HLine)
- self.layout.addWidget(sep)
- self.layout.addWidget(cb)
- self.all = QtGui.QCheckBox(tr("Label", "All of the above"))
- self.all.clicked.connect(self.all_changed)
- if TWO_COLUMNS:
- if (len(logger.predefined_parameters)-1)%2 == 1:
- self.i += 1
- add('sep')
- add('sep')
- add(self.all)
- else:
- sepa = QtGui.QFrame()
- sepa.setFrameShape(QtGui.QFrame.HLine)
- self.layout.addWidget(sepa)
- self.layout.addWidget(self.all)
- self.buttonLayout = QtGui.QHBoxLayout()
- self.sl.addLayout(self.buttonLayout)
- self.ok = QtGui.QPushButton(tr("Button", "OK"))
- self.cancel = QtGui.QPushButton(tr("Button", "Cancel"))
- self.buttonLayout.addWidget(self.ok)
- self.buttonLayout.addWidget(self.cancel)
- self.ok.pressed.connect(self.do)
- self.cancel.pressed.connect(self.close)
- self.scrollArea.setWidget(self.widget)
- self.initial_ticks()
- def initial_ticks(self):
- pf = np.array(logger.parameter_functions)
- for cb, _ in self.parameter:
- cb.setChecked(str(cb.text()) in pf[:, 2])
- def all_changed(self):
- if self.all.isChecked():
- for cb, _ in self.parameter:
- cb.setEnabled(False)
- else:
- for cb, _ in self.parameter:
- cb.setEnabled(True)
- def do(self):
- global logger
- logger.reset_parameters()
- def nif(cb, f, args):
- if cb.isChecked() or self.all.isChecked():
- logger.register_parameter(str(cb.text()), f, args)
- for cb, par in self.parameter:
- nif(cb, par[0], par[1])
- self.close()
|