1 #! /usr/bin/env python3
2 # SPDX-License-Identifier: BSD-3-Clause
3 # Copyright(c) 2020 Intel Corporation
6 Script to be used with V2 Telemetry.
7 Allows the user input commands and read the Telemetry response.
17 TELEMETRY_VERSION = "v2"
21 def read_socket(sock, buf_len, echo=True):
22 """ Read data from socket and return it in JSON format """
23 reply = sock.recv(buf_len).decode()
25 ret = json.loads(reply)
26 except json.JSONDecodeError:
27 print("Error in reply: ", reply)
31 print(json.dumps(ret))
35 def handle_socket(path):
36 """ Connect to socket and handle user input """
37 sock = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET)
39 print("Connecting to " + path)
43 print("Error connecting to " + path)
46 json_reply = read_socket(sock, 1024)
47 output_buf_len = json_reply["max_output_len"]
49 # get list of commands for readline completion
50 sock.send("/".encode())
51 CMDS = read_socket(sock, output_buf_len, False)["/"]
54 text = input('--> ').strip()
56 if text.startswith('/'):
57 sock.send(text.encode())
58 read_socket(sock, output_buf_len)
59 text = input('--> ').strip()
63 def readline_complete(text, state):
64 """ Find any matching commands from the list based on user input """
65 all_cmds = ['quit'] + CMDS
67 matches = [c for c in all_cmds if c.startswith(text)]
73 readline.parse_and_bind('tab: complete')
74 readline.set_completer(readline_complete)
75 readline.set_completer_delims(readline.get_completer_delims().replace('/', ''))
77 # Path to sockets for processes run as a root user
78 for f in glob.glob('/var/run/dpdk/*/dpdk_telemetry.%s' % TELEMETRY_VERSION):
80 # Path to sockets for processes run as a regular user
81 for f in glob.glob('%s/dpdk/*/dpdk_telemetry.%s' %
82 (os.environ.get('XDG_RUNTIME_DIR', '/tmp'), TELEMETRY_VERSION)):