graph/test/test.py

#!/bin/env python3

import requests
from math import sin, cos
from time import sleep
import numpy as np
from argparse import ArgumentParser
# import pyaudio

def sine(args):
  x = 0.0
  while True:
    try:
      requests.get(f"http://{args.address}:{args.port}/add?chart={args.chart}&series={args.series}&x={x}&y={sin(x)}")
      x += 0.01
      sleep(0.01)
    except KeyboardInterrupt:
      break

def cosine(args):
  x = 0.0
  while True:
    try:
      requests.get(f"http://{args.address}:{args.port}/add?chart={args.chart}&series={args.series}&x={x}&y={cos(x)}")
      x += 0.01
      sleep(0.01)
    except KeyboardInterrupt:
      break

def sine_array(args):
  start = 0
  while True:
    try:
      x = np.linspace(start, start+0.5, args.num)
      y = np.sin(x)

      start += 0.5

      data = {
        "chart": args.chart,
        "series": args.series,
        "points": [ {"x": i[0], "y":i[1]} for i in zip(x,y) ]
      }

      r = requests.post(f"http://{args.address}:{args.port}/add", json=data)
      sleep(args.delay)
    except KeyboardInterrupt:
      break

def star(args):
  points = [
    (2/4,2/2),
    (3/4,0),
    (0,1/2),
    (4/4,1/2),
    (1/4,0),
    (2/4,2/2)
  ]
  for point in points:
    requests.get(f"http://{args.address}:{args.port}/add?chart={args.chart}&series={args.series}&x={point[0]}&y={point[1]}")

# def audio():
#   p = pyaudio.PyAudio()
#   stream = p.open(
#     input=True,
#     input_device_index=0,
#     format=pyaudio.paInt16,
#     channels=1,
#     rate=44100,
#     frames_per_buffer=1024
#   )
#   stream.start_stream()
#   while True:
#     try:
#       data = stream.read(1024, False)
#       requests.post(
#         "http://0.0.0.0:8080/add",
#         json={
#           "chart": 0,
#           "series": "audio",
#           "points": [ {"x": i, "y": v} for (i, v) in enumerate(data) ]
#         },
#         headers={"Connection":"close"})
#     except KeyboardInterrupt:
#       break

#   stream.stop_stream()
#   stream.close()
#   p.terminate()

if __name__ == "__main__":
  parser = ArgumentParser()
  parser.add_argument("-t", action="store", dest="test",    help="test type", type=str, choices=["sine", "cosine", "sine_array", "star"], required=False, default="sine")
  parser.add_argument("-n", action="store", dest="num",     help="sin_array size/iteration", type=int, required=False, default=100)
  parser.add_argument("-d", action="store", dest="delay",   help="delay between iteration", type=float, required=False, default=0.25)
  parser.add_argument("-c", action="store", dest="chart",   help="chart id", type=int, required=True)
  parser.add_argument("-s", action="store", dest="series",  help="series name", type=str, required=True)
  parser.add_argument("-p", action="store", dest="port",    help="server port", type=int, required=False, default=8080)
  parser.add_argument("-a", action="store", dest="address", help="server address", type=str, required=False, default="127.0.0.1")

  args = parser.parse_args()
  if args.test == "sine":
    sine(args)
  if args.test == "cosine":
    cosine(args)
  elif args.test == "sine_array":
    sine_array(args)
  elif args.test == "star":
    star(args)