func_wraper() - 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-17 17:57 UTC

func_wraper() A

↳ Parent: rethrow_exception()

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	2
Bugs	0	Features	0

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

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

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):

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

SMART-Lab / smartdispatch

Pull Request — master (#167)

func_wraper() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like