get_advice() - Code Metrics - Inspection of "Handle resources error gracefully. (#155)" - SMART-Lab/smartdispatch - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#167)

unknown

created 2017-10-10 20:47 UTC

get_advice() B

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	2
Bugs	0	Features	0

Metric	Value
cc	5
c	2
b	0
f	0
dl	0
loc	12
rs	8.5454

import re
import hashlib
import unicodedata
import json
import sys
import six

from distutils.util import strtobool
from subprocess import Popen, PIPE

HELIOS_ADVICE = ("On Helios, don't forget that the queue gpu_1, gpu_2, gpu_4"
                 " and gpu_8 give access to a specific amount of gpus."
                 "\nFor more advices, please refer to the official"
                 " documentation 'https://wiki.calculquebec.ca/w/Helios/en'")

MAMMOUTH_ADVICE = ("On Mammouth, please refer to the official documentation"
                   " for more information:"
                   " 'https://wiki.ccs.usherbrooke.ca/Accueil/en'")

HADES_ADVICE = ("On Hades, don't forget that the queue name '@hades' needs"
                " to be used.\nFor more advices, please refer to the"
                " official documentation: 'https://wiki.calculquebec.ca/w"
                "/Ex%C3%A9cuter_une_t%C3%A2che/en#tab=tab5'")

GUILLIMIN_ADVICE = ("On Guillimin, please refer to the official documentation"
                    " for more information: 'http://www.hpc.mcgill.ca/"
                    "index.php/starthere'")


def get_advice(cluster_name):

    if cluster_name == "helios":
        return HELIOS_ADVICE
    elif cluster_name == 'mammouth':
        return MAMMOUTH_ADVICE
    elif cluster_name == 'hades':
        return HADES_ADVICE
    elif cluster_name == "guillimin":
        return GUILLIMIN_ADVICE

    return ''


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 strtobool(answer)
        except ValueError:
            if answer == '' and default is not None:
                return 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 is unknown.
        return None
    # 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_launcher(cluster_name):
    if cluster_name == "helios":
        return "msub"
    else:
        return "qsub"


def rethrow_exception(exception, new_message):

    def func_wraper(func):

        def test_func(*args, **kwargs):
            try:
                return func(*args, **kwargs)
            except exception as e:

                orig_exc_type, orig_exc_value, orig_exc_traceback = sys.exc_info()
                new_exc = Exception(new_message)
                new_exc.reraised = True
                new_exc.__cause__ = orig_exc_value

                new_traceback = orig_exc_traceback
                six.reraise(type(new_exc), new_exc, new_traceback)


        return test_func
    return func_wraper


