Skip to content

utils

close_logs 

close_logs(func: Callable)

Wrap a function/method to close the functions first arguments .log attribute FileHandlers after use

Source code in symdesign/protocols/utils.py 
23
24
25
26
27
28
29
30
31
32
def close_logs(func: Callable):
    """Wrap a function/method to close the functions first arguments .log attribute FileHandlers after use"""
    @functools.wraps(func)
    def wrapped(job, *args, **kwargs):
        func_return = func(job, *args, **kwargs)
        # Adapted from https://stackoverflow.com/questions/15435652/python-does-not-release-filehandles-to-logfile
        for handler in job.log.handlers:
            handler.close()
        return func_return
    return wrapped

remove_structure_memory 

remove_structure_memory(func)

Decorator to remove large memory attributes from the instance after processing is complete

Source code in symdesign/protocols/utils.py 
35
36
37
38
39
40
41
42
43
def remove_structure_memory(func):
    """Decorator to remove large memory attributes from the instance after processing is complete"""
    @functools.wraps(func)
    def wrapped(job, *args, **kwargs):
        func_return = func(job, *args, **kwargs)
        if job.job.reduce_memory:
            job.clear_state()
        return func_return
    return wrapped

handle_design_errors 

handle_design_errors(errors: tuple[Type[Exception], ...] = catch_exceptions) -> Callable

Wrap a function/method with try: except errors: and log exceptions to the functions first argument .log attribute

This argument is typically self and is in a class with .log attribute

Parameters:

  • errors (tuple[Type[Exception], ...], default: catch_exceptions ) –

    A tuple of exceptions to monitor. Must be a tuple even if single exception

Returns: Function return upon proper execution, else is error if exception raised, else None

Source code in symdesign/protocols/utils.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def handle_design_errors(errors: tuple[Type[Exception], ...] = catch_exceptions) -> Callable:
    """Wrap a function/method with try: except errors: and log exceptions to the functions first argument .log attribute

    This argument is typically self and is in a class with .log attribute

    Args:
        errors: A tuple of exceptions to monitor. Must be a tuple even if single exception
    Returns:
        Function return upon proper execution, else is error if exception raised, else None
    """
    def wrapper(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapped(job, *args, **kwargs) -> Any:
            try:
                return func(job, *args, **kwargs)
            except errors as error:
                # Perform exception reporting using self.log
                job.report_exception(context=func.__name__)
                return ReportException(str(error))
        return wrapped
    return wrapper

handle_job_errors 

handle_job_errors(errors: tuple[Type[Exception], ...] = catch_exceptions) -> Callable

Wrap a function/method with try/except errors

Parameters:

  • errors (tuple[Type[Exception], ...], default: catch_exceptions ) –

    A tuple of exceptions to monitor. Must be a tuple even if single exception

Returns: Function return upon proper execution, else is error if exception raised, else None

Source code in symdesign/protocols/utils.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def handle_job_errors(errors: tuple[Type[Exception], ...] = catch_exceptions) -> Callable:
    """Wrap a function/method with try/except `errors`

    Args:
        errors: A tuple of exceptions to monitor. Must be a tuple even if single exception
    Returns:
        Function return upon proper execution, else is error if exception raised, else None
    """
    def wrapper(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapped(*args, **kwargs) -> list[Any]:
            try:
                return func(*args, **kwargs)
            except errors as error:
                # Perform exception reporting
                return [ReportException(str(error))]
        return wrapped
    return wrapper

protocol_decorator 

protocol_decorator(errors: tuple[Type[Exception], ...] = catch_exceptions) -> Callable

Wrap a function/method with try: except errors: and log exceptions to the functions first argument .log attribute

This argument is typically self and is in a class with .log attribute

Parameters:

  • errors (tuple[Type[Exception], ...], default: catch_exceptions ) –

    A tuple of exceptions to monitor. Must be a tuple even if single exception

Returns: Function return upon proper execution, else is error if exception raised, else None

Source code in symdesign/protocols/utils.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
def protocol_decorator(errors: tuple[Type[Exception], ...] = catch_exceptions) -> Callable:
    """Wrap a function/method with try: except errors: and log exceptions to the functions first argument .log attribute

    This argument is typically self and is in a class with .log attribute

    Args:
        errors: A tuple of exceptions to monitor. Must be a tuple even if single exception
    Returns:
        Function return upon proper execution, else is error if exception raised, else None
    """
    def wrapper(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapped(job, *args, **kwargs) -> Any:
            # Todo
            #  Ensure that the below setting doesn't conflict with PoseJob inherent setting
            #  job.protocol = job.job.module
            # distribute_protocol()
            if job.job.distribute_work:
                # Skip any execution, instead create the command and add as job.current_script attribute
                base_cmd = list(putils.program_command_tuple) + job.job.get_parsed_arguments()
                base_cmd += ['--single', job.pose_directory]
                # cmd, *additional_cmds = getattr(job, f'get_cmd_{job.protocol}')()
                os.makedirs(job.scripts_path, exist_ok=True)
                job.current_script = distribute.write_script(
                    list2cmdline(base_cmd), name=f'{starttime}_{job.job.module}.sh', out_path=job.scripts_path,
                    # additional=[list2cmdline(_cmd) for _cmd in additional_cmds]
                )
                return None

            logger.info(f'Processing {func.__name__}({repr(job)})')
            # handle_design_errors()
            try:
                func_return = func(job, *args, **kwargs)
            except errors as error:
                # Perform exception reporting using job.log
                job.log.error(error)
                job.log.info(''.join(traceback.format_exc()))
                func_return = ReportException(str(error))
            # remove_structure_memory()
            if job.job.reduce_memory:
                job.clear_state()
            job.protocol = None
            # close_logs()
            # Adapted from https://stackoverflow.com/questions/15435652/python-does-not-release-filehandles-to-logfile
            for handler in job.log.handlers:
                handler.close()

            return func_return
        return wrapped
    return wrapper