Skip to content

Commit

Permalink
Update
Browse files Browse the repository at this point in the history
  • Loading branch information
@mraveri
mraveri committed Dec 29, 2024
1 parent 7de933b commit 0887dd0
Show file tree
Hide file tree
Showing 3 changed files with 133 additions and 102 deletions.
20 changes: 9 additions & 11 deletions tensiometer/tests/utilities_color_test.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,8 @@

import unittest

from tensiometer.utilities.color_utilities import color_linear_interpolation, nice_colors, bash_colors
from tensiometer.utilities.color_utilities import color_linear_interpolation, nice_colors
import tensiometer.utilities.color_utilities as cu

###############################################################################

Expand Down Expand Up @@ -55,35 +56,32 @@ def test_invalid_output_format(self):

class TestBashColors(unittest.TestCase):

def setUp(self):
self.bash_color = bash_colors()

def test_purple(self):
result = self.bash_color.purple('Test')
result = cu.bash_purple('Test')
self.assertEqual(result, '\033[95mTest\033[0m')

def test_blue(self):
result = self.bash_color.blue('Test')
result = cu.bash_blue('Test')
self.assertEqual(result, '\033[94mTest\033[0m')

def test_green(self):
result = self.bash_color.green('Test')
result = cu.bash_green('Test')
self.assertEqual(result, '\033[92mTest\033[0m')

def test_yellow(self):
result = self.bash_color.yellow('Test')
result = cu.bash_yellow('Test')
self.assertEqual(result, '\033[93mTest\033[0m')

def test_red(self):
result = self.bash_color.red('Test')
result = cu.bash_red('Test')
self.assertEqual(result, '\033[91mTest\033[0m')

def test_bold(self):
result = self.bash_color.bold('Test')
result = cu.bash_bold('Test')
self.assertEqual(result, '\033[1mTest\033[0m')

def test_underline(self):
result = self.bash_color.underline('Test')
result = cu.bash_underline('Test')
self.assertEqual(result, '\033[4mTest\033[0m')

