Bloom Filters

class BloomFilter:
    """
    Bloom filter that allows for basic membership tests.
    Only integers are supported as keys.
    """
    def __init__(self, size: int = 1024, hash_funcs: int = 23, seed: int = 0):
        """
        Create a new bloom filter.
        
        Parameters:
        - size: Size of bit array (must be > 0)
        - hash_funcs: Number of hash functions (must be > 0)
        - seed: Seed for hash functions (default 0)
        
        Raises:
        - AssertionError: If size or hash_funcs <= 0
        """
        
    @classmethod
    def from_error_rate(cls, members: int, error_rate: float = 1e-4):
        """
        Create bloom filter optimized for specific member count and error rate.
        
        Parameters:
        - members: Expected number of members to add
        - error_rate: Desired false positive rate (default 1e-4)
        
        Returns:
        - BloomFilter instance with optimal parameters
        """
        
    def add(self, item: int) -> None:
        """
        Add item to bloom filter.
        
        Parameters:
        - item: Integer item to add
        """
        
    def __contains__(self, item: int) -> bool:
        """
        Test membership in bloom filter.
        
        Parameters:
        - item: Integer item to test
        
        Returns:
        - True if item might be in set (possible false positive)
        - False if item is definitely not in set (no false negatives)
        """

def to_bytes(self) -> bytes:
    """
    Serialize bloom filter to bytes.
    
    Returns:
    - Byte representation of bloom filter
    """
    
def from_bytes(self, byte_string: bytes):
    """
    Deserialize bloom filter from bytes.
    
    Parameters:
    - byte_string: Bytes containing serialized bloom filter
    
    Returns:
    - Self (for method chaining)
    
    Raises:
    - ValueError: If serialization version is unknown or incompatible
    """

def calculate_size_and_hash_count(members: int, error_rate: float) -> tuple[int, int]:
    """
    Calculate optimal size and hash function count for bloom filter.
    
    Parameters:
    - members: Expected number of members
    - error_rate: Desired false positive rate
    
    Returns:
    - Tuple of (bit_count, hash_count) for optimal performance
    """

from preshed.bloom import BloomFilter

# Create bloom filter
bloom = BloomFilter(size=1024, hash_funcs=3)

# Add items
bloom.add(12345)
bloom.add(54321)
bloom.add(98765)

# Test membership
print(12345 in bloom)  # True (definitely added)
print(54321 in bloom)  # True (definitely added)
print(99999 in bloom)  # False (probably not added, no false negatives)
print(11111 in bloom)  # False or True (possible false positive)

from preshed.bloom import BloomFilter

# Create bloom filter optimized for 1000 members with 0.01% error rate
bloom = BloomFilter.from_error_rate(members=1000, error_rate=0.0001)

# Add expected number of items
for i in range(1000):
    bloom.add(i)

# Test membership
print(500 in bloom)  # True (added)
print(1500 in bloom)  # False or True (very low probability of false positive)

from preshed.bloom import BloomFilter, calculate_size_and_hash_count

# Calculate optimal parameters
members = 5000
error_rate = 0.001
size, hash_funcs = calculate_size_and_hash_count(members, error_rate)

print(f"Optimal size: {size} bits")
print(f"Optimal hash functions: {hash_funcs}")

# Create bloom filter with calculated parameters
bloom = BloomFilter(size=size, hash_funcs=hash_funcs)

from preshed.bloom import BloomFilter

# Create and populate bloom filter
bloom1 = BloomFilter(size=1024, hash_funcs=3)
for i in range(100):
    bloom1.add(i)

# Serialize to bytes
data = bloom1.to_bytes()
print(f"Serialized size: {len(data)} bytes")

# Create new bloom filter and deserialize
bloom2 = BloomFilter()
bloom2.from_bytes(data)

# Verify they work identically
for i in range(100):
    assert (i in bloom1) == (i in bloom2)

print("Serialization successful!")

from preshed.bloom import BloomFilter
import random

# Create large bloom filter for millions of items
bloom = BloomFilter.from_error_rate(members=1_000_000, error_rate=0.0001)

# Add random items
added_items = set()
for _ in range(50000):
    item = random.randint(1, 10_000_000)
    bloom.add(item)
    added_items.add(item)

# Test known items (should all be True)
true_positives = 0
for item in list(added_items)[:1000]:
    if item in bloom:
        true_positives += 1

print(f"True positives: {true_positives}/1000")  # Should be 1000

# Test random items (some false positives expected)
false_positives = 0
for _ in range(10000):
    item = random.randint(10_000_001, 20_000_000)  # Definitely not added
    if item in bloom:
        false_positives += 1

print(f"False positives: {false_positives}/10000")  # Should be ~1 (0.01%)

from preshed.bloom import BloomFilter

def is_prime(n):
    """Simple prime check for demonstration."""
    if n < 2:
        return False
    for i in range(2, int(n**0.5) + 1):
        if n % i == 0:
            return False
    return True

# Store known primes in bloom filter for fast lookup
prime_bloom = BloomFilter.from_error_rate(members=10000, error_rate=0.0001)

# Add first 10000 primes
count = 0
num = 2
while count < 10000:
    if is_prime(num):
        prime_bloom.add(num)
        count += 1
    num += 1

# Fast membership testing
def might_be_prime(n):
    """Quick filter for prime candidates."""
    if n in prime_bloom:
        return True  # Might be prime (check further)
    else:
        return False  # Definitely not prime (if n < threshold)

# Test some numbers
test_numbers = [17, 97, 997, 9973, 10007, 10009]
for num in test_numbers:
    if might_be_prime(num):
        print(f"{num}: Possible prime (check with full algorithm)")
    else:
        print(f"{num}: Definitely not prime")

from preshed.bloom import BloomFilter

# Use bloom filter for memory-efficient duplicate detection
seen_bloom = BloomFilter.from_error_rate(members=100000, error_rate=0.001)
definite_duplicates = set()

def process_item(item_id):
    """Process item, tracking potential duplicates."""
    if item_id in seen_bloom:
        # Might be duplicate - need further verification
        if item_id in definite_duplicates:
            print(f"Definite duplicate: {item_id}")
            return False
        else:
            # First potential duplicate
            definite_duplicates.add(item_id)
            print(f"Potential duplicate (false positive?): {item_id}")
    else:
        # Definitely not seen before
        seen_bloom.add(item_id)
        print(f"New item: {item_id}")
    
    return True

# Process some items
items = [1, 2, 3, 1, 4, 2, 5, 6, 1]
for item in items:
    process_item(item)