adding higher level beamline tools module with daq command

src/python/dm/aps_beamline_tools/__init__.py

+__version__ = "1.1 (2017.03.01)"

src/python/dm/aps_beamline_tools/cli/__init__.py

+__version__ = "1.1 (2017.03.01)"

src/python/dm/aps_beamline_tools/cli/apsBeamlineCli.py

+#!/usr/bin/env python
+
+from dm.common.cli.dmCli import DmCli
+from dm.common.exceptions.invalidRequest import InvalidRequest
+from dm.common.utility.configurationManager import ConfigurationManager
+
+class ApsBeamlineCli(DmCli):
+    """ Base APS beamline cli class. """
+
+    def __init__(self, validArgCount=0):
+        DmCli.__init__(self, validArgCount)
+
+        configManager = ConfigurationManager.getInstance()
+        self.allowedExperimentTypes = configManager.getAllowedExperimentTypes()
+        allowedExperimentTypesHelp = ''
+        if self.allowedExperimentTypes:
+            allowedExperimentTypesHelp = ' Allowed types: %s' % self.allowedExperimentTypes
+        self.stationName = configManager.getStationName()
+        self.beamlineName = configManager.getBeamlineName()
+        self.beamlineManagers = configManager.getBeamlineManagers()
+        self.dsServiceHost = configManager.getDsWebServiceHost()
+        self.dsServicePort = configManager.getDsWebServicePort()
+        self.daqServiceHost = configManager.getDaqWebServiceHost()
+        self.daqServicePort = configManager.getDaqWebServicePort()
+        self.serviceProtocol = configManager.getWebServiceProtocol()
+
+        loginGroup = 'Login Options'
+        self.addOptionGroup(loginGroup, None)
+        self.addOptionToGroup(loginGroup, '', '--login-file', dest='loginFile', help='DM login file, contains "<dm username>|<dm password>" pair. It may be specified using DM_LOGIN_FILE environment variable.')
+        self.addOptionToGroup(loginGroup, '', '--bss-login-file', dest='bssLoginFile', help='BSS login file, contains "<bss username>|<bss password>" pair. It may be specified via environment variable DM_BSS_LOGIN_FILE.')
+
+    def parseArgs(self, usage=None):
+        DmCli.parseArgs(self, usage)
+        (self.loginUsername,self.loginPassword) = self.parseLoginFile(self.getLoginFile())
+        self.bssLoginFile = self.getBssLoginFile()
+        return (self.options, self.args)
+
+    def getLoginFile(self):
+        if not self.options.loginFile:
+            return ConfigurationManager.getInstance().getLoginFile()
+        return self.options.loginFile
+
+    def getBssLoginFile(self):
+        if not self.options.bssLoginFile:
+            return ConfigurationManager.getInstance().getBssLoginFile()
+        return self.options.bssLoginFile
+
+    def getStationName(self):
+        return self.stationName
+
+    def parseLoginFile(self,loginFile):
+        username = None
+        password = None
+        try:
+            # Assume form <username>|<password>
+            if loginFile:
+                tokenList = open(loginFile).readline().split('|')
+                if len(tokenList) == 2:
+                    username = tokenList[0].strip()
+                    password = tokenList[1].strip()
+        except:
+            # Ignore invalid login file
+            pass
+        return (username,password)
+
+    def checkCredentials(self):
+        if not self.hasCredentials():
+            raise InvalidRequest('DM login credentials are not specified.')
+        if not self.hasBssCredentials():
+            raise InvalidRequest('BSS login credentials are not specified.')
+
+    def hasCredentials(self):
+        return (self.loginUsername != None and self.loginPassword != None)
+
+    def hasBssCredentials(self):
+        return self.bssLoginFile != None
+
+#######################################################################
+# Testing
+
+if __name__ == '__main__':
+    pass

src/python/dm/aps_beamline_tools/cli/daqCli.py

