walltime_to_seconds() - 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-10-14 03:06 UTC

walltime_to_seconds() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

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

import distutils
import hashlib
import json
import re
import unicodedata

from subprocess import Popen, PIPE


TIME_REGEX = re.compile(
    "^(?:(?:(?:(\d*):)?(\d*):)?(\d*):)?(\d*)$")


def walltime_to_seconds(walltime):
    if not TIME_REGEX.match(walltime):
        raise ValueError(
            "Invalid walltime format: %s\n"
            "It must be either DD:HH:MM:SS, HH:MM:SS, MM:SS or S" %
            walltime)

    split = walltime.split(":")

    while len(split) < 4:
        split = [0] + split

    days, hours, minutes, seconds = map(int, split)

    return ((((days * 24) + hours) * 60) + minutes * 60) + seconds


def jobname_generator(jobname, job_id):
    '''Crop the jobname to a maximum of 64 characters.
    Parameters
    ----------
    jobname : str
    Initial jobname.
    job_id: str
    ID of the job in the current batch.
    Returns
    -------
    str
    The cropped version of the string.
    '''
    # 64 - 1 since the total length including -1 should be less than 64
    job_id = str(job_id)
    if len(jobname) + len(job_id) > 63:
        croped_string = '{}_{}'.format(jobname[:63 - len(job_id)], job_id)
    else:
        croped_string = '{}_{}'.format(jobname, job_id)
    return croped_string


def print_boxed(string):
    splitted_string = string.split('\n')
    max_len = max(map(len, splitted_string))
    box_line = u"\u2500" * (max_len + 2)

    out = u"\u250c" + box_line + u"\u2510\n"
    out += '\n'.join([u"\u2502 {} \u2502".format(line.ljust(max_len)) for line in splitted_string])
    out += u"\n\u2514" + box_line + u"\u2518"
    print out


def yes_no_prompt(query, default=None):
    available_prompts = {None: " [y/n] ", 'y': " [Y/n] ", 'n': " [y/N] "}

    if default not in available_prompts:
        raise ValueError("Invalid default: '{}'".format(default))

    while True:
        try:
            answer = raw_input("{0}{1}".format(query, available_prompts[default]))
            return distutils.strtobool(answer)
        except ValueError:
            if answer == '' and default is not None:
                return distutils.strtobool(default)


def chunks(sequence, n):
    """ Yield successive n-sized chunks from sequence. """
    for i in xrange(0, len(sequence), n):
        yield sequence[i:i + n]


def generate_uid_from_string(value):
    """ Create unique identifier from a string. """
    return hashlib.sha256(value).hexdigest()


def slugify(value):
    """
    Converts to lowercase, removes non-word characters (alphanumerics and
    underscores) and converts spaces to underscores. Also strips leading and
    trailing whitespace.

    Reference
    ---------
    https://github.com/django/django/blob/1.7c3/django/utils/text.py#L436
    """
    value = unicodedata.normalize('NFKD', unicode(value, "UTF-8")).encode('ascii', 'ignore').decode('ascii')
    value = re.sub('[^\w\s-]', '', value).strip().lower()
    return str(re.sub('[-\s]+', '_', value))


def encode_escaped_characters(text, escaping_character="\\"):
    """ Escape the escaped character using its hex representation """
    def hexify(match):
        return "\\x{0}".format(match.group()[-1].encode("hex"))

    return re.sub(r"\\.", hexify, text)


def decode_escaped_characters(text):
    """ Convert hex representation to the character it represents """
    if len(text) == 0:
        return ''

    def unhexify(match):
        return match.group()[2:].decode("hex")

    return re.sub(r"\\x..", unhexify, text)


def save_dict_to_json_file(path, dictionary):
    with open(path, "w") as json_file:
        json_file.write(json.dumps(dictionary, indent=4, separators=(',', ': ')))


def load_dict_from_json_file(path):
    with open(path, "r") as json_file:
        return json.loads(json_file.read())


