|
| 1 | +from typing import List, Tuple |
| 2 | +from .cffi_bindings import ffi, lib |
| 3 | + |
| 4 | +class Assembly: |
| 5 | + def __init__(self, sequence: str, fragments: List[str], efficiency: float, sub_assemblies: List['Assembly']): |
| 6 | + self.sequence = sequence |
| 7 | + self.fragments = fragments |
| 8 | + self.efficiency = efficiency |
| 9 | + self.sub_assemblies = sub_assemblies |
| 10 | + |
| 11 | +def _create_c_string_array(python_strings: List[str]): |
| 12 | + c_strings = [ffi.new("char[]", s.encode('utf-8')) for s in python_strings] |
| 13 | + c_array = ffi.new("char *[]", c_strings) |
| 14 | + return c_array, c_strings # Return c_strings to keep them alive |
| 15 | + |
| 16 | +def _c_string_array_to_python(c_array, size): |
| 17 | + return [ffi.string(c_array[i]).decode('utf-8') for i in range(size)] |
| 18 | + |
| 19 | +def set_efficiency(overhangs: List[str]) -> float: |
| 20 | + c_overhangs, _ = _create_c_string_array(overhangs) |
| 21 | + return lib.SetEfficiency(c_overhangs, len(overhangs)) |
| 22 | + |
| 23 | +def next_overhangs(current_overhangs: List[str]) -> Tuple[List[str], List[float]]: |
| 24 | + c_overhangs, _ = _create_c_string_array(current_overhangs) |
| 25 | + result = lib.NextOverhangs(c_overhangs, len(current_overhangs)) |
| 26 | + |
| 27 | + if result.error != ffi.NULL: |
| 28 | + raise Exception(ffi.string(result.error).decode('utf-8')) |
| 29 | + |
| 30 | + overhangs = _c_string_array_to_python(result.overhangs, result.size) |
| 31 | + efficiencies = [result.efficiencies[i] for i in range(result.size)] |
| 32 | + return overhangs, efficiencies |
| 33 | + |
| 34 | +def next_overhang(current_overhangs: List[str]) -> str: |
| 35 | + c_overhangs, _ = _create_c_string_array(current_overhangs) |
| 36 | + result = lib.NextOverhang(c_overhangs, len(current_overhangs)) |
| 37 | + return ffi.string(result).decode('utf-8') |
| 38 | + |
| 39 | +def fragment(sequence: str, min_fragment_size: int, max_fragment_size: int, exclude_overhangs: List[str]) -> Tuple[List[str], float, str]: |
| 40 | + c_sequence = ffi.new("char[]", sequence.encode('utf-8')) |
| 41 | + c_exclude_overhangs, _ = _create_c_string_array(exclude_overhangs) |
| 42 | + |
| 43 | + result = lib.FragmentSequence(c_sequence, min_fragment_size, max_fragment_size, c_exclude_overhangs, len(exclude_overhangs)) |
| 44 | + |
| 45 | + if result.error != ffi.NULL: |
| 46 | + error = ffi.string(result.error).decode('utf-8') |
| 47 | + return [], 0.0, error |
| 48 | + |
| 49 | + fragments = _c_string_array_to_python(result.fragments, result.size) |
| 50 | + return fragments, result.efficiency, None |
| 51 | + |
| 52 | +def fragment_with_overhangs(sequence: str, min_fragment_size: int, max_fragment_size: int, |
| 53 | + exclude_overhangs: List[str], include_overhangs: List[str]) -> Tuple[List[str], float, str]: |
| 54 | + c_sequence = ffi.new("char[]", sequence.encode('utf-8')) |
| 55 | + c_exclude_overhangs, _ = _create_c_string_array(exclude_overhangs) |
| 56 | + c_include_overhangs, _ = _create_c_string_array(include_overhangs) |
| 57 | + |
| 58 | + result = lib.FragmentSequenceWithOverhangs(c_sequence, min_fragment_size, max_fragment_size, |
| 59 | + c_exclude_overhangs, len(exclude_overhangs), |
| 60 | + c_include_overhangs, len(include_overhangs)) |
| 61 | + |
| 62 | + if result.error != ffi.NULL: |
| 63 | + error = ffi.string(result.error).decode('utf-8') |
| 64 | + return [], 0.0, error |
| 65 | + |
| 66 | + fragments = _c_string_array_to_python(result.fragments, result.size) |
| 67 | + return fragments, result.efficiency, None |
| 68 | + |
| 69 | +def _assembly_from_c(c_assembly) -> Assembly: |
| 70 | + sequence = ffi.string(c_assembly.sequence).decode('utf-8') |
| 71 | + fragments = _c_string_array_to_python(c_assembly.fragments, c_assembly.fragmentCount) |
| 72 | + efficiency = c_assembly.efficiency |
| 73 | + sub_assemblies = [_assembly_from_c(c_assembly.subAssemblies[i]) for i in range(c_assembly.subAssemblyCount)] |
| 74 | + return Assembly(sequence, fragments, efficiency, sub_assemblies) |
| 75 | + |
| 76 | +def recursive_fragment(sequence: str, max_coding_size_oligo: int, assembly_pattern: List[int], |
| 77 | + exclude_overhangs: List[str], include_overhangs: List[str]) -> Assembly: |
| 78 | + c_sequence = ffi.new("char[]", sequence.encode('utf-8')) |
| 79 | + c_assembly_pattern = ffi.new("int[]", assembly_pattern) |
| 80 | + c_exclude_overhangs, _ = _create_c_string_array(exclude_overhangs) |
| 81 | + c_include_overhangs, _ = _create_c_string_array(include_overhangs) |
| 82 | + |
| 83 | + result = lib.RecursiveFragmentSequence(c_sequence, max_coding_size_oligo, c_assembly_pattern, len(assembly_pattern), |
| 84 | + c_exclude_overhangs, len(exclude_overhangs), |
| 85 | + c_include_overhangs, len(include_overhangs)) |
| 86 | + |
| 87 | + if result.error != ffi.NULL: |
| 88 | + raise Exception(ffi.string(result.error).decode('utf-8')) |
| 89 | + |
| 90 | + return _assembly_from_c(result.assembly) |
0 commit comments