socube.utils package¶

Submodules¶

socube.utils.concurrence module¶

socube.utils.concurrence.parallel(func)¶

A decorator for asynchronous method

Parameters

func (Callable) – The function to be decorated as asynchronous.

Returns

a new function. If it is called under ParallelManager, it will be
executed in a process pool.

Examples

>>> @parallel
    def func(*args, **kwargs):
        ...
>>> parallelFunc = parallel(func)

class socube.utils.concurrence.ParallelManager(verbose: bool = False, max_workers: int = 8, paral_type: str = 'thread', raise_error: bool = True)¶

Bases: socube.utils.context.ContextManagerBase

A context manager for multi-thread. In this context, Multiple thread is enable will be set as True and enable multi-thread for function decorated with parallel. Multi-level nesting is allowed, and each level has an independent asynchronous task pool. When exiting the context, it will wait for the tasks in the current asynchronous task pool to complete.

Parameters

verbose (bool) – Whether to print log.
max_workers (int) – The maximum number of workers in the asynchronous task pool.
paral_type (str) – The type of asynchronous task pool. Only support “thread” and “process”.
raise_error (bool) – Whether to raise error when the task is not completed. Default is True. If you do not want interrupt the program, set it to False.

Examples

>>> with ThreadManager() as tm:
        # Here is you code
        asynFunc1(*args)  # this function is decorated with `parallel`
        with tm:
            asynFunc2(*args)  # this function is decorated with `parallel`
        print("asynFunc2 finished") # this code will not be executed util asynFunc2 finished.
    print("asynFunc1 finished")  # this code will not be executed util asynFunc1finished.

submit(func: Callable, *args, **kwargs) → concurrent.futures._base.Future¶: Add a task to the asynchronous task pool. For muti-thread task, you can use parallel decorator instead. But it not support multi-process task.

socube.utils.context module¶

class socube.utils.context.ContextManagerBase¶

Bases: object

Basical interface for context manager.

socube.utils.exception module¶

class socube.utils.exception.ExceptionManager(mail_service: socube.utils.mail.MailService = None, is_continue: bool = False, ignore_keyboard_interrupt: bool = False, ignore_normal_system_exit: bool = True)¶

Bases: socube.utils.context.ContextManagerBase

A context manager to deal with exception and notify users if it is required. Any exception occurred within this context will be catched and send notification email to user if mailService is not equal to ‘None’. You use it as following.

Parameters

mail_service (MailService) – A MailService object used to send email to users, If None, no email will be sent
is_continue (bool) – If continue to run this program when an error occurred.
ignore_keyboard_interrupt (bool) – If true, it will ignore keyboard interrupt signal, such as “Ctrl + C”. It prevent quit program by pressing keyboard unexpectedly. But it also is a shortcoming when you want to terminate program.

Examples

>>> mailService = MailService(...)
>>> with ExceptionManager(mailService)
>>>     a = 1/0  # will throw ZeroDivideError

setMailService(mail_service: socube.utils.mail.MailService) → None¶: Set the mail service object.

setNormalInfo(text: str) → None¶

Set the notification email when quiting context without any exception occurred. This can be used to send a processing result. Result should as short as possible.

Parameters: email main text (text) –

>>> with ExceptionManager(mailService) as em:
>>>     # Here are some processing codes
>>>     em.setNormalInfo("Prcoessing results")

socube.utils.io module¶

socube.utils.io.uniquePath(path: str, exists_warning: bool = False) → str¶

Check whether path existed in the file system. And generate a unique path by add index at the end of raw path.

Parameters

path (str) – The raw path to be checked.
exists_warning (bool) – Whether to print warning when the path is existed.

Returns

Return type

str, the unique path.

Examples

>>> # pkgs/lapjv-1.3.1.tar.gz already existed
>>> uniquePath("pkgs/lapjv-1.3.1.tar.gz")
'pkgs/lapjv-1.3.1.tar(1).gz'

>>> # pkgs/lapjv-1.3.1.tar.gz already existed and
>>> # pkgs/lapjv-1.3.1.tar(1).gz already existed
>>> uniquePath("pkgs/lapjv-1.3.1.tar.gz")
'pkgs/lapjv-1.3.1.tar(2).gz'

socube.utils.io.mkDirs(path: str) → None¶

Create a directory if it not exists. If the parent is not existed, it will be created as well.

