PBS.__init__() - Code Metrics - Inspection of "[WIP] Test for pbs files on slurm" - SMART-Lab/smartdispatch - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#166)

unknown

created 2017-09-19 21:23 UTC

PBS.init() A

↳ Parent: PBS

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
c	0
b	0
f	0
dl	0
loc	20
rs	9.4285

import re
from collections import OrderedDict

regex_walltime = re.compile("(\d+:){1,4}")
regex_resource_nodes = re.compile("[a-zA-Z0-9]+(:ppn=\d+)?(:gpus=\d+)?(:[a-zA-Z0-9]+)*")
regex_resource_pmem = re.compile("[0-9]+(b|kb|mb|gb|tb)?")


class PBS(object):
    """ Offers functionalities to manage a PBS file.

    For more information about the PBS file format see:
    `http://docs.adaptivecomputing.com/suite/8-0/basic/help.htm#topics/torque/2-jobs/requestingRes.htm?TocPath=TORQUE Resource Manager|Submitting and managing jobs|Job submission|_____3`

    Parameters
    ----------
    queue_name : str
        name of the queue on which commands will be executed
    walltime : str
        maximum time allocated to execute every commands (DD:HH:MM:SS)
    """
    def __init__(self, queue_name, walltime):
        if queue_name is None or len(queue_name) == 0:
            raise ValueError("Queue's name must be provided.")

        self.queue_name = queue_name
        self.modules = []
        self.prolog = []
        self.commands = []
        self.epilog = []

        self.resources = OrderedDict()
        self.add_resources(walltime=walltime)

        self.options = OrderedDict()
        self.add_options(q=queue_name)

        self.sbatch_options = OrderedDict()

        # Declares that all environment variables in the qsub command's environment are to be exported to the batch job.
        self.add_options(V="")

    def add_options(self, **options):
        """ Adds options to this PBS file.

        Parameters
        ----------
        **options : dict
            each key is the name of a PBS option (see `Options`)

        Options
        -------
        *A* : account_string
            Defines the account string associated with the job.
        *N* : name (up to 64 characters)
            Declares a name for the job. It must consist of printable,
            non white space characters with the first character alphabetic.
        """
        for option_name, option_value in options.items():
            # If known option, validate it.
            if option_name.strip('-') == 'N':
                if len(option_name) > 64:
                    raise ValueError("Maximum number of characters for the name is: 64")

            self.options["-" + option_name] = option_value

    def add_sbatch_options(self, **options):
        """ Adds sbatch options to this PBS file.

        Parameters
        ----------
        **options : dict
            each key is the name of a SBATCH option (see `Options`)
        """

        for option_name, option_value in options.items():
            self.sbatch_options[option_name] = option_value

    def add_resources(self, **resources):
        """ Adds resources to this PBS file.

        Parameters
        ----------
        **resources : dict
            each key is the name of a PBS resource (see `Resources`)

        Resources
        ---------
        *nodes* : nodes={<node_count>|<hostname>}[:ppn=<ppn>][:gpus=<gpu>][:<property>[:<property>]...]
            Specifies how many and what type of nodes to use
            **nodes={<node_count>|<hostname>}**: type of nodes
            **ppn=#**: Number of process per node requested for this job
            **gpus=#**: Number of process per node requested for this job
            **property**: A string specifying a node's feature
        *pmem*: pmem=[0-9]+(b|kb|mb|gb|tb)
            Specifies the maximum amount of physical memory used by any single process of the job.
        """
        for resource_name, resource_value in resources.items():
            # If known ressource, validate it.
            if resource_name == 'nodes':
                if re.match(regex_resource_nodes, str(resource_value)) is None:
                    raise ValueError("Unknown format for PBS resource: nodes")
            elif resource_name == 'pmem':
                if re.match(regex_resource_pmem, str(resource_value)) is None:
                    raise ValueError("Unknown format for PBS resource: pmem")
            elif resource_name == 'walltime':
                if re.match(regex_walltime, str(resource_value)) is None:
                    raise ValueError("Unknown format for PBS resource: walltime (dd:hh:mm:ss)")

            self.resources[resource_name] = resource_value

    def add_modules_to_load(self, *modules):
        """ Adds modules to load prior to execute the job on a node.

        Parameters
        ----------
        *modules : list of str
            each string represents the name of the module to load
        """
        self.modules += modules

    def add_to_prolog(self, *code):
        """ Adds the code to be executed before the commands.

        Parameters
        ----------
        *code : list of str
            Each string holds the code to be executed before the commands
        """
        self.prolog += code

    def add_commands(self, *commands):
        """ Sets commands to execute on a node.

        Parameters
        ----------
        *commands : list of str
            each string represents a command that is part of this job
        """
        self.commands += commands

    def add_to_epilog(self, *code):
        """ Adds the code to be executed after the commands.

        Parameters
        ----------
        *code : list of str
            Each string holds the code to be executed after the commands
        """
        self.epilog += code

    def save(self, filename):
        """ Saves this PBS job to a file.

        Parameters
        ----------
        filename : str
            specified where to save this PBS file
        """
        with open(filename, 'w') as pbs_file:
            pbs_file.write(str(self))

    def __str__(self):
        pbs = []
        pbs += ["#!/bin/bash"]

        for option_name, option_value in self.options.items():
            if option_value == "":
                pbs += ["#PBS {0}".format(option_name)]
            else:
                pbs += ["#PBS {0} {1}".format(option_name, option_value)]

        for resource_name, resource_value in self.resources.items():
            pbs += ["#PBS -l {0}={1}".format(resource_name, resource_value)]

        for option_name, option_value in self.sbatch_options.items():
            pbs += ["#SBATCH {0}={1}".format(option_name, option_value)]

        pbs += ["\n# Modules #"]
        for module in self.modules:
            pbs += ["module load " + module]

        pbs += ["\n# Prolog #"]
        pbs += self.prolog

        pbs += ["\n# Commands #"]
        pbs += ["{command}".format(command=command) for command in self.commands]

        pbs += ["\n# Epilog #"]
        pbs += self.epilog

        return "\n".join(pbs)


