<commit_before>#!/usr/bin/env python

import weakref
import logging

l = logging.getLogger("claripy.frontends.replacement_frontend")

from .constrained_frontend import ConstrainedFrontend

class ReplacementFrontend(ConstrainedFrontend):
    def __init__(self, actual_frontend, allow_symbolic=None, replacements=None, replacement_cache=None, unsafe_replacement=None, complex_auto_replace=None, auto_replace=None, replace_constraints=None, **kwargs):
        super(ReplacementFrontend, self).__init__(**kwargs)
        self._actual_frontend = actual_frontend
        self._allow_symbolic = True if allow_symbolic is None else allow_symbolic
        self._auto_replace = True if auto_replace is None else auto_replace
        self._complex_auto_replace = False if complex_auto_replace is None else complex_auto_replace
        self._replace_constraints = False if replace_constraints is None else replace_constraints
        self._unsafe_replacement = False if unsafe_replacement is None else unsafe_replacement
        self._replacements = {} if replacements is None else replacements
        self._replacement_cache = weakref.WeakKeyDictionary() if replacement_cache is None else replacement_cache

        self._validation_frontend = None

    def _blank_copy(self, c):
        super(ReplacementFrontend, self)._blank_copy(c)
        c._actual_frontend = self._actual_frontend.blank_copy()
        c._allow_symbolic = self._allow_symbolic
        c._auto_replace = self._auto_replace
        c._complex_auto_replace = self._complex_auto_replace
        c._replace_constraints = self._replace_constraints
        c._unsafe_replacement = self._unsafe_replacement
        c._replacements = {}
        c._replacement_cache = weakref.WeakKeyDictionary()

        if self._validation_frontend is not None:
            c._validation_frontend = self._validation_frontend.blank_copy()
        else:
            c._validation_frontend = None

    def _copy(self, c):
        super(ReplacementFrontend, self)._copy(c)
        self._actual_frontend._copy(c._actual_frontend)
        if self._validation_frontend is not None:
            self._validation_frontend._copy(c._validation_frontend)

        c._replacements = self._replacements
        c._replacement_cache = self._replacement_cache

    #
    # Replacements
    #

    def add_replacement(self, old, new, invalidate_cache=True, replace=True, promote=True):
        if not isinstance(old, Base):
            return

        if old is new:
            return

        if not replace and old.cache_key in self._replacements:
            return

        if not promote and old.cache_key in self._replacement_cache:
            return

        if not isinstance(new, Base):
            if isinstance(new, bool):
                new = BoolV(new)
            elif isinstance(new, (int, long)):
                new = BVV(new, old.length)
            else:
                return

        if invalidate_cache:
            self._replacements = dict(self._replacements)
            self._replacement_cache = weakref.WeakKeyDictionary(self._replacements)

        self._replacements[old.cache_key] = new
        self._replacement_cache[old.cache_key] = new

    def _replacement(self, old):
        if not isinstance(old, Base):
            return old

        try:
            return self._replacement_cache[old.cache_key]
        except KeyError:
            # not found in the cache
            new = old.replace_dict(self._replacement_cache)
            if new is not old:
                self._replacement_cache[old.cache_key] = new
            return new

    def _add_solve_result(self, e, er, r):
        if not self._auto_replace:
            return
        if not isinstance(e, Base) or not e.symbolic:
            return
        if er.symbolic:
            return
        self.add_replacement(e, r, invalidate_cache=False)

    #
    # Storable support
    #

    def downsize(self):<commit_msg>Added methods to safely allow removing replacements<commit_after>#!/usr/bin/env python

import weakref
import logging

l = logging.getLogger("claripy.frontends.replacement_frontend")

from .constrained_frontend import ConstrainedFrontend

class ReplacementFrontend(ConstrainedFrontend):
    def __init__(self, actual_frontend, allow_symbolic=None, replacements=None, replacement_cache=None, unsafe_replacement=None, complex_auto_replace=None, auto_replace=None, replace_constraints=None, **kwargs):
        super(ReplacementFrontend, self).__init__(**kwargs)
        self._actual_frontend = actual_frontend
        self._allow_symbolic = True if allow_symbolic is None else allow_symbolic
        self._auto_replace = True if auto_replace is None else auto_replace
        self._complex_auto_replace = False if complex_auto_replace is None else complex_auto_replace
        self._replace_constraints = False if replace_constraints is None else replace_constraints
        self._unsafe_replacement = False if unsafe_replacement is None else unsafe_replacement
        self._replacements = {} if replacements is None else replacements
        self._replacement_cache = weakref.WeakKeyDictionary() if replacement_cache is None else replacement_cache

        self._validation_frontend = None

    def _blank_copy(self, c):
        super(ReplacementFrontend, self)._blank_copy(c)
        c._actual_frontend = self._actual_frontend.blank_copy()
        c._allow_symbolic = self._allow_symbolic
        c._auto_replace = self._auto_replace
        c._complex_auto_replace = self._complex_auto_replace
        c._replace_constraints = self._replace_constraints
        c._unsafe_replacement = self._unsafe_replacement
        c._replacements = {}
        c._replacement_cache = weakref.WeakKeyDictionary()

        if self._validation_frontend is not None:
            c._validation_frontend = self._validation_frontend.blank_copy()
        else:
            c._validation_frontend = None

    def _copy(self, c):
        super(ReplacementFrontend, self)._copy(c)
        self._actual_frontend._copy(c._actual_frontend)
        if self._validation_frontend is not None:
            self._validation_frontend._copy(c._validation_frontend)

        c._replacements = self._replacements
        c._replacement_cache = self._replacement_cache

    #
    # Replacements
    #

    def add_replacement(self, old, new, invalidate_cache=True, replace=True, promote=True):
        if not isinstance(old, Base):
            return

        if old is new:
            return

        if not replace and old.cache_key in self._replacements:
            return

        if not promote and old.cache_key in self._replacement_cache:
            return

        if not isinstance(new, Base):
            if isinstance(new, bool):
                new = BoolV(new)
            elif isinstance(new, (int, long)):
                new = BVV(new, old.length)
            else:
                return

        if invalidate_cache:
            self._replacements = dict(self._replacements)
            self._replacement_cache = weakref.WeakKeyDictionary(self._replacements)

        self._replacements[old.cache_key] = new
        self._replacement_cache[old.cache_key] = new

    def remove_replacements(self, old_entries):
        self._replacements = {k: v for k, v in self._replacements if k not in old_entries}
        self._replacement_cache = weakref.WeakKeyDictionary(self._replacements)

    def clear_replacements(self):
        self._replacements = dict()
        self._replacement_cache = weakref.WeakKeyDictionary(self._replacements)

    def _replacement(self, old):
        if not isinstance(old, Base):
            return old

        try:
            return self._replacement_cache[old.cache_key]
        except KeyError:
            # not found in the cache
            new = old.replace_dict(self._replacement_cache)
            if new is not old:
                self._replacement_cache[old.cache_key] = new
            return new

    def _add_solve_result(self, e, er, r):
        if not self._auto_replace:
            return
        if not isinstance(e, Base) or not e.symbolic:
            return
        if er.symbolic:
            return
        self.add_replacement(e, r, invalidate_cache=False)

    #
    # Storable support
    #

    def downsize(self):
{'fraction_non_alphanumeric': 0.06990478486199832, 'fraction_numerical': 0.0, 'mean_word_length': 2.6847246891651864, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '356458', 'n_tokens_mistral': 2363, 'n_tokens_neox': 2294, 'n_words': 531}