63 lines
1.6 KiB
Python
63 lines
1.6 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""
|
||
|
|
Test KrautStatus's UDP API for a given IP and (optionally) port.
|
||
|
|
"""
|
||
|
|
|
||
|
|
import argparse
|
||
|
|
import ipaddress
|
||
|
|
import socket
|
||
|
|
|
||
|
|
def port(string):
|
||
|
|
"Convert string to an unsigend integer that is a valid port number"
|
||
|
|
port = int(string)
|
||
|
|
|
||
|
|
if port < 0 or port > 65535:
|
||
|
|
raise ValueError()
|
||
|
|
|
||
|
|
return port
|
||
|
|
|
||
|
|
def main():
|
||
|
|
parser = argparse.ArgumentParser(__doc__)
|
||
|
|
|
||
|
|
parser.add_argument("ip",
|
||
|
|
metavar="IP",
|
||
|
|
type=ipaddress.ip_address,
|
||
|
|
help="IP address of the KrautStatus Arduino")
|
||
|
|
|
||
|
|
parser.add_argument("port",
|
||
|
|
nargs="?",
|
||
|
|
metavar="PORT",
|
||
|
|
default=12345,
|
||
|
|
type=port,
|
||
|
|
help="port that the KrautStatus Arduino listens to")
|
||
|
|
|
||
|
|
args = parser.parse_args()
|
||
|
|
|
||
|
|
|
||
|
|
receiver = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||
|
|
receiver.bind(('0.0.0.0', args.port))
|
||
|
|
print(f'Listening on 0.0.0.0:{args.port}.')
|
||
|
|
|
||
|
|
sender = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||
|
|
# We do not use \0 or \1 here, so that we do not trigger ourselves when
|
||
|
|
# testing the script locally (127.0.0.1)
|
||
|
|
sender.sendto(b'\2', (str(args.ip), args.port))
|
||
|
|
print(f'Sent null byte to {args.ip}:{args.port}.')
|
||
|
|
|
||
|
|
while True:
|
||
|
|
status, address = receiver.recvfrom(1)
|
||
|
|
|
||
|
|
if not address[0] == str(args.ip):
|
||
|
|
continue
|
||
|
|
|
||
|
|
if status[0] == 0:
|
||
|
|
print('Responded: door closed')
|
||
|
|
return
|
||
|
|
|
||
|
|
if status[0] == 1:
|
||
|
|
print('Resondend: door open')
|
||
|
|
return
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
main()
|