Parameters: path (str) – The path to be created.

Examples

>>> #parent is not existed
>>> mkDirs("parent/target/")

socube.utils.io.rm(path: str) → None¶

Remove a directory or file.

Parameters: path (str) – The path to be removed.

Examples

>>> rm("targetFile")

socube.utils.io.checkExist(path: str, types: str = 'file', raise_error: bool = True) → bool¶

Check whether a file or directory is existed.

Parameters

path (str) – The path to be checked.
types (str) – The type of the path. “file” or “dir”.
raise_error (bool) – Whether to raise error when the path is not existed.

Returns

Return type

bool, whether the path is existed.

Examples

>>> checkExist("/public/file")
>>> checkExist("/public/home/", types="dir")
>>> checkExist("/public/home/", types="dir", raiseError=False)

socube.utils.io.writeCsv(data: Union[pandas.core.generic.NDFrame, numpy.ndarray], file: str, callback: Optional[Callable] = None, **kwargs) → None¶

Write ndarray or any instance of NDFrame to a CSV file.

Parameters

data (Union[NDFrame, np.ndarray]) – The data to be written.
file (str) – The path to be written.
callback (Optional[Callable]) – The callback function to be called after writing.
**kwargs – The keyword arguments to be passed to pandas.DataFrame.to_csv.

Examples

>>> ndarray = np.array([1, 2, 3])
>>> writeCsv(ndarray, "ndarray.csv", callback=lambda:print("Finish"))
>>> series = pd.Series(ndarray)
>>> writeCsv(series, "series.csv")

socube.utils.io.writeNpy(data: numpy.ndarray, file: str, callback: Optional[Callable] = None, allow_pickle: bool = True, **kwargs)¶

Write ndarray to NPY binary file. NPY format is numpy’s specific file format.

Parameters

data (np.ndarray) – The data to be written.
file (str) – The path to be written.
callback (Optional[Callable]) – The callback function to be called after writing.
allow_pickle (bool) – Wether to allow use pickle to serialize python object.
**kwargs – The keyword arguments to be passed to numpy.save.

Examples

>>> ndarray = np.array([1, 2, 3])
>>> writeNpy(ndarray, "ndarray.npy")

socube.utils.io.writeHdf(data: pandas.core.generic.NDFrame, file: str, key='data', mode='w', callback: Optional[Callable] = None, **kwargs)¶

Write NDFrame instance to a HDF5 binary format file.

Parameters

data (NDFrame) – The data to be written.
file (str) – The path to be written.
key (str) – The key to be used to store the data.
mode (str) – The mode to be used to open the file.
callback (Optional[Callable]) – The callback function to be called after writing.
**kwargs – The keyword arguments to be passed to pandas.DataFrame.to_hdf.

Examples

>>> data = pd.DataFrame({"a": [1, 2, 3], "b": [4, 5, 6]})
>>> writeHdf(data, "data.hdf", callback=lambda:print("Finish"))

socube.utils.io.loadTorchModule(module: torch.nn.modules.module.Module, file: str, skipped: bool = True, verbose=False)¶

Load a torch module from a file.

Parameters

module (torch.nn.Module) – The module waited to be updated.
file (str) – The module file path.
skipped (bool) – Whether to skip the loading of the module when the file is not existed.
verbose (bool) – Whether to print the loading message.

class socube.utils.io.Redirect(target: str, verbose: bool = True)¶

Bases: socube.utils.context.ContextManagerBase

A context manager to redirect stdout and stderr.

Parameters

target (str) – The redirect target file.
verbose (bool) – Whether to print the redirect message.

Examples

>>> with Redirect("stdout.txt", verbose=True):
>>>     print("Hello")
>>>     print("World")

class socube.utils.io.ReportManager(reports: Dict[str, pandas.core.generic.NDFrame] = {}, verbose: bool = False)¶

Bases: socube.utils.context.ContextManagerBase

A context manager to save important data.

Parameters

reports (Dict[str, NDFrame]) – a dictionary of reports, key is report filename, value is a NDFrame object waited to be saved.
verbose (bool) – Whether to print the saving message. If True, ReportManager will work in silence.

Examples

>>> with ReportManager({"report1": df1, "report2": df2}, verbose=True) as rm:
>>>     print("Hello")
>>>     print("World")
>>>     rm.addReports("report1", df3)
>>>     rm.addReports("report2", df4)

