mirror of
https://github.com/dragonpilot/dragonpilot.git
synced 2026-09-10 10:03:42 +08:00
Dragonpilot Team
4d7a40310f
version: dragonpilot v0.8.16 beta for EON/C2 date: 2022-08-11T09:38:43 dp-dev(priv2) master commit: c6cd233f23f60e7a6055d42850bc593c0e69082e
30 lines
652 B
Python
30 lines
652 B
Python
import time
|
|
|
|
from messaging_pyx import Context, Poller, SubSocket, PubSocket # pylint: disable=no-name-in-module, import-error
|
|
|
|
MSGS = 1e5
|
|
|
|
if __name__ == "__main__":
|
|
c = Context()
|
|
sub_sock = SubSocket()
|
|
pub_sock = PubSocket()
|
|
|
|
sub_sock.connect(c, "controlsState")
|
|
pub_sock.connect(c, "controlsState")
|
|
|
|
poller = Poller()
|
|
poller.registerSocket(sub_sock)
|
|
|
|
t = time.time()
|
|
for i in range(int(MSGS)):
|
|
bts = i.to_bytes(4, 'little')
|
|
pub_sock.send(bts)
|
|
|
|
for s in poller.poll(100):
|
|
dat = s.receive()
|
|
ii = int.from_bytes(dat, 'little')
|
|
assert(i == ii)
|
|
|
|
dt = time.time() - t
|
|
print("%.1f msg/s" % (MSGS / dt))
|