Function main

examples/Protobuf Example/sensor_simulator.py:165–216

()

def main() -> None:
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    period = 1.0 / RATE_HZ

    sys.stdout.write(
        f"\nProtobuf Sensor Simulator\n"
        f"  target  : udp://{HOST}:{PORT}\n"
        f"  rate    : {RATE_HZ:.0f} Hz\n"
        f"  mix     : Sensor (default) + bare Vec3 every {VEC3_EVERY_N_FRAMES} frames\n"
        f"  stop    : Ctrl+C\n\n"
    )
    sys.stdout.flush()

    t0 = time.monotonic()
    counter = 0
    sensor_count = 0
    vec3_count = 0
    next_t = t0

    try:
        while True:
            t = time.monotonic() - t0
            counter += 1

            if counter % VEC3_EVERY_N_FRAMES == 0:
                frame = make_vec3_frame(t, counter)
                kind = "Vec3"
                vec3_count += 1
            else:
                frame = make_sensor_frame(t, counter)
                kind = "Sensor"
                sensor_count += 1

            sock.sendto(frame, (HOST, PORT))

            # Throttle.
            next_t += period
            sleep_for = next_t - time.monotonic()
            if sleep_for > 0:
                time.sleep(sleep_for)
            elif sleep_for < -0.5:
                next_t = time.monotonic()  # we fell badly behind, resync

            # Status line (refreshes in place).
            sys.stdout.write(
                render_status(t, counter, sensor_count, vec3_count, kind, len(frame))
            )
            sys.stdout.flush()

    finally:
        sys.stdout.write("\n\nstopped.\n")
        sys.stdout.flush()


if __name__ == "__main__":