Data Provider API

from wfa.measurement.system.v1alpha import requisitions_pb2

# The Kingdom makes requisitions available to data providers
# Data providers must poll or stream for new requisitions

def poll_for_requisitions(data_provider_id):
    # Query requisitions for this data provider
    # Implementation varies by deployment
    requisitions = get_unfulfilled_requisitions(data_provider_id)
    
    for requisition in requisitions:
        if requisition.state == requisitions_pb2.Requisition.UNFULFILLED:
            print(f"New requisition: {requisition.name}")
            fulfill_requisition(requisition)

import hashlib
from typing import List
from wfa.measurement.api.v2alpha import event_group_pb2

class SketchGenerator:
    def __init__(self, decay_rate: float, max_size: int):
        self.decay_rate = decay_rate
        self.max_size = max_size
    
    def generate_sketch(self, events: List[Event]) -> bytes:
        """
        Generate Liquid Legions sketch from event data.
        
        Args:
            events: List of user events with VIDs and metadata
            
        Returns:
            Serialized sketch bytes
        """
        sketch_registers = {}
        
        for event in events:
            # Hash VID to register index
            vid_hash = hashlib.sha256(event.vid.encode()).digest()
            register_index = int.from_bytes(vid_hash[:8], 'big') % self.max_size
            
            # Apply decay sampling
            if random.random() < self.decay_rate:
                if register_index not in sketch_registers:
                    sketch_registers[register_index] = {
                        'key': vid_hash,
                        'count': 0
                    }
                sketch_registers[register_index]['count'] += 1
        
        # Serialize sketch
        return self._serialize_sketch(sketch_registers)
    
    def _serialize_sketch(self, registers) -> bytes:
        # Protocol-specific serialization
        # Returns encrypted sketch bytes
        pass

from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.hkdf import HKDF

def encrypt_sketch_for_duchy(sketch_data: bytes, duchy_public_key: ec.EllipticCurvePublicKey) -> bytes:
    """
    Encrypt sketch data for a duchy using ECIES.
    
    Args:
        sketch_data: Raw sketch bytes
        duchy_public_key: Duchy's public key
        
    Returns:
        Encrypted sketch
    """
    # Generate ephemeral key pair
    ephemeral_private_key = ec.generate_private_key(ec.SECP256R1())
    ephemeral_public_key = ephemeral_private_key.public_key()
    
    # Perform ECDH
    shared_secret = ephemeral_private_key.exchange(
        ec.ECDH(), duchy_public_key
    )
    
    # Derive symmetric key
    derived_key = HKDF(
        algorithm=hashes.SHA256(),
        length=32,
        salt=None,
        info=b'sketch-encryption'
    ).derive(shared_secret)
    
    # Encrypt sketch data with AES-GCM
    encrypted_data = aes_gcm_encrypt(sketch_data, derived_key)
    
    # Package: ephemeral_public_key || encrypted_data
    return serialize_encrypted_sketch(ephemeral_public_key, encrypted_data)

from wfa.measurement.system.v1alpha import requisitions_service_pb2

def fulfill_requisition(requisition, sketch_data, nonce):
    # Upload sketch to storage
    sketch_uri = upload_to_blob_storage(sketch_data)
    
    # Mark requisition as fulfilled
    request = requisitions_service_pb2.FulfillRequisitionRequest(
        name=requisition.name,
        nonce=nonce,
        fulfillment_context=requisitions_service_pb2.Requisition.FulfillmentContext(
            build_label="edp-simulator-v1.2.3",
            warnings=[
                "Some events excluded due to data quality issues"
            ]
        )
    )
    
    response = requisitions_client.FulfillRequisition(request)
    print(f"Requisition fulfilled: {response.state}")
    return response

from wfa.measurement.api.v2alpha import event_groups_pb2
from google.type import interval_pb2

request = event_groups_pb2.CreateEventGroupRequest(
    parent="dataProviders/456",
    event_group=event_groups_pb2.EventGroup(
        event_group_reference_id="campaign-2024-Q1",
        event_templates=[
            event_groups_pb2.EventGroup.EventTemplate(
                type="wfa.measurement.api.v2alpha.event_templates.BannerAd"
            )
        ],
        media_types=[event_groups_pb2.BANNER_AD],
        data_availability_interval=interval_pb2.Interval(
            start_time=timestamp_pb2.Timestamp(seconds=start_seconds),
            end_time=timestamp_pb2.Timestamp(seconds=end_seconds)
        ),
        event_group_metadata=event_groups_pb2.EventGroup.EventGroupMetadata(
            ad_metadata=event_groups_pb2.EventGroup.EventGroupMetadata.AdMetadata(
                campaign_metadata=event_groups_pb2.EventGroup.EventGroupMetadata.AdMetadata.CampaignMetadata(
                    brand_name="Acme Corp",
                    campaign_name="Spring 2024 Promo"
                )
            )
        )
    ),
    event_group_id="eg-123456"
)

event_group = event_groups_client.CreateEventGroup(request)

# To refuse a requisition, data providers typically use internal APIs
# The requisition state transitions to REFUSED
# This causes the parent measurement to enter FAILED state

def refuse_requisition(requisition_name, reason):
    # Implementation depends on deployment
    # Mark requisition as refused with reason
    mark_requisition_refused(requisition_name, reason)