1			import re
2			from collections import OrderedDict
3
4			regex_walltime = re.compile("(\d+:){1,4}")
5			regex_resource_nodes = re.compile("[a-zA-Z0-9]+(:ppn=\d+)?(:gpus=\d+)?(:[a-zA-Z0-9]+)*")
6			regex_resource_pmem = re.compile("[0-9]+(b\|kb\|mb\|gb\|tb)?")
7
8
9			class PBS(object):
10			""" Offers functionalities to manage a PBS file.
11
12			For more information about the PBS file format see:
13			`http://docs.adaptivecomputing.com/suite/8-0/basic/help.htm#topics/torque/2-jobs/requestingRes.htm?TocPath=TORQUE Resource Manager\|Submitting and managing jobs\|Job submission\|_____3`
14
15			Parameters
16			----------
17			queue_name : str
18			name of the queue on which commands will be executed
19			walltime : str
20			maximum time allocated to execute every commands (DD:HH:MM:SS)
21			"""
22			def __init__(self, queue_name, walltime):
23			if queue_name is None or len(queue_name) == 0:
24			raise ValueError("Queue's name must be provided.")
25
26			self.queue_name = queue_name
27			self.modules = []
28			self.prolog = []
29			self.commands = []
30			self.epilog = []
31
32			self.resources = OrderedDict()
33			self.add_resources(walltime=walltime)
34
35			self.options = OrderedDict()
36			self.add_options(q=queue_name)
37
38			self.sbatch_options = OrderedDict()
39
40			# Declares that all environment variables in the qsub command's environment are to be exported to the batch job.
41			self.add_options(V="")
42
43			def add_options(self, **options):
44			""" Adds options to this PBS file.
45
46			Parameters
47			----------
48			**options : dict
49			each key is the name of a PBS option (see `Options`)
50
51			Options
52			-------
53			A : account_string
54			Defines the account string associated with the job.
55			N : name (up to 64 characters)
56			Declares a name for the job. It must consist of printable,
57			non white space characters with the first character alphabetic.
58			"""
59			for option_name, option_value in options.items():
60			# If known option, validate it.
61			if option_name.strip('-') == 'N':
62			if len(option_name) > 64:
63			raise ValueError("Maximum number of characters for the name is: 64")
64
65			self.options["-" + option_name] = option_value
66
67			def add_sbatch_options(self, **options):
68			""" Adds sbatch options to this PBS file.
69
70			Parameters
71			----------
72			**options : dict
73			each key is the name of a SBATCH option (see `Options`)
74			"""
75
76			for option_name, option_value in options.items():
77			self.sbatch_options[option_name] = option_value
78
79			def add_resources(self, **resources):
80			""" Adds resources to this PBS file.
81
82			Parameters
83			----------
84			**resources : dict
85			each key is the name of a PBS resource (see `Resources`)
86
87			Resources
88			---------
89			nodes : nodes={<node_count>\|<hostname>}[:ppn=<ppn>][:gpus=<gpu>][:<property>[:<property>]...]
90			Specifies how many and what type of nodes to use
91			nodes={<node_count>\|<hostname>}: type of nodes
92			ppn=#: Number of process per node requested for this job
93			gpus=#: Number of process per node requested for this job
94			property: A string specifying a node's feature
95			pmem: pmem=[0-9]+(b\|kb\|mb\|gb\|tb)
96			Specifies the maximum amount of physical memory used by any single process of the job.
97			"""
98			for resource_name, resource_value in resources.items():
99			# If known ressource, validate it.
100			if resource_name == 'nodes':
101			if re.match(regex_resource_nodes, str(resource_value)) is None:
102			raise ValueError("Unknown format for PBS resource: nodes")
103			elif resource_name == 'pmem':
104			if re.match(regex_resource_pmem, str(resource_value)) is None:
105			raise ValueError("Unknown format for PBS resource: pmem")
106			elif resource_name == 'walltime':
107			if re.match(regex_walltime, str(resource_value)) is None:
108			raise ValueError("Unknown format for PBS resource: walltime (dd:hh:mm:ss)")
109
110			self.resources[resource_name] = resource_value
111
112			def add_modules_to_load(self, *modules):
113			""" Adds modules to load prior to execute the job on a node.
114
115			Parameters
116			----------
117			*modules : list of str
118			each string represents the name of the module to load
119			"""
120			self.modules += modules
121
122			def add_to_prolog(self, *code):
123			""" Adds the code to be executed before the commands.
124
125			Parameters
126			----------
127			*code : list of str
128			Each string holds the code to be executed before the commands
129			"""
130			self.prolog += code
131
132			def add_commands(self, *commands):
133			""" Sets commands to execute on a node.
134
135			Parameters
136			----------
137			*commands : list of str
138			each string represents a command that is part of this job
139			"""
140			self.commands += commands
141
142			def add_to_epilog(self, *code):
143			""" Adds the code to be executed after the commands.
144
145			Parameters
146			----------
147			*code : list of str
148			Each string holds the code to be executed after the commands
149			"""
150			self.epilog += code
151
152			def save(self, filename):
153			""" Saves this PBS job to a file.
154
155			Parameters
156			----------
157			filename : str
158			specified where to save this PBS file
159			"""
160			with open(filename, 'w') as pbs_file:
161			pbs_file.write(str(self))
162
163			def __str__(self):
164			pbs = []
165			pbs += ["#!/bin/bash"]
166
167			for option_name, option_value in self.options.items():
168			if option_value == "":
169			pbs += ["#PBS {0}".format(option_name)]
170			else:
171			pbs += ["#PBS {0} {1}".format(option_name, option_value)]
172
173			for resource_name, resource_value in self.resources.items():
174			pbs += ["#PBS -l {0}={1}".format(resource_name, resource_value)]
175
176			for option_name, option_value in self.sbatch_options.items():
177			pbs += ["#SBATCH {0}={1}".format(option_name, option_value)]
178
179			pbs += ["\n# Modules #"]
180			for module in self.modules:
181			pbs += ["module load " + module]
182
183			pbs += ["\n# Prolog #"]
184			pbs += self.prolog
185
186			pbs += ["\n# Commands #"]
187			pbs += ["{command}".format(command=command) for command in self.commands]
188
189			pbs += ["\n# Epilog #"]
190			pbs += self.epilog
191
192			return "\n".join(pbs)
193

SMART-Lab / smartdispatch

Pull Request — master (#166)

PBS.__init__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

PBS.init() A