addReports(reports: Dict[str, pandas.core.generic.NDFrame]) → None¶

add new reports in context

Parameters: reports (Dict[str, NDFrame]) – a dictionary of reports, key is report filename, value is a NDFrame object waited to be saved.

updateReportName(old: str, new: str) → None¶

Update the name of a report.

Parameters

old (str) – The old report name.
new (str) – The new report name.

socube.utils.logging module¶

socube.utils.logging.log(name: str, msg: str, level: str = 'info', quiet: bool = False, *args, **kwargs) → None¶

Log a message.

Parameters

name (str) – The name of the logger.
msg (str) – The message to be logged.
level (str) – The level of the message.
quiet (bool) – If True, the message will not be logged.
*args – The rest of the arguments will be passed to the logger.
**kwargs – The rest of the keyword arguments will be passed to the logger.

Examples

>>> log("mylogger", "hello world", level="debug")
>>> log("mylogger", "hello world", level="info")
>>> log("mylogger", "hello world", level="warn")
>>> log("mylogger", "hello world", level="error")

socube.utils.logging.getJobId() → str¶

Generate job id randomly. It’s useful to seperate different jobs.

Returns
Return type: a string id

Examples

>>> getJobId()
'20211102-141041-237'

>>> getJobId()
'20211102-132411-806'

socube.utils.mail module¶

This module provides some mail service tools

class socube.utils.mail.MailService(sender: str, passwd: str, server: str, port: int, nickname: str = 'SoCube', ssl: bool = True, verbose: bool = False)¶

Bases: socube.utils.context.ContextManagerBase

Provide basic SMTP mail service for mail sending. SMTP is the abbreviation of Simple Mail Transfer Protocol. For detail, please search it in google or bing. Context manager is supported. You can use it as following.

Parameters

sender (str) – email address as a sender
passwd (str) – sender’s password for SMTP service. For some SMTP service vendors, this password is the same as password used to visit mail website, such as Zhejiang University Mail Service. But for other vendors like QQ Mail Service, you need use independent password for SMTP service. For detail, you need visit related mail service vendors.
server (str) – the domain of SMTP server. such as ‘smtp.zju.edu.cn’,’smtp.qq.com’, which provided by the mail service vendors.
port (int) – web port for SMTP service provided by service vendors.
nickname (str) – the nickname of sender when sending a mail
ssl (bool) – If True, use SSL for security.
verbose (bool) – If True, mail service will run in background without any output info.

>>> with MailService("account@gmail.com", "password", "smtp.gmail.com", 465) as ms
>>>     ms.sendText(...)

login() → bool¶: Login to enable service. After logining, you should use service before connection timeout.

quit()¶: Quit to close service and release resources occuppied

sendSelf(subject: str, text: str, attachments: List[str] = []) → bool¶

Send a plain text mail without any attachment to sendor himself.

Parameters

subject (str) – the subject of email.
text (str) – email’s main text. rich html format is not supported.
attachments (List[str]) – a list of attachments’ filename. attachments which are not existed will be skipped automatically

Returns

Return type

bool, True if sending succeed, False if sending failed.

sendText(to_dict: Dict[str, str], subject: str, text: str, attachments: List[str] = []) → bool¶

Send a plain text email with or without attachments

Parameters

to_dict (Dict[str, str]) – a dict of email address and nickname.
subject (str) – the subject of email.
text (str) – email’s main text. rich html format is not supported.
attachments (List[str]) – a list of attachments’ filename. attachments which are not existed will be skipped automatically

Returns

Return type

bool, True if sending succeed, False if sending failed.

property status¶: Is service available

socube.utils.memory module¶

socube.utils.memory.visualBytes(size: int) → str¶

Make memory size more friendly with proper unit.

Parameters: size (a int type of bytes szie) –
Returns
Return type: a formated string with size unit.

Examples

>>> visualBytes(2<<20)
'2.00MB'
>>> visualBytes(18200565665)
'16.95GB'

socube.utils.memory.getGPUReport(*devices: int) → str¶

Get gpu memory usage of this program by pytorch.

Parameters: devices (a list of device id) –
Returns
Return type: A string report of gpu memory

socube.utils.memory.autoClearIter(iter: Iterable[Any]) → email.generator.Generator¶: Create a generator to automatically clean GPU memory when iterating.

Module contents¶

This module and its submodules provide some basic tools