Cloud Binary Passthrough

topic = f"haltian-iot/events/{subscription_id}/{key_id}/payloads/wirepas/+"
client.subscribe(topic)

{
  "received_at": "2026-01-29T12:34:56.789Z",
  "value": {
    "mqtt_topic": "haltian-iot/wirepas/gw-event/received_data/BS225000789/sink1/7156763/21/21",
    "payload": "CkpCTgofCgtCUzIyNTAwMDc4ORIFc2luazEY65DNj9fX+JPTARj..."
  }
}

{
  "tsmId": 12345678,
  "tsmEv": 10,
  "tsmTs": 1706528096,
  "temp": 23.45,
  "humd": 45.2,
  "batl": 95
}

import base64
import json

def extract_payload(mqtt_message: bytes) -> tuple[bytes, str, str]:
    """
    Extract binary payload and endpoints from MQTT JSON message.
    
    Returns:
        Tuple of (decoded_payload_bytes, source_endpoint, dest_endpoint)
    """
    data = json.loads(mqtt_message.decode("utf-8"))
    value = data.get("value", {})
    
    # Decode base64 payload
    payload_b64 = value.get("payload", "")
    payload_bytes = base64.b64decode(payload_b64) if payload_b64 else b""
    
    # Extract endpoints from origin topic
    origin_topic = value.get("mqtt_topic", "")
    parts = origin_topic.split("/")
    source_ep = parts[-2] if len(parts) >= 2 else "unknown"
    dest_ep = parts[-1] if len(parts) >= 1 else "unknown"
    
    return payload_bytes, source_ep, dest_ep

message WirepasMessage {
    uint32 source_address = 1;      // Sensor node address (unique per device)
    uint32 destination_address = 2;  // Target address (usually gateway)
    uint32 source_endpoint = 3;      // Identifies data type/protocol
    uint32 destination_endpoint = 4; // Usually matches source_endpoint
    uint64 travel_time_ms = 5;       // Network transit time
    uint32 qos = 6;                  // Quality of service level
    bytes  payload_data = 7;         // The actual sensor data (inner payload)
    uint64 rx_time_ms_epoch = 8;     // Receive timestamp (milliseconds since epoch)
}

import struct
from dataclasses import dataclass
from typing import Optional

@dataclass
class WirepasMessage:
    source_address: int
    destination_address: int
    source_endpoint: int
    destination_endpoint: int
    travel_time_ms: int
    qos: int
    payload_data: bytes
    rx_time_ms_epoch: int

def decode_wirepas_message(raw_data: bytes) -> WirepasMessage:
    """
    Decode Wirepas protobuf message.
    
    Protobuf wire format:
    - Field number and wire type packed in varint
    - Wire type 0 = varint, type 2 = length-delimited (bytes)
    """
    pos = 0
    fields = {}
    
    while pos < len(raw_data):
        # Read field tag (varint)
        tag, pos = _read_varint(raw_data, pos)
        field_number = tag >> 3
        wire_type = tag & 0x07
        
        if wire_type == 0:  # Varint
            value, pos = _read_varint(raw_data, pos)
        elif wire_type == 2:  # Length-delimited
            length, pos = _read_varint(raw_data, pos)
            value = raw_data[pos:pos + length]
            pos += length
        else:
            raise ValueError(f"Unsupported wire type: {wire_type}")
        
        fields[field_number] = value
    
    return WirepasMessage(
        source_address=fields.get(1, 0),
        destination_address=fields.get(2, 0),
        source_endpoint=fields.get(3, 0),
        destination_endpoint=fields.get(4, 0),
        travel_time_ms=fields.get(5, 0),
        qos=fields.get(6, 0),
        payload_data=fields.get(7, b''),
        rx_time_ms_epoch=fields.get(8, 0),
    )

def _read_varint(data: bytes, pos: int) -> tuple[int, int]:
    """Read a protobuf varint from data at position pos."""
    result = 0
    shift = 0
    while True:
        byte = data[pos]
        pos += 1
        result |= (byte & 0x7F) << shift
        if not (byte & 0x80):
            break
        shift += 7
    return result, pos