###############################################################################
Expand Down
153 changes: 65 additions & 88 deletions tensiometer/utilities/color_utilities.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -161,115 +161,92 @@ def nice_colors(num, colormap='the_gold_standard', interpolation_method='linear'
# ******************************************************************************
# Definition of bash colors:

class bash_colors:
BASH_PURPLE = '\033[95m' #: ANSI color for light purple.
BASH_BLUE = '\033[94m' #: ANSI color for blue.
BASH_GREEN = '\033[92m' #: ANSI color for green.
BASH_YELLOW = '\033[93m' #: ANSI color for yellow.
BASH_RED = '\033[91m' #: ANSI color for red.
BASH_BOLD = '\033[1m' #: ANSI code for bold text.
BASH_UNDERLINE = '\033[4m' #: ANSI code for underlined text.
BASH_ENDC = '\033[0m' #: ANSI code to restore the bash default.

# --------------------------------------------------------------------------

def bash_purple(string):
"""
This class contains the necessary definitions to print to bash screen with colors.
Sometimes it can be useful and nice!
:ivar PURPLE: ANSI color for light purple.
:ivar BLUE: ANSI color for blue.
:ivar GREEN: ANSI color for green.
:ivar YELLOW: ANSI color for yellow.
:ivar RED: ANSI color for red.
:ivar BOLD: ANSI code for bold text.
:ivar UNDERLINE: ANSI code for underlined text.
:ivar ENDC: ANSI code to restore the bash default.
Function that returns a string that can be printed to bash in :class:`tensiometer.utilities.color_utilities.bash_colors.BASH_PURPLE` color.
:param string: input string.
:return: the input string with the relevant ANSI code at the beginning and at the end.
"""
return BASH_PURPLE + str(string) + BASH_ENDC

# --------------------------------------------------------------------------
# --------------------------------------------------------------------------

PURPLE = '\033[95m' #: ANSI color for light purple.
BLUE = '\033[94m' #: ANSI color for blue.
GREEN = '\033[92m' #: ANSI color for green.
YELLOW = '\033[93m' #: ANSI color for yellow.
RED = '\033[91m' #: ANSI color for red.
BOLD = '\033[1m' #: ANSI code for bold text.
UNDERLINE = '\033[4m' #: ANSI code for underlined text.
ENDC = '\033[0m' #: ANSI code to restore the bash default.

# --------------------------------------------------------------------------

def __init__(self):
pass

# --------------------------------------------------------------------------

def purple(self, string):
"""
Function that returns a string that can be printed to bash in :class:`tensiometer.utilities.color_utilities.bash_colors.PURPLE` color.
:param string: input string.
:return: the input string with the relevant ANSI code at the beginning and at the end.
"""
return self.PURPLE+str(string)+self.ENDC

# --------------------------------------------------------------------------

def blue(self,string):
"""
Function that returns a string that can be printed to bash in :class:`tensiometer.utilities.color_utilities.bash_colors.BLUE` color.
def bash_blue(string):
"""
Function that returns a string that can be printed to bash in :class:`tensiometer.utilities.color_utilities.bash_colors.BASH_BLUE` color.
:param string: input string.
:return: the input string with the relevant ANSI code at the beginning and at the end.
"""
return self.BLUE+str(string)+self.ENDC
:param string: input string.
:return: the input string with the relevant ANSI code at the beginning and at the end.
"""
return BASH_BLUE + str(string) + BASH_ENDC

# --------------------------------------------------------------------------
# --------------------------------------------------------------------------

def green(self,string):
"""
Function that returns a string that can be printed to bash in :class:`tensiometer.utilities.color_utilities.bash_colors.GREEN` color.
def bash_green(string):
"""
Function that returns a string that can be printed to bash in :class:`tensiometer.utilities.color_utilities.bash_colors.BASH_GREEN` color.
:param string: input string.
:return: the input string with the relevant ANSI code at the beginning and at the end.
"""
return self.GREEN+str(string)+self.ENDC
:param string: input string.
:return: the input string with the relevant ANSI code at the beginning and at the end.
"""
return BASH_GREEN + str(string) + BASH_ENDC

# --------------------------------------------------------------------------
# --------------------------------------------------------------------------

def yellow(self,string):
"""
Function that returns a string that can be printed to bash in :class:`tensiometer.utilities.color_utilities.bash_colors.YELLOW` color.
def bash_yellow(string):
"""
Function that returns a string that can be printed to bash in :class:`tensiometer.utilities.color_utilities.bash_colors.BASH_YELLOW` color.
:param string: input string.
:return: the input string with the relevant ANSI code at the beginning and at the end.
"""
return self.YELLOW+str(string)+self.ENDC
:param string: input string.
:return: the input string with the relevant ANSI code at the beginning and at the end.
"""
return BASH_YELLOW + str(string) + BASH_ENDC

# --------------------------------------------------------------------------
# --------------------------------------------------------------------------

def red(self,string):
"""
Function that returns a string that can be printed to bash in :class:`tensiometer.utilities.color_utilities.bash_colors.RED` color.
def bash_red(string):
"""
Function that returns a string that can be printed to bash in :class:`tensiometer.utilities.color_utilities.bash_colors.BASH_RED` color.
:param string: input string.
:return: the input string with the relevant ANSI code at the beginning and at the end.
"""
return self.RED+str(string)+self.ENDC
:param string: input string.
:return: the input string with the relevant ANSI code at the beginning and at the end.
"""
return BASH_RED + str(string) + BASH_ENDC

# --------------------------------------------------------------------------
# --------------------------------------------------------------------------

def bold(self,string):
"""
Function that returns a string that can be printed to bash in :class:`tensiometer.utilities.color_utilities.bash_colors.BOLD` color.
def bash_bold(string):
"""
Function that returns a string that can be printed to bash in :class:`tensiometer.utilities.color_utilities.bash_colors.BASH_BOLD` color.
:param string: input string.
:return: the input string with the relevant ANSI code at the beginning and at the end.
"""
return self.BOLD+str(string)+self.ENDC
:param string: input string.
:return: the input string with the relevant ANSI code at the beginning and at the end.
"""
return BASH_BOLD + str(string) + BASH_ENDC

# --------------------------------------------------------------------------
# --------------------------------------------------------------------------

def underline(self,string):
"""
Function that returns a string that can be printed to bash in :class:`tensiometer.utilities.color_utilities.bash_colors.UNDERLINE` color.
def bash_underline(string):
"""
Function that returns a string that can be printed to bash in :class:`tensiometer.utilities.color_utilities.bash_colors.BASH_UNDERLINE` color.
:param string: input string.
:return: the input string with the relevant ANSI code at the beginning and at the end.
"""
return self.UNDERLINE+str(string)+self.ENDC
:param string: input string.
:return: the input string with the relevant ANSI code at the beginning and at the end.
"""
return BASH_UNDERLINE + str(string) + BASH_ENDC

# --------------------------------------------------------------------------
# --------------------------------------------------------------------------

# ******************************************************************************
62 changes: 59 additions & 3 deletions tensiometer/utilities/subprocess_runner.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -171,25 +171,81 @@ def target_function(pipe_conn, *args, **kwargs):
# Set up and start the subprocess.
ctx = mp.get_context(context)
parent_conn, child_conn = ctx.Pipe()

# create the process:
process = ctx.Process(target=target_function, args=(child_conn, *args), kwargs=kwargs)
if feedback_level > 2:
print('* Sub-process created.', flush=True)

process.start()
if feedback_level > 2:
print('* Sub-process started.', flush=True)

# initial memory usage:
if monitoring:
initial_memory = psutil.Process(process.pid).memory_info().rss / 1024 / 1024
if feedback_level > 1:
print('* Initial memory usage:', initial_memory, 'MB', flush=True)
print(feedback_separator, flush=True)
peak_memory = initial_memory / 1024 / 1024

# Monitor the process and handle timeout.
initial_time = time.time() if monitoring else None

while True:
if not process.is_alive():

# exit by process status:
_process_status = process.is_alive()
if feedback_level > 2:
print('* Process is alive:', _process_status, flush=True)
if not _process_status:
if feedback_level > 2:
print(f'* Process exited with code {process.exitcode}.', flush=True)
if process.exitcode == 0:
break
else:
raise Exception(f'Process exited with code {process.exitcode}')
raise Exception('Process exited with code %d' % process.exitcode)

# monitoring:
if monitoring:
current_memory = psutil.Process(process.pid).memory_info().rss / 1024 / 1024
if feedback_level > 2:
print('* Current memory usage:', current_memory, 'MB', flush=True)
peak_memory = max(peak_memory, current_memory)

# break if pipe is full:
if parent_conn.poll():
if feedback_level > 2:
print('* Breaking, sub-process pipe is full.', flush=True)
break

# break by timeout:
if timeout and time.time() - initial_time > timeout:
process.terminate()
raise TimeoutError("Process timed out.")

# sleep for monitoring frequency:
time.sleep(monitoring_frequency)

# Retrieve the result from the pipe.
# Receive the result from the pipe
result = parent_conn.recv()
if feedback_level > 2:
print('* Result received.', flush=True)

# Wait for the process to finish and get the result:
process.join()
if feedback_level > 2:
print('* Process joined.', flush=True)

# monitoring stats if needed:
if monitoring:
final_time = time.time()
total_time = final_time - initial_time
if feedback_level > 0:
print(feedback_separator, flush=True)
print(f'* Total time elapsed: {total_time:.2f} seconds', flush=True)
print(f'* Peak memory usage: {peak_memory:.2f} MB', flush=True)
print(feedback_separator, flush=True)

# Raise exceptions received from the subprocess.
if isinstance(result, Exception):
Expand Down

0 comments on commit 0887dd0

Please sign in to comment.