Source code for scipion.scripts.kickoff

# -*- coding: utf-8 -*-
#!/usr/bin/env python
# **************************************************************************
# *
# * Authors:     Pablo Conesa (
# *
# * Unidad de  Bioinformatica of Centro Nacional de Biotecnologia , CSIC
# *
# * This program is free software; you can redistribute it and/or modify
# * it under the terms of the GNU General Public License as published by
# * the Free Software Foundation; either version 3 of the License, or
# * (at your option) any later version.
# *
# * This program is distributed in the hope that it will be useful,
# * but WITHOUT ANY WARRANTY; without even the implied warranty of
# * GNU General Public License for more details.
# *
# * You should have received a copy of the GNU General Public License
# * along with this program; if not, write to the Free Software
# * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA
# * 02111-1307  USA
# *
# *  All comments concerning this program package may be sent to the
# *  e-mail address ''
# *
# **************************************************************************
Creates a scipion workflow file (json formatted) base on a template.
The template may have some ~placeholders~ that will be overwritten with values
Template may look like this, separator is "~" and within it you can define:
Template string sits at the end of the file ready for a running streaming demo.
import subprocess
import sys
import os
import re
import tkinter as tk
import tkinter.font as tkFont
import traceback
import time

import pyworkflow as pw
import pyworkflow.utils as pwutils
from pyworkflow.gui import Message, dialog
from pyworkflow.plugin import Template
from pyworkflow.project import ProjectSettings, Project
import pyworkflow.gui as pwgui
from pyworkflow.gui.project.base import ProjectBaseWindow
from pyworkflow.gui.widgets import HotButton, Button
from pyworkflow.template import TemplateList
from scipion.constants import SCIPION_EP, MODE_PROJECT

# Custom labels
from scipion.utils import getExternalJsonTemplates

ENTRY = 'entry'
LABEL = 'label'
CHECKBUTTON = 'Checkbutton'
YES = "Yes"
NO = "No"



VIEW_WIZARD = 'wizardview'

# Project name files
PROJECT_NAME = 'Project name'
DO_NOT_SCHEDULE = "Cancel schedule"
DO_NOT_SHOW_GUI = "Don't show the project"

# Project regex to validate the session id name
PROJECT_PATTERN = "^\w{2}\d{4,6}-\d+$"