1			import re
2			import hashlib
3			import unicodedata
4			import json
5			import sys
6			import six
7
8			from distutils.util import strtobool
9			from subprocess import Popen, PIPE
10
11			HELIOS_ADVICE = ("On Helios, don't forget that the queue gpu_1, gpu_2, gpu_4"
12			" and gpu_8 give access to a specific amount of gpus."
13			"\nFor more advices, please refer to the official"
14			" documentation 'https://wiki.calculquebec.ca/w/Helios/en'")
15
16			MAMMOUTH_ADVICE = ("On Mammouth, please refer to the official documentation"
17			" for more information:"
18			" 'https://wiki.ccs.usherbrooke.ca/Accueil/en'")
19
20			HADES_ADVICE = ("On Hades, don't forget that the queue name '@hades' needs"
21			" to be used.\nFor more advices, please refer to the"
22			" official documentation: 'https://wiki.calculquebec.ca/w"
23			"/Ex%C3%A9cuter_une_t%C3%A2che/en#tab=tab5'")
24
25			GUILLIMIN_ADVICE = ("On Guillimin, please refer to the official documentation"
26			" for more information: 'http://www.hpc.mcgill.ca/"
27			"index.php/starthere'")
28
29
30			def get_advice(cluster_name):
31
32			if cluster_name == "helios":
33			return HELIOS_ADVICE
34			elif cluster_name == 'mammouth':
35			return MAMMOUTH_ADVICE
36			elif cluster_name == 'hades':
37			return HADES_ADVICE
38			elif cluster_name == "guillimin":
39			return GUILLIMIN_ADVICE
40
41			return ''
42
43
44			def jobname_generator(jobname, job_id):
45			'''Crop the jobname to a maximum of 64 characters.
46			Parameters
47			----------
48			jobname : str
49			Initial jobname.
50			job_id: str
51			ID of the job in the current batch.
52			Returns
53			-------
54			str
55			The cropped version of the string.
56			'''
57			# 64 - 1 since the total length including -1 should be less than 64
58			job_id = str(job_id)
59			if len(jobname) + len(job_id) > 63:
60			croped_string = '{}_{}'.format(jobname[:63 - len(job_id)], job_id)
61			else:
62			croped_string = '{}_{}'.format(jobname, job_id)
63			return croped_string
64
65
66			def print_boxed(string):
67			splitted_string = string.split('\n')
68			max_len = max(map(len, splitted_string))
69			box_line = u"\u2500" * (max_len + 2)
70
71			out = u"\u250c" + box_line + u"\u2510\n"
72			out += '\n'.join([u"\u2502 {} \u2502".format(line.ljust(max_len)) for line in splitted_string])
73			out += u"\n\u2514" + box_line + u"\u2518"
74			print out
75
76
77			def yes_no_prompt(query, default=None):
78			available_prompts = {None: " [y/n] ", 'y': " [Y/n] ", 'n': " [y/N] "}
79
80			if default not in available_prompts:
81			raise ValueError("Invalid default: '{}'".format(default))
82
83			while True:
84			try:
85			answer = raw_input("{0}{1}".format(query, available_prompts[default]))
86			return strtobool(answer)
87			except ValueError:
88			if answer == '' and default is not None:
89			return strtobool(default)
90
91
92			def chunks(sequence, n):
93			""" Yield successive n-sized chunks from sequence. """
94			for i in xrange(0, len(sequence), n):
95			yield sequence[i:i + n]
96
97
98			def generate_uid_from_string(value):
99			""" Create unique identifier from a string. """
100			return hashlib.sha256(value).hexdigest()
101
102
103			def slugify(value):
104			"""
105			Converts to lowercase, removes non-word characters (alphanumerics and
106			underscores) and converts spaces to underscores. Also strips leading and
107			trailing whitespace.
108
109			Reference
110			---------
111			https://github.com/django/django/blob/1.7c3/django/utils/text.py#L436
112			"""
113			value = unicodedata.normalize('NFKD', unicode(value, "UTF-8")).encode('ascii', 'ignore').decode('ascii')
114			value = re.sub('[^\w\s-]', '', value).strip().lower()
115			return str(re.sub('[-\s]+', '_', value))
116
117
118			def encode_escaped_characters(text, escaping_character="\\"):
119			""" Escape the escaped character using its hex representation """
120			def hexify(match):
121			return "\\x{0}".format(match.group()[-1].encode("hex"))
122
123			return re.sub(r"\\.", hexify, text)
124
125
126			def decode_escaped_characters(text):
127			""" Convert hex representation to the character it represents """
128			if len(text) == 0:
129			return ''
130
131			def unhexify(match):
132			return match.group()[2:].decode("hex")
133
134			return re.sub(r"\\x..", unhexify, text)
135
136
137			def save_dict_to_json_file(path, dictionary):
138			with open(path, "w") as json_file:
139			json_file.write(json.dumps(dictionary, indent=4, separators=(',', ': ')))
140
141
142			def load_dict_from_json_file(path):
143			with open(path, "r") as json_file:
144			return json.loads(json_file.read())
145
146
147			def detect_cluster():
148			# Get server status
149			try:
150			output = Popen(["qstat", "-B"], stdout=PIPE).communicate()[0]
151			except OSError:
152			# If qstat is not available we assume that the cluster is unknown.
153			return None
154			# Get server name from status
155			server_name = output.split('\n')[2].split(' ')[0]
156			# Cleanup the name and return it
157			cluster_name = None
158			if server_name.split('.')[-1] == 'm':
159			cluster_name = "mammouth"
160			elif server_name.split('.')[-1] == 'guil':
161			cluster_name = "guillimin"
162			elif server_name.split('.')[-1] == 'helios':
163			cluster_name = "helios"
164			elif server_name.split('.')[-1] == 'hades':
165			cluster_name = "hades"
166			return cluster_name
167
168
169			def get_launcher(cluster_name):
170			if cluster_name == "helios":
171			return "msub"
172			else:
173			return "qsub"
174
175
176			def rethrow_exception(exception, new_message):
177
178			def func_wraper(func):
179
180			def test_func(args, *kwargs):
181			try:
182			return func(args, *kwargs)
183			except exception as e:
184
185			orig_exc_type, orig_exc_value, orig_exc_traceback = sys.exc_info()
186			new_exc = Exception(new_message)
187			new_exc.reraised = True
188			new_exc.__cause__ = orig_exc_value
189
190			new_traceback = orig_exc_traceback
191			six.reraise(type(new_exc), new_exc, new_traceback)
192
193
194			return test_func
195			return func_wraper
196

SMART-Lab / smartdispatch

Pull Request — master (#167)

get_advice() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like