This repository has been archived by the owner on Sep 1, 2023. It is now read-only.
forked from Tanganelli/CoAPthon
-
Notifications
You must be signed in to change notification settings - Fork 0
/
coapping.py
107 lines (84 loc) · 2.88 KB
/
coapping.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
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
68
69
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
105
106
107
#!/usr/bin/env python
# COAP ping implementation
# 0x4000 0001 <--> 0x7000 0001
# 0x4000 0002 <--> 0x7000 0002
# 0x4000 0003 <--> 0x7000 0003
import socket
import struct
import sys
from time import sleep, time
from optparse import OptionParser
# Parse Options
if __name__ == '__main__':
parser = OptionParser()
parser.add_option("-n", "--hostname",
action="append",
dest="host_name",
help="Define COAP host name")
parser.add_option("-p", "--port",
action="append",
dest="host_port",
default=5683,
help="Define COAP host port (default: 5683)")
parser.add_option("-l", "--loops",
type="int",
dest="ping_loops",
default=0,
help="Number of ping loops (default: 0 - forever)")
parser.add_option("-t", "--sleep",
type="float",
dest="sleep_sec",
default=1,
help="Time in seconds between two pings (default: 1 sec)")
(options, args) = parser.parse_args()
# COAP ping parameters setup
host = options.host_name[0]
port = options.host_port
sleep_sec = options.sleep_sec
ping_loops = options.ping_loops
ping_no = 1 # ping payload counter
ping_cnt = 0 # global ping cnt
print 'COAP ping script'
print 'COAP ping to: %s:%s...' % (host, port)
try:
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
except socket.error:
print 'Error: Failed to create socket'
sys.exit()
while(1):
loop_time = time()
msg = '' #[0x40, 0x00, 0x00, 0x00]
msg += struct.pack("B", 0x40)
msg += struct.pack("B", 0x00)
msg += struct.pack("B", 0x00)
msg += struct.pack("B", ping_no)
try :
print '[0x%08X] Send ping:' % (ping_cnt), [hex(ord(c)) for c in msg]
#Set the whole string
s.sendto(msg, (host, port))
s.settimeout(2 + sleep_sec)
# receive data from client (data, addr)
d = s.recvfrom(4)
reply = d[0]
addr = d[1]
# We need to check if ping peyload counter is the same in reply
status = bytes(msg)[3] == bytes(reply)[3]
print '[0x%08X] Recv ping:' % (ping_cnt), [hex(ord(c)) for c in reply], 'ok' if status else 'fail'
except socket.error as e:
print 'Error: socket.error: ', str(e)
#print 'Error Code : ' + str(msg[0]) + ' Message ' + msg[1]
sleep(3) # Waiting to recover ;)
except socket.timeout:
print("Error: closing socket")
s.close()
if ping_no >= 0xFF:
ping_no = 1
else:
ping_no += 1
sleep(sleep_sec - (time() - loop_time))
print 'In %.2f sec' % (time() - loop_time)
if ping_loops > 0:
if ping_loops == 1:
break
ping_loops -= 1
ping_cnt += 1