rm signatureMaker.py

signatureMaker.py

-import re
-from typing import List
-
-from saqc.funcs import *
-from saqc.core.register import FUNC_MAP
-import os
-
-
-def start_with_exactly_N_spaces(line: str, N: int):
-    return line.startswith(' ' * N) and not line.startswith(' ' * (N + 1))
-
-
-def find_original_signature(fh, func_name):
-    """
-    Extract the signature code from a file.
-    
-    Parameters
-    ----------
-    fh : file
-        file descriptor, to read the code from
-        
-    func_name : str
-        function name, of the signature in question
-
-    Returns
-    -------
-    lines: list or None
-        list of lines of code if found, otherwise None.
-
-    """
-    sig = []
-    start = end = False
-    for line in fh.readlines():
-
-        # find start of signature
-        if not start:
-
-            if line.startswith(f'def {func_name}'):
-                sig.append(line)
-                start = True
-            continue
-
-        # find end of signature
-        if '"""' in line or start_with_exactly_N_spaces(line, 4):
-            end = True
-            break  # do not append line
-
-        # found last line of signature
-        if '->' in line:
-            end = True
-
-        sig.append(line)
-
-        if end:
-            break
-
-    # if end or/and start was not found,
-    # something went wrong
-    if end is False:
-        sig = None
-
-    return sig
-
-
-def replace_core_signatures(core_code: List[str], sig_code: List[str], func_name: str, target_file):
-    """
-    Replace a signature in the core code with a signature from a module.
-
-    Parameters
-    ----------
-    core_code : list
-        lines of code, one by one
-
-    sig_code : list
-        lines of code, one by one (only the signature in question)
-
-    func_name : str
-        function name in question
-
-    target_file : file
-        file descriptor to write the modified code to
-    """
-    start = end = False
-    for line in core_code:
-
-        # append the rest of the file, the loop ends here
-        if end is True:
-            target_file.write(line)
-            continue
-
-        # find start of signature, loop starts here
-        if not start:
-
-            if line.startswith(f'    def {func_name}'):
-                start = True
-
-                # insert the replacement
-                for rline in sig_code:
-                    target_file.write('    ')
-                    target_file.write(rline)
-
-            # start of sig, not found yet
-            else:
-                target_file.write(line)
-                continue
-
-        # found line after end of signature
-        if '"""' in line or start_with_exactly_N_spaces(line, 8):
-            end = True
-            target_file.write(line)
-            continue
-
-        # found last line of signature
-        if '->' in line:
-            end = True
-            continue
-
-
-def replace_datafieldflagger(lines):
-    """
-    Remove 'data' and 'flagger' from signature, and insert 'self' instead.
-    """
-    empty = re.compile(' *\n')
-    data = re.compile('.*(data[=: ][^,]*, ?)')  # eg. 'data: DictOfSeries,'
-    flagger = re.compile('.*(flagger[=: ][^,]*, ?)')  # eg. 'flagger: Flagger,'
-    pattern_list = [data, flagger]
-    i = 0
-    replaced = []
-
-    for line in lines:
-
-        if i < len(pattern_list):
-
-            # search for one patter after the other in the current line,
-            # if any is NOT found, we stop and continued from there in the
-            # next line (next loop outer integration)
-            for j in range(i, len(pattern_list)):
-                found = pattern_list[i].match(line)
-                if found:
-                    # we replace the first match ('data') with 'self'
-                    line = line.replace(found[1], 'self, ' if i == 0 else '', 1)
-                    i += 1  # next pattern please
-                else:
-                    break
-
-        empty_line = empty.match(line)
-        if empty_line:
-            continue
-
-        replaced.append(line)
-
-    return replaced
-
-
-def autoreplace_signatures():
-    """
-    Replaces core-signatures with the module-signatures, one-by-one.
-    """
-    postfix = '_autosignature'
-    saqc_path = 'saqc/core/modules/'
-    touched_modules = []
-
-    # one-by-one: we only process one signature at a time, this means
-    # that we see most files multiple times.
-    for name in FUNC_MAP:
-        module, fname = name.split('.')
-
-        with open(f'saqc/funcs/{module}.py', 'r') as fh:
-            lines = find_original_signature(fh, fname)
-
-        if lines is None:
-            warnings.warn(f"end of signature of '{fname}' not found - ignoring")
-            continue
-
-        # modify original function signature
-        lines = replace_datafieldflagger(lines)
-        print(''.join(lines))
-
-        # find the right file. If we already processed a signature
-        # of the same module, we already have a modified file, so we
-        # need to read and write the same.
-        readfile = f'{saqc_path}{module}.py'
-        writefile = f'{saqc_path}{module}{postfix}.py'
-        if module in touched_modules:
-            readfile = writefile
-        else:
-            touched_modules.append(module)
-
-        # READ
-        with open(readfile, 'r') as fh:
-            readlines = fh.readlines()
-
-        # WRITE
-        # replace sig's in Saqc.module to a temporary file
-        with open(writefile, 'w') as fh:
-            replace_core_signatures(readlines, lines, fname, fh)
-
-    # replace all original files with the temporary ones
-    files = os.listdir('saqc/core/modules/')
-    for new in files:
-        if postfix in new:
-            old = saqc_path + new.replace(postfix, '')
-            new = saqc_path + new
-            os.replace(new, old)
-
-if __name__ == '__main__':
-    autoreplace_signatures()