def detect_cluster():
    # Get server status
    try:
        output = Popen(["qstat", "-B"], stdout=PIPE).communicate()[0]
    except OSError:
        # If qstat is not available we assume that the cluster uses slurm. 
        # (Otherwise return None)
        cluster_name = get_slurm_cluster_name()
        return cluster_name
    # Get server name from status
    server_name = output.split('\n')[2].split(' ')[0]
    # Cleanup the name and return it
    cluster_name = None
    if server_name.split('.')[-1] == 'm':
        cluster_name = "mammouth"
    elif server_name.split('.')[-1] == 'guil':
        cluster_name = "guillimin"
    elif server_name.split('.')[-1] == 'helios':
        cluster_name = "helios"
    elif server_name.split('.')[-1] == 'hades':
        cluster_name = "hades"
    return cluster_name

def get_slurm_cluster_name():
    try:
        stdout = Popen("sacctmgr list cluster", stdout=PIPE, shell=True).communicate()[0]
    except OSError:
        return None
    stdout = stdout.decode()
    cluster_name = stdout.splitlines()[2].strip().split(' ')[0]
    return cluster_name


MSUB_CLUSTERS = ["helios"]
SUPPORTED_LAUNCHERS = ["qsub", "msub", "sbatch"]


def command_is_available(command):
    return distutils.spawn.find_executable(command) is not None


def get_launcher(cluster_name):
    # Gives priority to msub if qsub is also present
    if cluster_name in MSUB_CLUSTERS:
        return "msub"

    for launcher in SUPPORTED_LAUNCHERS:
        if command_is_available(launcher):
            return launcher

    raise RuntimeError("No compatible launcher found on the system")


