tests/rpc/frontend.py

   1 def modules():
   2         return ["python_inv"]
   3
   4 def derive(module, parameters):
   5         assert module == r"python_inv"
   6         if parameters.keys() != {r"\width"}:
   7                 raise ValueError("Invalid parameters")
   8         return "ilang", r"""
   9 module \impl
  10         wire width {width:d} input 1 \i
  11         wire width {width:d} output 2 \o
  12         cell $neg $0
  13                 parameter \A_SIGNED 1'0
  14                 parameter \A_WIDTH 32'{width:b}
  15                 parameter \Y_WIDTH 32'{width:b}
  16                 connect \A \i
  17                 connect \Y \o
  18         end
  19 end
  20 module \python_inv
  21         wire width {width:d} input 1 \i
  22         wire width {width:d} output 2 \o
  23         cell \impl $0
  24                 connect \i \i
  25                 connect \o \o
  26         end
  27 end
  28 """.format(width=parameters[r"\width"])
  29
  30 # ----------------------------------------------------------------------------
  31
  32 import json
  33 import argparse
  34 import sys, socket, os
  35 try:
  36         import msvcrt, win32pipe, win32file
  37 except ImportError:
  38         msvcrt = win32pipe = win32file = None
  39
  40 def map_parameter(parameter):
  41         if parameter["type"] == "unsigned":
  42                 return int(parameter["value"], 2)
  43         if parameter["type"] == "signed":
  44                 width = len(parameter["value"])
  45                 value = int(parameter["value"], 2)
  46                 if value & (1 << (width - 1)):
  47                         value = -((1 << width) - value)
  48                 return value
  49         if parameter["type"] == "string":
  50                 return parameter["value"]
  51         if parameter["type"] == "real":
  52                 return float(parameter["value"])
  53
  54 def call(input_json):
  55         input = json.loads(input_json)
  56         if input["method"] == "modules":
  57                 return json.dumps({"modules": modules()})
  58         if input["method"] == "derive":
  59                 try:
  60                         frontend, source = derive(input["module"],
  61                                 {name: map_parameter(value) for name, value in input["parameters"].items()})
  62                         return json.dumps({"frontend": frontend, "source": source})
  63                 except ValueError as e:
  64                         return json.dumps({"error": str(e)})
  65
  66 def main():
  67         parser = argparse.ArgumentParser()
  68         modes = parser.add_subparsers(dest="mode")
  69         mode_stdio = modes.add_parser("stdio")
  70         if os.name == "posix":
  71                 mode_path = modes.add_parser("unix-socket")
  72         if os.name == "nt":
  73                 mode_path = modes.add_parser("named-pipe")
  74         mode_path.add_argument("path")
  75         args = parser.parse_args()
  76
  77         if args.mode == "stdio":
  78                 while True:
  79                         input = sys.stdin.readline()
  80                         if not input: break
  81                         sys.stdout.write(call(input) + "\n")
  82                         sys.stdout.flush()
  83
  84         if args.mode == "unix-socket":
  85                 sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
  86                 sock.bind(args.path)
  87                 try:
  88                         sock.listen(1)
  89                         conn, addr = sock.accept()
  90                         file = conn.makefile("rw")
  91                         while True:
  92                                 input = file.readline()
  93                                 if not input: break
  94                                 file.write(call(input) + "\n")
  95                                 file.flush()
  96                 finally:
  97                         sock.close()
  98                         os.unlink(args.path)
  99
 100         if args.mode == "named-pipe":
 101                 pipe = win32pipe.CreateNamedPipe(args.path, win32pipe.PIPE_ACCESS_DUPLEX,
 102                     win32pipe.PIPE_TYPE_BYTE|win32pipe.PIPE_READMODE_BYTE|win32pipe.PIPE_WAIT,
 103                     1, 4096, 4096, 0, None)
 104                 win32pipe.ConnectNamedPipe(pipe, None)
 105                 try:
 106                         while True:
 107                                 input = b""
 108                                 while not input.endswith(b"\n"):
 109                                         result, data = win32file.ReadFile(pipe, 4096)
 110                                         assert result == 0
 111                                         input += data
 112                                         assert not b"\n" in input or input.endswith(b"\n")
 113                                 output = (call(input.decode("utf-8")) + "\n").encode("utf-8")
 114                                 length = len(output)
 115                                 while length > 0:
 116                                         result, done = win32file.WriteFile(pipe, output)
 117                                         assert result == 0
 118                                         length -= done
 119                 except win32file.error as e:
 120                         if e.args[0] == 109: # ERROR_BROKEN_PIPE
 121                                 pass
 122                         else:
 123                                 raise
 124
 125 if __name__ == "__main__":
 126         main()