+#!/usr/bin/env python
+
+import os
+from dm.aps_bss.api.apsBssApi import ApsBssApi
+from dm.ds_web_service.api.experimentDsApi import ExperimentDsApi
+from dm.ds_web_service.api.userDsApi import UserDsApi
+from dm.daq_web_service.api.experimentDaqApi import ExperimentDaqApi
+
+from dm.common.utility.ftpUtility import FtpUtility
+from dm.common.exceptions.invalidRequest import InvalidRequest
+from dm.common.exceptions.objectNotFound import ObjectNotFound
+from dm.common.utility.configurationManager import ConfigurationManager
+from dm.aps_beamline_tools.cli.apsBeamlineCli import ApsBeamlineCli
+
+class DaqCli(ApsBeamlineCli):
+    def __init__(self, validArgCount=ApsBeamlineCli.ANY_NUMBER_OF_POSITIONAL_ARGS):
+        ApsBeamlineCli.__init__(self, validArgCount)
+        configManager = ConfigurationManager.getInstance()
+        self.allowedExperimentTypes = configManager.getAllowedExperimentTypes()
+        allowedExperimentTypesHelp = ''
+	self.defaultExperimentType = None
+        if self.allowedExperimentTypes:
+            allowedExperimentTypesHelp = ' Allowed types: %s' % self.allowedExperimentTypes
+	    self.defaultExperimentType = self.allowedExperimentTypes.split(',')[0]
+
+        self.addOption('', '--experiment', dest='experimentName', help='Experiment name.')
+        self.addOption('', '--data-directory', dest='dataDirectory', help='Experiment data directory.')
+
+        # Experiment options.
+        expGroup = 'Add/Update Experiment Options'
+        self.addOptionGroup(expGroup, prepend=True)
+        self.addOptionToGroup(expGroup, '', '--type', dest='typeName', default=self.defaultExperimentType, help='Experiment type name.%s' % allowedExperimentTypesHelp)
+        self.addOptionToGroup(expGroup, '', '--description', dest='description', help='Experiment description.')
+        self.addOptionToGroup(expGroup, '', '--start-date', dest='startDate', help='Experiment start date in format DD-MMM-YY.')
+        self.addOptionToGroup(expGroup, '', '--end-date', dest='endDate', help='Experiment end date in format DD-MMM-YY.')
+        self.addOptionToGroup(expGroup, '', '--users', dest='users', help='Comma-separated list of DM usernames to be added to the new experiment as users.')
+        self.addOptionToGroup(expGroup, '', '--proposal-id', dest='proposalId', help='Beamline proposal id. If specified, all users listed on the proposal will be added to the new experiment.')
+        self.addOptionToGroup(expGroup, '', '--run', dest='runName', help='Run name. If not specified, current run name is assumed for beamline proposal.')
+
+        # Daq Options
+        daqGroup = 'DAQ Options'
+        self.addOptionGroup(daqGroup, prepend=True)
+        self.addOptionToGroup(daqGroup, '', '--dest-directory', dest='destDirectory', help='Destination directory relative to experiment root path.')
+        self.addOptionToGroup(daqGroup, '', '--duration', dest='duration', help='DAQ duration; it must be specified in hours (h) or days (d). Examples: "8h", "14d".')
+        self.addOptionToGroup(daqGroup, '', '--upload-data-directory-on-exit', dest='uploadDataDirectoryOnExit', help='Data directory that will be uploaded automatically after DAQ is stopped.')
+        self.addOptionToGroup(daqGroup, '', '--upload-dest-directory-on-exit', dest='uploadDestDirectoryOnExit', help='Destination directory relative to experiment root path for automatic upload after DAQ is stopped. Requires upload data directory to be specified.')
+        self.addOptionToGroup(daqGroup, '', '--process-hidden', dest='processHidden', action='store_true', default=False, help='Process hidden source files.')
+
+    def checkArgs(self):
+        if self.options.experimentName is None:
+            raise InvalidRequest('Experiment name must be provided.')
+        if self.options.dataDirectory is None:
+            raise InvalidRequest('Experiment data directory must be provided.')
+        if self.getTypeName() and self.allowedExperimentTypes:
+            if self.getTypeName() not in self.allowedExperimentTypes.split(','):
+                raise InvalidRequest('Experiment type %s is not allowed on this station. Allowed types are: %s.' % (self.getTypeName(), self.allowedExperimentTypes))
+
+    def updateDaqInfoFromOptions(self, daqInfo):
+        if self.options.processHidden:
+            daqInfo['processHiddenFiles'] = True
+        if self.options.duration:
+            duration = self.options.duration
+            if duration.endswith('h'):
+                daqInfo['maxRunTimeInHours'] = int(duration[0:-1])
+            elif duration.endswith('d'):
+                daqInfo['maxRunTimeInHours'] = int(duration[0:-1])*self.HOURS_PER_DAY
+            else:
+                raise InvalidRequest('Maximum run time must contain valid unit specifier: "h" for hours or "d" for days.')
+        if self.options.destDirectory:
+            daqInfo['destDirectory'] = self.options.destDirectory
+        if self.options.uploadDataDirectoryOnExit:
+            daqInfo['uploadDataDirectoryOnExit'] = self.options.uploadDataDirectoryOnExit
+        if self.options.uploadDestDirectoryOnExit:
+            if not self.options.uploadDataDirectoryOnExit:
+                raise InvalidRequest('Upload destination directory on exit requires that upload data directory is specified as well.')
+            daqInfo['uploadDestDirectoryOnExit'] = self.options.uploadDestDirectoryOnExit
+
+    def getExperimentName(self):
+        return self.options.experimentName
+
+    def getTypeName(self):
+        return self.options.typeName
+
+    def getDescription(self):
+        return self.options.description
+
+    def getStartDate(self):
+        return self.options.startDate
+
+    def getEndDate(self):
+        return self.options.endDate
+
+    def getProposalId(self):
+        proposalId = self.options.proposalId
+        if proposalId:
+            proposalId = int(proposalId)
+        return proposalId
+
+    def getUsers(self):
+        # Return list of users and beamline managers that can access data
+        users = self.options.users
+        if users:
+            users = users.split(',')
+        else:
+            users = []
+        beamlineManagers = self.beamlineManagers
+        if beamlineManagers:
+            beamlineManagers = beamlineManagers.split(',')
+        else:
+            beamlineManagers = []
+        # Remove duplicates by converting into set
+        return list(set(users+beamlineManagers))
+
+    def getDataDirectory(self):
+        dataDirectory = self.options.dataDirectory
+        replacementMap = os.environ.get('DM_DATA_DIRECTORY_MAP', '')
+        (scheme, host, port, dirPath) = FtpUtility.parseUrl(dataDirectory)
+        if dirPath and replacementMap:
+            # Map entries are expected to be in the form
+            # <original>|<replacement>;<original>|<replacement>;...
+            for entry in replacementMap.split(';'):
+                original = entry.split('|')[0]
+                replacement = entry.split('|')[1]
+                dirPath = dataDirectory.replace(original,replacement)
+        return FtpUtility.assembleUrl(scheme, host, port, dirPath)
+
+    def addOrUpdateExperiment(self):
+        dsExperimentApi = ExperimentDsApi(self.loginUsername, self.loginPassword, self.dsServiceHost, self.dsServicePort, self.serviceProtocol)
+        dsUserApi = UserDsApi(self.loginUsername, self.loginPassword, self.dsServiceHost, self.dsServicePort, self.serviceProtocol)
+
+        description = self.getDescription()
+        proposalId = self.getProposalId()
+        experimenters = []
+        if proposalId:
+            bssApi = ApsBssApi(loginFile=self.options.bssLoginFile)
+            proposal = bssApi.getBeamlineProposal(proposalId=proposalId, runName=self.options.runName)
+            experimenters = proposal.get('experimenters', [])
+            if not description:
+                description = '%s (Proposal id: %s)' % (proposal['title'], proposalId)
+
+
+        users = self.getUsers()
+        pis = []
+        for experimenter in experimenters:
+            badge = int(experimenter['badge'])
+            if not badge:
+                #print 'Skipping user %s due to invalid badge.' % lastName 
+                continue
+            username = 'd%s' % badge
+
+            # Clasify user
+            if experimenter.get('piFlag') == 'Y':
+                if not pis.count(username):
+                    pis.append(username)
+                if users.count(username):
+                    users.remove(username)
+            else:
+                if not users.count(username):
+                    users.append(username)
+
+        for username in users+pis:
+            # Check that user exists
+            dsUserApi.getUserByUsername(username)
+           
+
+        # Everything looks good, add experiment
+        try:
+            experiment = dsExperimentApi.getExperimentByName(self.getExperimentName())
+            experimentStation = experiment.get('experimentStation')
+            stationName = ''
+            if experimentStation:
+                stationName = experimentStation.get('name')
+            if stationName != self.getStationName():
+                raise InvalidRequest('Experiment %s already exists for station %s.' % (self.getExperimentName(), stationName))
+        except ObjectNotFound, ex:
+            experiment = dsExperimentApi.addExperiment(self.getExperimentName(), self.getStationName(), self.getTypeName(), description, self.getStartDate(), self.getEndDate())
+
+        # Add pis.
+        experimentUsernameList = experiment.get('experimentUsernameList', [])
+        experimentName = experiment['name']
+        roleName = 'PI'
+        for username in pis:
+            if username not in experimentUsernameList: 
+                dsUserApi.addUserExperimentRole(username, roleName, experimentName)
+        roleName = 'User'
+        for username in users:
+            if username not in experimentUsernameList: 
+                dsUserApi.addUserExperimentRole(username, roleName, experimentName)
+        if len(users+pis):
+            experiment = dsExperimentApi.getExperimentByName(experimentName)
+        return experiment
+
+    def startDaq(self):
+        daqExperimentApi = ExperimentDaqApi(self.loginUsername, self.loginPassword, self.daqServiceHost, self.daqServicePort, self.serviceProtocol)
+        daqInfo = self.splitArgsIntoDict()
+        self.updateDaqInfoFromOptions(daqInfo)
+        daqInfo = daqExperimentApi.startDaq(self.getExperimentName(), self.getDataDirectory(), daqInfo=daqInfo)
+        return daqInfo
+
+    def runCommand(self):
+        self.parseArgs(usage="""
+    dm-%s-daq --experiment=EXPERIMENTNAME --data-directory=DATADIRECTORY
+        [--duration=DURATION]
+        [--dest-directory=DESTDIRECTORY]
+        [--upload-data-directory-on-exit=UPLOADDATADIRECTORYONEXIT]
+        [--upload-dest-directory-on-exit=UPLOADDESTDIRECTORYONEXIT]
+        [--process-hidden]
+        [--type=TYPENAME]
+        [--description=DESCRIPTION] 
+        [--start-date=STARTDATE] 
+        [--end-date=ENDDATE]
+        [--users=USERS]
+        [--proposal-id=PROPOSALID]
+        [--run=RUNNAME]
+        [key1:value1, key2:value2, ...]
+
+Description:
+    Run DAQ for experiment on station %s. If experiment does not exist, it will be
+    added to the DM database. If list of users or proposal id is specified, this command will also add roles for all users listed on the proposal.
+        """ % (self.getStationName().lower(), self.getStationName()))
+        self.checkArgs()
+        self.checkCredentials()
+        experiment = self.addOrUpdateExperiment()
+        print 'EXPERIMENT INFO'
+        print experiment.getDisplayString(self.getDisplayKeys(), self.getDisplayFormat())
+
+        print 
+        daqInfo = self.startDaq()
+        print 'DAQ INFO'
+        print daqInfo.getDisplayString(self.getDisplayKeys(), self.getDisplayFormat())
+
+
+#######################################################################
+# Run command.
+if __name__ == '__main__':
+    cli = DaqCli()
+    cli.run()
+