Wanlop Surakampontorn

import numpy as np
import mmh3

class MinHashCustom:
    """
    DTO for an actual MinHash
    <minhash>: a binary sequence of packed int8/32 values
    <minhash_int>: the equivalent representation of <minhash> but as list of int8/32
    """

    _HASH_MAX = 0xFFFFFFFF
    _MINHASH_BITS = 32

    def getSignatureEntrySize(self):
        return 1 if self.MINHASH_BITS <= 8 else 4

    def __init__(self, function_addr=None, minhash_bytes=None, minhash_signature=None, minhash_bits=32):
        self.minhash = b""
        self.minhash_int = []
        if minhash_bits:
            self._MINHASH_BITS = minhash_bits
        if minhash_bytes and minhash_signature:
            raise ValueError("Can use only one keyword argument")
        if minhash_bytes:
            if self._MINHASH_BITS <= 8:
                minhash_signature = np.frombuffer(minhash_bytes, dtype=np.uint8)
            else:
                minhash_signature = np.frombuffer(minhash_bytes, dtype=np.uint32)
            self.setMinHash(minhash_signature)
        elif minhash_signature:
            self.setMinHash(minhash_signature)

        self.shingler_composition = {}
        self.function_addr = function_addr

    def hasMinHash(self):
        return len(self.minhash) > 0

    def getMinHash(self):
        return self.minhash

    def getMinHashInt(self):
        return self.minhash_int

    def setMinHash(self, minhash_signature):
        self.minhash_int = [i % 2 ** self._MINHASH_BITS for i in minhash_signature]
        if self._MINHASH_BITS <= 8:
            self.minhash = np.array(self.minhash_int, dtype=np.uint8).tobytes()
        else:
            self.minhash = np.array(self.minhash_int, dtype=np.uint32).tobytes()

    def getComposition(self):
        return self.shingler_composition

    def scoreAgainst(self, other):
        return self.calculateMinHashScore(self.minhash, other.minhash, minhash_bits=self._MINHASH_BITS)

    @staticmethod
    def getHashMax():
        return MinHashCustom._HASH_MAX

    @staticmethod
    def hashData(data, seed) -> int:
        if isinstance(data, (str, bytes, bytearray)):
            return mmh3.hash(data, seed) & MinHashCustom._HASH_MAX
        elif isinstance(data, (list, tuple)):
            to_hash = "|".join(str(elem) for elem in data)
            return mmh3.hash(to_hash, seed) & MinHashCustom._HASH_MAX
        else:
            raise NotImplementedError(
                f"Type not supported for hashData: {type(data).__name__}"
            )

    @staticmethod
    def calculateMinHashScore(first, second, minhash_bits=32):
        if minhash_bits <= 8:
            first_np = np.frombuffer(first, dtype=np.uint8)
            second_np = np.frombuffer(second, dtype=np.uint8)
        else:
            first_np = np.frombuffer(first, dtype=np.uint32)
            second_np = np.frombuffer(second, dtype=np.uint32)
        return 100.0 * sum(first_np == second_np) / len(first_np)

    @staticmethod
    def calculateMinHashIntScore(first, second):
        score = 0
        num_hashes = len(first)
        if num_hashes:
            for index, part in enumerate(first):
                score += 1 if part == second[index] else 0
            return 100.0 * score / num_hashes
        return 0.0

    @property
    def MINHASH_BITS(self):
        return self._MINHASH_BITS