sinqepicsapp/utils/deltatau.py

#!/usr/bin/env python3
"""
  This is some code for communicating with a Delta-Tau PMAC motor 
  controller as used at SINQ from python. Python 3 is assumed and 
  tested. We use the ethernet protocol of the Pmac. This code benefits 
  from the SICS driver for Pmac which in turn benefits from the driver 
  written by Pete Leicester at Diamond.


  Usage: python36 deltatau.py  pmachost:port command
         Then returns the response for command. 

  Mark Koennecke, October 2019
"""

import struct
import socket
import curses

def packPmacCommand(command):
    # 0x40 = VR_DOWNLOAD 
    # 0xBF = VR_PMAC_GETRESPONSE
    buf = struct.pack('BBHHH',0x40,0xBF,0,0,socket.htons(len(command)))
    buf = buf + bytes(command,'utf-8')
    return buf

def readPmacReply(input):
    msg = bytearray()
    expectAck = True
    while True:
        b = input.recv(1)
        bint = int.from_bytes(b,byteorder='little')
        if bint == 2 or bint == 7: #STX or BELL
            expectAck = False
            continue
        if expectAck and bint == 6: # ACK
            return bytes(msg)
        else:
            if bint == 13 and not expectAck: # CR
                return bytes(msg)
            else:
                msg.append(bint)

if __name__ == "__main__":
    from sys import argv

    try:

        addr = argv[1].split(':')
        s = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
        s.connect((addr[0],int(addr[1])))

        if len(argv) == 3:
            buf = packPmacCommand(argv[2])
            s.send(buf)
            reply = readPmacReply(s)
            print(reply.decode('utf-8') + '\n')

        else:

            try:

                stdscr = curses.initscr()
                curses.noecho()
                curses.cbreak()
                stdscr.keypad(True)
                stdscr.scrollok(True)

                stdscr.addstr(">> ")
                stdscr.refresh()

                history = [""]
                ptr = len(history) - 1

                while True:
                    c = stdscr.getch()
                    if c == curses.KEY_RIGHT:
                        (y, x) = stdscr.getyx()
                        if x < len(history[ptr]) + 3:
                            stdscr.move(y, x+1)
                            stdscr.refresh()
                    elif c == curses.KEY_LEFT:
                        (y, x) = stdscr.getyx()
                        if x > 3:
                            stdscr.move(y, x-1)
                            stdscr.refresh()
                    elif c == curses.KEY_UP:
                        if ptr > 0:
                            ptr -= 1
                            stdscr.addch("\r")
                            stdscr.clrtoeol() 
                            stdscr.addstr(">> " + history[ptr])
                    elif c == curses.KEY_DOWN:
                        if ptr < len(history) - 1:
                            ptr += 1
                            stdscr.addch("\r")
                            stdscr.clrtoeol() 
                            stdscr.addstr(">> " + history[ptr])
                    elif c == curses.KEY_ENTER or c == ord('\n') or c == ord('\r'):
                        if history[ptr] == 'quit':
                            break

                        # because of arrow keys move back to the end of the line
                        (y, x) = stdscr.getyx()
                        stdscr.move(y, 3+len(history[ptr]))

                        if history[ptr]:
                            buf = packPmacCommand(history[ptr])
                            s.send(buf)
                            reply = readPmacReply(s)
                            stdscr.addstr("\n" + reply.decode('utf-8')[0:-1])

                        if ptr == len(history) - 1 and history[ptr] != "":
                            history += [""]
                        else:
                            history[-1] = ""
                        ptr = len(history) - 1

                        stdscr.addstr("\n>> ")
                        stdscr.refresh()

                    else:
                        if ptr < len(history) - 1: # Modifying previous input
                            if len(history[-1]) == 0:
                                history[-1] = history[ptr]
                                ptr = len(history) - 1

                            else:
                                history += [history[ptr]]
                                ptr = len(history) - 1

                        if c == curses.KEY_BACKSPACE:
                            if len(history[ptr]) == 0:
                                continue
                            (y, x) = stdscr.getyx()
                            history[ptr] = history[ptr][0:x-4] + history[ptr][x-3:]
                            stdscr.addch("\r")
                            stdscr.clrtoeol() 
                            stdscr.addstr(">> " + history[ptr])
                            stdscr.move(y, x-1)
                            stdscr.refresh()

                        else:
                            (y, x) = stdscr.getyx()
                            history[ptr] = history[ptr][0:x-3] + chr(c) + history[ptr][x-3:]
                            stdscr.addch("\r")
                            stdscr.clrtoeol() 
                            stdscr.addstr(">> " + history[ptr])
                            stdscr.move(y, x+1)
                            stdscr.refresh()

            finally:

                # to quit
                curses.nocbreak()
                stdscr.keypad(False)
                curses.echo()
                curses.endwin()

    except:
        print("""
        Invalid Arguments

        Option 1: Single Command
        ------------------------

        Usage: deltatau.py pmachost:port command
        This then returns the response for command. 

        Option 2: CLI Mode
        ------------------

        Usage: deltatau.py pmachost:port

        You can then type in a command, hit enter, and the response will see
        the reponse, before being prompted to again enter a command. Type
        'quit' to close prompt.
        """)