1			import distutils
2			import hashlib
3			import json
4			import re
5			import unicodedata
6
7			from subprocess import Popen, PIPE
8
9
10			TIME_REGEX = re.compile(
11			"^(?:(?:(?:(\d):)?(\d):)?(\d):)?(\d)$")
12
13
14			def walltime_to_seconds(walltime):
15			if not TIME_REGEX.match(walltime):
16			raise ValueError(
17			"Invalid walltime format: %s\n"
18			"It must be either DD:HH:MM:SS, HH:MM:SS, MM:SS or S" %
19			walltime)
20
21			split = walltime.split(":")
22
23			while len(split) < 4:
24			split = [0] + split
25
26			days, hours, minutes, seconds = map(int, split)
27
28			return ((((days * 24) + hours) * 60) + minutes * 60) + seconds
29
30
31			def jobname_generator(jobname, job_id):
32			'''Crop the jobname to a maximum of 64 characters.
33			Parameters
34			----------
35			jobname : str
36			Initial jobname.
37			job_id: str
38			ID of the job in the current batch.
39			Returns
40			-------
41			str
42			The cropped version of the string.
43			'''
44			# 64 - 1 since the total length including -1 should be less than 64
45			job_id = str(job_id)
46			if len(jobname) + len(job_id) > 63:
47			croped_string = '{}_{}'.format(jobname[:63 - len(job_id)], job_id)
48			else:
49			croped_string = '{}_{}'.format(jobname, job_id)
50			return croped_string
51
52
53			def print_boxed(string):
54			splitted_string = string.split('\n')
55			max_len = max(map(len, splitted_string))
56			box_line = u"\u2500" * (max_len + 2)
57
58			out = u"\u250c" + box_line + u"\u2510\n"
59			out += '\n'.join([u"\u2502 {} \u2502".format(line.ljust(max_len)) for line in splitted_string])
60			out += u"\n\u2514" + box_line + u"\u2518"
61			print out
62
63
64			def yes_no_prompt(query, default=None):
65			available_prompts = {None: " [y/n] ", 'y': " [Y/n] ", 'n': " [y/N] "}
66
67			if default not in available_prompts:
68			raise ValueError("Invalid default: '{}'".format(default))
69
70			while True:
71			try:
72			answer = raw_input("{0}{1}".format(query, available_prompts[default]))
73			return distutils.strtobool(answer)
74			except ValueError:
75			if answer == '' and default is not None:
76			return distutils.strtobool(default)
77
78
79			def chunks(sequence, n):
80			""" Yield successive n-sized chunks from sequence. """
81			for i in xrange(0, len(sequence), n):
82			yield sequence[i:i + n]
83
84
85			def generate_uid_from_string(value):
86			""" Create unique identifier from a string. """
87			return hashlib.sha256(value).hexdigest()
88
89
90			def slugify(value):
91			"""
92			Converts to lowercase, removes non-word characters (alphanumerics and
93			underscores) and converts spaces to underscores. Also strips leading and
94			trailing whitespace.
95
96			Reference
97			---------
98			https://github.com/django/django/blob/1.7c3/django/utils/text.py#L436
99			"""
100			value = unicodedata.normalize('NFKD', unicode(value, "UTF-8")).encode('ascii', 'ignore').decode('ascii')
101			value = re.sub('[^\w\s-]', '', value).strip().lower()
102			return str(re.sub('[-\s]+', '_', value))
103
104
105			def encode_escaped_characters(text, escaping_character="\\"):
106			""" Escape the escaped character using its hex representation """
107			def hexify(match):
108			return "\\x{0}".format(match.group()[-1].encode("hex"))
109
110			return re.sub(r"\\.", hexify, text)
111
112
113			def decode_escaped_characters(text):
114			""" Convert hex representation to the character it represents """
115			if len(text) == 0:
116			return ''
117
118			def unhexify(match):
119			return match.group()[2:].decode("hex")
120
121			return re.sub(r"\\x..", unhexify, text)
122
123
124			def save_dict_to_json_file(path, dictionary):
125			with open(path, "w") as json_file:
126			json_file.write(json.dumps(dictionary, indent=4, separators=(',', ': ')))
127
128
129			def load_dict_from_json_file(path):
130			with open(path, "r") as json_file:
131			return json.loads(json_file.read())
132
133
134			def detect_cluster():
135			# Get server status
136			try:
137			output = Popen(["qstat", "-B"], stdout=PIPE).communicate()[0]
138			except OSError:
139			# If qstat is not available we assume that the cluster uses slurm.
140			# (Otherwise return None)
141			cluster_name = get_slurm_cluster_name()
142			return cluster_name
143			# Get server name from status
144			server_name = output.split('\n')[2].split(' ')[0]
145			# Cleanup the name and return it
146			cluster_name = None
147			if server_name.split('.')[-1] == 'm':
148			cluster_name = "mammouth"
149			elif server_name.split('.')[-1] == 'guil':
150			cluster_name = "guillimin"
151			elif server_name.split('.')[-1] == 'helios':
152			cluster_name = "helios"
153			elif server_name.split('.')[-1] == 'hades':
154			cluster_name = "hades"
155			return cluster_name
156
157			def get_slurm_cluster_name():
158			try:
159			stdout = Popen("sacctmgr list cluster", stdout=PIPE, shell=True).communicate()[0]
160			except OSError:
161			return None
162			stdout = stdout.decode()
163			cluster_name = stdout.splitlines()[2].strip().split(' ')[0]
164			return cluster_name
165
166
167			MSUB_CLUSTERS = ["helios"]
168			SUPPORTED_LAUNCHERS = ["qsub", "msub", "sbatch"]
169
170
171			def command_is_available(command):
172			return distutils.spawn.find_executable(command) is not None
173
174
175			def get_launcher(cluster_name):
176			# Gives priority to msub if qsub is also present
177			if cluster_name in MSUB_CLUSTERS:
178			return "msub"
179
180			for launcher in SUPPORTED_LAUNCHERS:
181			if command_is_available(launcher):
182			return launcher
183
184			raise RuntimeError("No compatible launcher found on the system")
185

SMART-Lab / smartdispatch

Pull Request — master (#166)

walltime_to_seconds() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like