Python Tutorial (Simple Keypose)

In this page, you can find an example in Python to do basic tasks with edelkrone SDK

# -----------------------------------------------------------------------------
#
#
# -----------------------------------------------------------------------------
import requests
import datetime
import websocket
import functools
import signal
import json
import time
import sys

# -----------------------------------------------------------------------------
# Required for Windows for better print experience. Either this or remember to
# call python with -u option every time.
# -----------------------------------------------------------------------------
print = functools.partial(print, flush=True)

# -----------------------------------------------------------------------------
# Urls for websocket and http requests
# -----------------------------------------------------------------------------
httpURL = "http://127.0.0.1:32222/edelkrone"
wsURL = "ws://127.0.0.1:32222/edelkrone"

# -----------------------------------------------------------------------------
def linkDevicesQuery(debug=False):

  # ...
  url = httpURL + "/device"
  payload = {"command": "device_found_list"}
  headers = {"Content-Type": "application/json"}

  # ...
  response = requests.request("GET", url, headers=headers, json=payload, timeout=1)
  response_json = json.loads(response.text)

  # ...
  if(debug == True):
    print("")
    print("LINK DEVICES QUERY")
    print("==================\n")
    print(json.dumps(response_json, indent=2))

  # ...
  return response_json

# ---------------------------------------------------------------------------
def statusQuery(socket, debug=False):

  # ...
  socket.send(json.dumps({}))
  response = socket.recv()
  response_json = json.loads(response)

  # ...
  if(debug == True):
    print("")
    print("PERIODIC STATUS")
    print("===============\n")
    print(json.dumps(response_json, indent=2))

  # ...
  return response_json

# ---------------------------------------------------------------------------
def connectToBundle_stopMotionModule(bundleID):

  # ...
  url = httpURL + "/device"
  payload = {"command": "bundle_connect", "uniqueID": bundleID}
  headers = {"Content-Type": "application/json"}

  # ...
  response = requests.request("GET", url, headers=headers, json=payload, timeout=1)
  response_json = json.loads(response.text)
  return response_json

# ---------------------------------------------------------------------------
def keyposeStoreNumeric(bundleID, pan, tilt, slide, focus, index):

  # ...
  url = httpURL + "/device/" + bundleID
  payload = {
    "command": "keypose_store_numeric",
    "tilt": tilt,
    "pan": pan,
    "slide": slide,
    "focus": focus,
    "index": index,
    "uniqueID": 1610356720
  }
  headers = {"Content-Type": "application/json"}

  # ...
  response = requests.request("POST", url, headers=headers, json=payload, timeout=1)
  response_json = json.loads(response.text)
  return response_json

# ---------------------------------------------------------------------------
def keyposeMoveFixSpeed(
  bundleID, startIndex, aimIndex, is_loop, accelIn, accelOut, relativeSpeed
):
  # ...
  url = httpURL + "/device/" + bundleID
  payload = {
    "command": "keypose_move_fixed_speed",
    "start_index": startIndex,
    "aim_index": aimIndex,
    "is_loop": is_loop,
    "accel_in_ratio": accelIn,
    "accel_out_ratio": accelOut,
    "relative_speed": relativeSpeed,
  }
  headers = {"Content-Type": "application/json"}

  # ...
  response = requests.request("POST", url, headers=headers, json=payload, timeout=1)
  response_json = json.loads(response.text)
  return response_json

# ---------------------------------------------------------------------------
def keyposeAbortMotion(bundleID):
  # ...
  url = httpURL + "/device/" + bundleID
  payload = {"command": "keypose_abort"}
  headers = {"Content-Type": "application/json"}

  # ...
  response = requests.request("POST", url, headers=headers, json=payload, timeout=1)
  response_json = json.loads(response.text)
  return response_json

# ---------------------------------------------------------------------------
if __name__ == "__main__":

  # ...
  signal.signal(signal.SIGINT, signal.SIG_DFL)

  # List currently connected link devices
  linkDevices = linkDevicesQuery(debug=True)

  # Get the bundleID (First one in the list)
  bundleID = linkDevices["device_list"][0]["uniqueID"]

  # ...
  connectToBundle_stopMotionModule(bundleID)
  time.sleep(0.1) # Without this delay, daemon crashes

  # ...
  keyposeAbortMotion(bundleID)
  print("\nKeypose abort motion")
  time.sleep(1.0)

  # ...
  pan = 90.0
  tilt = -30.0
  slide = 0.0
  focus = 0.0
  index = 0
  keyposeStoreNumeric(bundleID, pan, tilt, slide, focus, index)
  print("\nKeypose storeNumeric slot#%d" % index)
  time.sleep(0.1)

  # ...
  pan = -30.0
  tilt = 90.0
  slide = 20.0
  focus = 0.0
  index = 1
  keyposeStoreNumeric(bundleID, pan, tilt, slide, focus, index)
  print("\nKeypose storeNumeric slot#%d" % index)
  time.sleep(0.1)

  # ...
  startIndex = 0
  aimIndex = 1
  is_loop = True
  accelIn = 0.4
  accelOut = 0.4
  relativeSpeed = 0.9
  keyposeMoveFixSpeed(
    bundleID, startIndex, aimIndex, is_loop,
    accelIn, accelOut, relativeSpeed
  )

  # Create the URL and connect to websocket
  wsURL_status = wsURL + "/device/" + bundleID + "/status"
  statusWebSocket = websocket.WebSocket()
  statusWebSocket.connect(wsURL_status)

  # ...
  start = datetime.datetime.now()
  lastStatus = datetime.datetime.now()
  while(True):

    # ...
    now = datetime.datetime.now()

    # ...
    start_delta = now - start
    lastStatus_delta = now - lastStatus

    # ...
    if(lastStatus_delta.total_seconds() > 0.015):
      lastStatus = now
      statusQuery(statusWebSocket, debug=True)

    # ...
    if(start_delta.total_seconds() > 20):
      break

  # ...
  keyposeAbortMotion(bundleID)