Skip to content

main.py

DEV5559_simulator()

Simulates a DEV5559 uplink.

Source code in src/main.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def DEV5559_simulator():
    """Simulates a DEV5559 uplink."""

    import random

    payload = {
        "battery_soc": random.randint(0, 100),
        "battery_volts": round(random.uniform(2.5, 4.2), 3),
        "humidity": round(random.uniform(0, 100), 1),
        "moisture": round(random.uniform(0, 100), 1),
        "temperature": round(random.uniform(0, 100), 1),
        "lat": round(random.uniform(-90, 90), 5),
        "lng": round(random.uniform(-180, 180), 5),
        "fw": random.choice([0, 1, 2])
    }

    return payload

decode_payload(encoded_payload, device_name, message_type)

Decodes a base64 encoded protobuf message into a dictionary.

Parameters:

Name Type Description Default
encoded_payload str

A base64 encoded protobuf message.

 required
device_name str

The name of the device.

 required
message_type str

The type of message.

 required

Returns:

Name Type Description
message dict

A dictionary representation of the protobuf message.
Source code in src/main.py 
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def decode_payload(encoded_payload: str, device_name: str, message_type: str) -> dict:
    """Decodes a base64 encoded protobuf message into a dictionary.

    Args:
        encoded_payload (str): A base64 encoded protobuf message.
        device_name (str): The name of the device.
        message_type (str): The type of message.

    Returns:
        message (dict): A dictionary representation of the protobuf message.
    """

    try:
        # Dynamically retrieve the message class based on the device name and message type
        message_class = getattr(globals()[device_name], message_type)
    except AttributeError:
        print(f"Error: Invalid device name or message type: {device_name}.{message_type}")
        return

    # Decode the base64 encoded message
    decoded_message = base64.b64decode(encoded_payload)

    # Create an instance of the message class and parse the decoded message
    message = message_class()
    message.ParseFromString(decoded_message)

    # Convert the message to a dictionary
    decoded = json_format.MessageToDict(
        message, 
        preserving_proto_field_name=True)

    logger.debug(f"{device_name}.{message_class.__name__} decoded. Result: {decoded}, Bytes: {len(json.dumps(decoded))}")


    return decoded

encode_payload(payload, device_name, message_type)

Converts a Python dictionary to a compact Base64 encoded protobuf message.

Parameters:

Name Type Description Default
payload dict

A dictionary containing the payload data.

 required
device_name str

The name of the device model, e.g. "DEV5559".

 required
message_type str

The type of message, e.g. "Uplink" or "Downlink".

 required

Returns:

Name Type Description
encoded str

A base64 encoded protobuf message.
Source code in src/main.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def encode_payload(payload: dict, device_name: str, message_type: str) -> str:
    """Converts a Python dictionary to a compact Base64 encoded protobuf message.

    Args:
        payload (dict): A dictionary containing the payload data.
        device_name (str): The name of the device model, e.g. "DEV5559".
        message_type (str): The type of message, e.g. "Uplink" or "Downlink".

    Returns:
        encoded (str): A base64 encoded protobuf message.
    """

    message_class = getattr(globals()[device_name], message_type)

    device = message_class()

    json_format.ParseDict(
        payload,
        device,
        ignore_unknown_fields=False)

    bytes = device.SerializeToString()

    encoded = base64.b64encode(bytes).decode("utf-8")

    logger.debug(f"{device_name}.{message_class.__name__} encoded. Result: {encoded}, Bytes: {len(encoded)}")

    return encoded