1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
|
# This file is part of htc_utils.
#
# htc_utils 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.
#
# htc_utils is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with htc_utils. If not, see <http://www.gnu.org/licenses/>.
import os
from . import recast
from collections import OrderedDict
class Job(object):
def __init__(self, filename, ext='job', * args, **kwargs):
''' Example usage
import random
from condor_batch import Job, Submit
# Create job object (filename will be "my_neato_condor.job")
job = Job('my_neato_condor')
# Populate initial job attributes
job.attr('getenv', True)
job.attr('executable', '/bin/sleep')
job.attr('arguments', 1)
job.logging('logs', create=True)
job.attr('queue')
# Populate additional sub-attributes (i.e. re-execute job with
# a different set of arguments.
for fileno in range(5):
x = random.Random().randint(2, 5)
job.subattr('arguments', x)
job.subattr('queue')
# Write job file to disk
job.commit()
# Pass our job to the Submit class
submit = Submit(job)
# Send our job to the cluster
submit.execute()
'''
self.filename = os.path.abspath('.'.join([filename, ext]))
self.shebang = '#!/usr/bin/env condor_submit'
self.config_ext = []
self.config = OrderedDict()
self.default_config = [
self.attr('universe', 'vanilla'),
self.attr('getenv', True),
self.attr('environment', ''),
self.attr('executable', ''),
self.attr('arguments', ''),
self.attr('notification', 'Never'),
self.attr('notify_user', ''),
self.attr('priority', 0),
self.attr('rank', ''),
self.attr('input', ''),
self.attr('request_cpus', ''),
self.attr('request_disk', ''),
self.attr('request_memory', ''),
self.attr('requirements', ''),
self.attr('transfer_executable', True),
self.attr('transfer_input_files', []),
self.attr('transfer_output_files', ''),
self.attr('transfer_output_remaps', ''),
self.attr('should_transfer_files', 'IF_NEEDED'),
self.attr('when_to_transfer_output', 'ON_EXIT'),
self.attr('hold', False),
self.attr('initialdir', ''),
self.attr('remote_initialdir', ''),
self.attr('run_as_owner', ''),
self.attr('nice_user', ''),
self.attr('stack_size', ''),
]
for key, value in kwargs.items():
self.attr(key, value)
def generate(self):
'''Return a
'''
if not self.config:
print("Warning: No attributes defined!")
output = ''
# Parse initial job attributes
for key, value in self.config.items():
if value and key != 'queue':
output += ' = '.join([key, value])
output += os.linesep
if key == 'queue':
output += ' '.join([key, value])
output += os.linesep
# Parse job sub-attributes
for sub in self.config_ext:
output += os.linesep
for key, value in sub.items():
if value and key != 'queue':
output += ' = '.join([key, value])
# output += os.linesep
if key == 'queue':
output += ' '.join([key, value])
output += os.linesep
return output
def logging(self, path, ext='log', create=False):
if create and not os.path.exists(path):
os.mkdir(os.path.abspath(path))
self.attr('log', os.path.normpath(os.path.join(path, '.'.join(['condor_$(Cluster)', ext]))))
self.attr('output', os.path.normpath(os.path.join(path, '.'.join(['stdout_$(Cluster)_$(Process)', ext]))))
self.attr('error', os.path.normpath(os.path.join(path, '.'.join(['stderr_$(Cluster)_$(Process)', ext]))))
def attr(self, key, *args):
value = ' '.join([recast(x) for x in args])
self.config[key] = recast(value)
def subattr(self, key, *args):
value = ' '.join([recast(x) for x in args])
self.config_ext.append(OrderedDict([(key, recast(value))]))
def commit(self, ext='job'):
with open(self.filename, 'w+') as submit_file:
submit_file.write(self.generate())
def reset(self):
self.config = OrderedDict()
self.config_ext = OrderedDict()
|
