import ctypes
from typing import Callable, Dict, List, Optional, Union
from .. import _binaryninjacore as core
from . import changeset, merge
ProgressFuncType = Callable[[int, int], bool]
NameChangesetFuncType = Callable[['changeset.Changeset'], bool]
ConflictHandlerFuncType = Callable[[Dict[str, 'merge.MergeConflict']], bool]
ConflictHandlerType = Union['merge.ConflictHandler', ConflictHandlerFuncType]
def _last_error() -> str:
"""
Get last error from the api
:return: Last error string
"""
return "Operation failed"
# TODO: keep track of last error in thread
#return core.BNCollaborationGetLastError()
[docs]
def nop(*args, **kwargs):
"""
Function that just returns True, used as default for callbacks
:return: True
"""
return True
[docs]
def wrap_progress(progress_func: ProgressFuncType):
"""
Wraps a progress function in a ctypes function for passing to the FFI
:param progress_func: Python progress function
:return: Wrapped ctypes function
"""
return ctypes.CFUNCTYPE(ctypes.c_bool, ctypes.c_void_p, ctypes.c_ulonglong, ctypes.c_ulonglong)(
lambda ctxt, cur, total: progress_func(cur, total))
[docs]
def wrap_name_changeset(name_changeset_func: NameChangesetFuncType):
"""
Wraps a changeset naming function in a ctypes function for passing to the FFI
:param name_changeset_func: Python changeset naming function
:return: Wrapped ctypes function
"""
return ctypes.CFUNCTYPE(ctypes.c_bool, ctypes.c_void_p, core.BNCollaborationChangesetHandle)(
lambda ctxt, cs: name_changeset_func(changeset.Changeset(handle=cs)))
[docs]
def wrap_conflict_handler(handler: Union[ConflictHandlerFuncType, merge.ConflictHandler]):
"""
Wraps a conflict handler function in a ConflictHandler object so you can be lazy and just use a lambda
:param handler: Python conflict handler function
:return: Wrapped ConflictHandler object
"""
if isinstance(handler, merge.ConflictHandler):
handler_class = handler
else:
class LambdaConflictHandler(merge.ConflictHandler):
def handle(self, conflicts: Dict[str, 'merge.MergeConflict']) -> bool:
return handler(conflicts)
handler_class = LambdaConflictHandler()
return core.BNCollaborationAnalysisConflictHandler(handler_class._handle)
[docs]
def split_progress(progress_func: Optional[ProgressFuncType], subpart: int,
subpart_weights: List[float]) -> ProgressFuncType:
"""
Split a single progress function into equally sized subparts.
This function takes the original progress function and returns a new function whose signature
is the same but whose output is shortened to correspond to the specified subparts.
The length of a subpart is proportional to the sum of all the weights.
E.g. If subpart = 1 and subpartWeights = { 0.25, 0.5, 0.25 }, this will return a function that calls
progress_func and maps its progress to the range [0.25, 0.75]
Internally this works by calling progress_func with total = 1000000 and doing math on the current value
:param progress_func: Original progress function (usually updates a UI)
:param subpart: Index of subpart whose function to return, from 0 to (subpartWeights.size() - 1)
:param subpart_weights: Weights of subparts, described above
:return: A function that will call progress_func() within a modified progress region
"""
if not progress_func:
return lambda cur, total: True
subpart_sum = sum(subpart_weights)
if subpart_sum < 0.00001:
return lambda cur, total: True
# Normalize weights and keep a running count of weights for the start
subpart_starts = []
start = 0
for i in range(len(subpart_weights)):
subpart_starts.append(start)
subpart_weights[i] /= subpart_sum
start += subpart_weights[i]
def inner(cur: int, total: int) -> bool:
steps = 1000000
subpart_size = steps * subpart_weights[subpart]
subpart_progress = float(cur) / float(total) * subpart_size
return progress_func(int(subpart_starts[subpart] * steps + subpart_progress), steps)
return inner
[docs]
class LazyT:
"""
Lazily loaded objects (but FFI)
Pretend this class is templated, because the C++ version is
"""
def __init__(self, ctor: Optional[Callable[[], object]] = None, handle=None):
"""
Create a new LazyT that will be initialized with the result of the given function, when it is first needed.
:param ctor: Function to construct object
:param handle: FFI handle for internal use
"""
if handle is not None:
self._handle = handle
else:
self.ctor = ctor
self.value = None
self._handle = core.BNCollaborationLazyTCreate(ctypes.CFUNCTYPE(ctypes.c_void_p, ctypes.c_void_p)(
lambda ctxt: self._perform_deref()), None)
def _perform_deref(self) -> ctypes.c_void_p:
if self.value is None:
self.value = self.ctor()
result = ctypes.cast(ctypes.py_object(self.value), ctypes.c_void_p)
return result
[docs]
def get(self, expected_type=object):
"""
Access the lazily loaded object. Will construct it if this is the first usage.
:param expected_type: Expected type of result, ctypes will try to cast to it
:return: Result object
"""
result = core.BNCollaborationLazyTDereference(self._handle)
if result is None:
return None
if type == object:
result = ctypes.cast(result, ctypes.py_object)
return result
else:
result = ctypes.cast(result, expected_type)
return result