[docs]class KickoffWindow(ProjectBaseWindow): """ Windows to manage all projects. """ def __init__(self, template, argsList, showScheduleOption, schedule, showProjectOption, showProject, showProjectName): try: title = '%s (%s on %s)' % ('Workflow template customizer', pwutils.getLocalUserName(), pwutils.getLocalHostName()) except Exception: title = Message.LABEL_PROJECTS ProjectBaseWindow.__init__(self, title, minsize=(800, 350)) self.template = template self.argsList = argsList self.showScheduleOption = showScheduleOption self.schedule = schedule self.showProjectOption = showProjectOption self.showProject = showProject self.showProjectName = showProjectName self.viewFuncs = {VIEW_WIZARD: KickoffView} self.action = Message.LABEL_BUTTON_CANCEL self.switchView(VIEW_WIZARD)
[docs] def getTemplate(self): return self.template
[docs] def getAction(self): return self.action
[docs] def switchView(self, newView, **kwargs): # Destroy the previous view if exists: if self.viewWidget: self.viewWidget.grid_forget() self.viewWidget.destroy() # Create the new view: Instantiates KickoffView HERE!. self.viewWidget = self.viewFuncs[newView](self.footer, self, template=self.template, argsList=self.argsList, showScheduleOption=self.showScheduleOption, schedule=self.schedule, showProjectOption=self.showProjectOption, showProject=self.showProject, showProjectName=self.showProjectName) # Grid in the second row (1) self.viewWidget.grid(row=0, column=0, columnspan=10, sticky='news') self.footer.rowconfigure(0, weight=1) self.footer.columnconfigure(0, weight=1) self.view = newView
def _onClosing(self): self.root.destroy() if self.showScheduleOption and self.showProjectOption: sys.exit()
[docs]class KickoffView(tk.Frame): def __init__(self, parent, windows, template:Template=None, argsList=[], showScheduleOption=True, schedule=True, showProjectOption=True, showProject=True, showProjectName=True, **kwargs): tk.Frame.__init__(self, parent, bg='white', **kwargs) = windows self.root = windows.root self.vars = {} self.checkvars = [] self.template = template self.argsList = argsList self.showScheduleOption = showScheduleOption self.schedule = schedule self.showProjectOption = showProjectOption self.showProject = showProject self.showProjectName = showProjectName bigSize = pw.Config.SCIPION_FONT_SIZE + 2 smallSize = pw.Config.SCIPION_FONT_SIZE - 2 fontName = pw.Config.SCIPION_FONT_NAME self.bigFont = tkFont.Font(size=bigSize, family=fontName) self.bigFontBold = tkFont.Font(size=bigSize, family=fontName, weight='bold') self.projDateFont = tkFont.Font(size=smallSize, family=fontName) self.projDelFont = tkFont.Font(size=smallSize, family=fontName, weight='bold') # Body section bodyFrame = tk.Frame(self, bg='white') bodyFrame.columnconfigure(0, minsize=120) bodyFrame.columnconfigure(1, minsize=120, weight=1) bodyFrame.grid(row=0, column=0, sticky='news') self._fillContent(bodyFrame) # Add the create project button btnFrame = tk.Frame(self, bg='white') btn = HotButton(btnFrame, text=ACCEPT_BUTTON, font=self.bigFontBold, command=self._onReadDataFromTemplateForm) btn.grid(row=0, column=1, sticky='ne', padx=10, pady=10) # Add the Import project button btn = Button(btnFrame, Message.LABEL_BUTTON_CANCEL, font=self.bigFontBold, command=self._closeCallback) btn.grid(row=0, column=0, sticky='ne', pady=10) btnFrame.columnconfigure(0, weight=1) btnFrame.grid(row=1, column=0, sticky='news') self.columnconfigure(0, weight=1) def _closeCallback(self): self.root.destroy() if self.showScheduleOption and self.showProjectOption: sys.exit() def _fillContent(self, frame): # Add project name self.template.genProjectName() self._addPair(PROJECT_NAME, PROJECT_NAME, 1, frame, value=self.template.projectName, visible=self.showProjectName) self._addPair(DO_NOT_SCHEDULE, DO_NOT_SCHEDULE, 2, frame, widget=CHECKBUTTON, value=not self.schedule, visible=self.showScheduleOption) self._addPair(DO_NOT_SHOW_GUI, DO_NOT_SHOW_GUI, 3, frame, widget=CHECKBUTTON, value=not self.showProject, pady=(5, 30), visible=self.showProjectOption) # Add template params self._addTemplateFieldsToForm(frame) def _addPair(self, text, title, r, lf, widget=ENTRY, traceCallback=None, mouseBind=False, value=None, pady=2, visible=True): if visible: label = tk.Label(lf, text=text, bg='white', font=self.bigFont) label.grid(row=r, column=0, padx=(10, 5), pady=pady, sticky='nes') if not widget: return var = tk.StringVar() if value is not None: var.set(value) self.vars[title] = var if visible: if widget == ENTRY: widget = tk.Entry(lf, width=30, font=self.bigFont, textvariable=var) if traceCallback: if mouseBind: # call callback on click widget.bind("<Button-1>", traceCallback, "eee") else: # call callback on type var.trace('w', traceCallback) elif widget == LABEL: widget = tk.Label(lf, font=self.bigFont, textvariable=var) elif widget == CHECKBUTTON: var.set(YES if value else NO) widget = tk.Checkbutton(lf, text="", font=self.bigFont, variable=var, onvalue=YES, offvalue=NO, bg=pw.Config.SCIPION_BG_COLOR) widget.grid(row=r, column=1, sticky='news', padx=(5, 10), pady=pady) def _addTemplateFieldsToForm(self, labelFrame): row = 5 for field in self.template.params.values(): alias = field.getAlias() text = field.getTitle() if alias is None else "%s (%s)" % (field.getTitle(), alias) self._addPair(text, field.getTitle(), row, labelFrame, value=field.getValue()) row += 1 def _getVar(self, varKey): return self.vars[varKey] def _getValue(self, varKey): return self.vars[varKey].get() def _setValue(self, varKey, value): return self.vars[varKey].set(value) # noinspection PyUnusedLocal def _onReadDataFromTemplateForm(self, e=None): errors = [] # Check the entered data for field in self.template.params.values(): newValue = self._getValue(field.getTitle()) field.setValue(newValue) validationMsg = field.validate() if validationMsg: errors.append("%s value does not validate: %s. Value: %s, Type: %s." % (field.getTitle(), validationMsg, field.getValue(), field.getType())) # Do more checks only if there are no previous errors if errors: errors.insert(0, "*Errors*:")"\n - ".join(errors)) else: self.template.projectName = self._getValue(PROJECT_NAME) # Set parent with the data = self.template if self._getValue(DO_NOT_SCHEDULE) == YES: self.argsList.append(NOSCHEDULE_FLAG) if self._getValue(DO_NOT_SHOW_GUI) == YES: self.argsList.append(NOGUI_FLAG) = ACCEPT_BUTTON return
[docs]def getTemplates(templateName=None): """ Get a template or templates either from arguments or from the templates directory. If more than one template is found or passed, a dialog is raised to choose one. """ tempList = TemplateList() tempId = None if templateName: if os.path.isfile(templateName) and os.path.exists(templateName): t = Template("custom_template", templateName) tempList.addTemplate(t) else: tempId = templateName # Try to find all templates from the template folder and the plugins if len(tempList.templates) == 0: tempList.addScipionTemplates(tempId) if not (tempId is not None and len(tempList.templates) == 1): tempList.addPluginTemplates(tempId) if not len(tempList.templates): raise Exception("No valid file found (*.json.template).\n" "Please, add (at least one) at %s " "or pass it as argument(s).\n" "\n -> Usage: scipion template [PATH.json.template]\n" "\n see 'scipion help'\n" % getExternalJsonTemplates()) return tempList.sortListByPluginName().templates
[docs]def chooseTemplate(templates, parentWindow=None): chosenTemplate = None if len(templates) == 1: chosenTemplate = templates[0] else: provider = pwgui.tree.ListTreeProviderTemplate(templates) dlg = dialog.ListDialog(parentWindow, "Workflow templates", provider, "Select one of the templates.", selectOnDoubleClick=True) if dlg.result == dialog.RESULT_YES: chosenTemplate = dlg.values[0] if chosenTemplate is not None: print("Template to use: %s" % chosenTemplate) # Replace environment variables chosenTemplate.replaceEnvVariables() return chosenTemplate
[docs]def resolveTemplate(template, argsList, showScheduleOption=True, schedule=True, showProjectOption=True, showProject=True, showProjectName=True): """ Resolve a template assigning CML params to the template. if not enough, a window will pop pup to ask for missing ones only""" if not assignAllParams(argsList, template): wizWindow = KickoffWindow(template=template, argsList=argsList, showScheduleOption=showScheduleOption, schedule=schedule, showProjectOption=showProjectOption, showProject=showProject, showProjectName=showProjectName) return wizWindow.action == ACCEPT_BUTTON else: # All parameters have been assigned and template is fully populated # Add schedule and showProject flags back: removed at flag2value if not schedule: argsList.append(NOSCHEDULE_FLAG) if not showProject: argsList.append(NOGUI_FLAG) return True
[docs]def assignAllParams(argsList, template): """ Assign CML params to the template, if missing params after assignment return False """ paramsSetted = 0 template.parseContent() if argsList: for attr in argsList: # skipp --params if attr.startswith(FLAG_PARAM): continue aliasAttr, valAttr = attr.split('=') try: paramsSetted += template.setParamValue(aliasAttr, valAttr) except Exception as e: print(pwutils.redStr(e)) sys.exit(os.EX_DATAERR) return len(template.params) == paramsSetted
[docs]def createTemplateFile(template): try: workflow = template.createTemplateFile() except Exception as e: workflow = None errorStr = "Couldn't create the template.\n" + str(e) print(errorStr) traceback.print_exc() return workflow
[docs]def launchTemplate(argsList, template): """ Launches a resolved template""" workflow = createTemplateFile(template) if workflow is not None: # Create the project if not template.projectName: template.genProjectName() createProjectFromWorkflow(workflow, template.projectName, argsList)
[docs]def importTemplate(template, window): """ Import a resolved template """ workflow = createTemplateFile(template) if workflow is not None: try: window.getViewWidget().info('Importing workflow %s' % workflow) window.project.loadProtocols(workflow) window.getViewWidget().updateRunsGraph(True) window.getViewWidget().cleanInfo() except Exception as ex: window.showError(str(ex), exception=ex)
[docs]def createProjectFromWorkflow(workflow, projectName, argsList): scipion = SCIPION_EP scriptsPath = pw.join('project', 'scripts') # Clean the project name as pyworkflow will do projectName = Project.cleanProjectName(projectName) # Create the project print("Creating project %s" % projectName) createProjectScript = os.path.join(scriptsPath, '') os.system("python -m %s python %s %s %s" % (scipion, createProjectScript, projectName, workflow)) # Wait 2 seconds to avoid activity time.sleep(2) if scheduleProject(argsList): # Schedule the project scheduleProjectScript = os.path.join(scriptsPath, '') print("Scheduling project %s" % projectName) subprocess.Popen(["python", "-m", scipion, "python", scheduleProjectScript, projectName]) # Wait 5 seconds to avoid activity time.sleep(5) if launchGUI(argsList): print("Showing project %s" % projectName) # Launch scipion subprocess.Popen(["python", "-m", scipion, MODE_PROJECT, projectName])
[docs]def flag2Value(argsList, flag): # Remove the flag from sys.argsv value = getFlagArg(argsList, flag) if value: argsList.remove(flag) return value
[docs]def launchGUI(argsList): """Checks if project GUI has to be launched. Only if --noGUI param is found in the argument List it will return False""" return not getFlagArg(argsList, NOGUI_FLAG)
[docs]def scheduleProject(argsList): return not getFlagArg(argsList, NOSCHEDULE_FLAG)
[docs]def getFlagArg(argsList, flag): """Checks if a flag exists (True) or not (False)""" for arg in argsList: if flag == arg.lower(): return True # Flag not found return False
[docs]def main(): """ Resolves command line arguments for scipion template""" # Remove "template" argsList = sys.argv[1:] # Now, there 2 cases: # 1.- it comes with a template name (full path) or id (name) # 2.- is empty templateName = argsList[0] if argsList else None templates = getTemplates(templateName) # Remove the name from the args, in case it is passed argsList = argsList[1:] chosenTemplate = chooseTemplate(templates) if chosenTemplate is not None and resolveTemplate(chosenTemplate, argsList, schedule=not flag2Value(argsList, NOSCHEDULE_FLAG), showProject=not flag2Value(argsList, NOGUI_FLAG)): launchTemplate(argsList, chosenTemplate)
if __name__ == "__main__": main()