ensight_grpc.py

 import grpc
 import os, os.path
 import platform
 import random
 import subprocess
 import threading
 import uuid
 from concurrent import futures
  
 # these modules are the result of running protoc on the .proto file
 import ensight_pb2
 import ensight_pb2_grpc
 import dynamic_scene_graph_pb2
 import dynamic_scene_graph_pb2_grpc
  
 # Shared memory interface
 import ensight_grpc_shmem
  
  
  
  
  
 class EnSightGRPC(object):
     
     def __init__(self, port=12345, host='127.0.0.1', version=''):
         self._host_host = host
         self._port_port = port
         self._pid_pid = None
         self._channel_channel = None
         self._stub_stub = None
         self._dsg_stub_dsg_stub = None
         self._version_version = version
         self._security_token_security_token = str(random.randint(0, 1000000))
         # Streaming APIs
         # Images
         self._image_stream_image_stream = None
         self._image_thread_image_thread = None
         self._image_image = None
         self._image_number_image_number = 0
         self._shmem_client_shmem_client = None
         self._shmem_filename_shmem_filename = None
         # Event (strings)
         self._event_stream_event_stream = None
         self._event_thread_event_thread = None
         self._events_events = list()
         self._prefix_prefix = None
         self._sub_service_sub_service = None
  
     
     def host(self):
         return self._host_host
  
     
     def port(self):
         return self._port_port
  
     
     def set_security_token(self, n):
         self._security_token_security_token = n
  
     
     def security_token(self):
         return self._security_token_security_token
  
     
     def shutdown(self):
         # remove any subscription services we may have started
         self.unsubscribeunsubscribe()
         # if we launched EnSight, shut it down.
         if self._pid_pid is not None:
             _ = self.stop_serverstop_server()
         # shutdown any shared memory client
         if self._shmem_client_shmem_client is not None:
             ensight_grpc_shmem.stream_destroy(self._shmem_client_shmem_client)
             self._shmem_client_shmem_client = None
         # destroy any shared memory file we generated
         if self._shmem_filename_shmem_filename is not None:
             try:
                 os.remove(self._shmem_filename_shmem_filename)
             except Exception as e:
                 print("Unable to delete shared memory file: {}".format(str(e)))
             self._shmem_filename_shmem_filename = None
  
     
     def start_server(self):
         if self._pid_pid is not None:
             return self._pid_pid
  
         my_env = os.environ.copy()
         exe = f'ensight{self._version}'
         cmd = [exe, '-batch', '-grpc_server', str(self._port_port)]
         if self._security_token_security_token:
             cmd.append("-security")
             cmd.append(self._security_token_security_token)
         if platform.system() == 'Windows':
             DETACHED_PROCESS = 0x00000008
             cmd[0] += ".bat"
             cmd.append('-minimize_console')
             self._pid_pid = subprocess.Popen(cmd, creationflags=DETACHED_PROCESS, close_fds=True, env=my_env).pid
         else:
             self._pid_pid = subprocess.Popen(exe, close_fds=True, env=my_env).pid
         return self._pid_pid
  
     
     def stop_server(self):
         response = None
         # if we are connected and we started the server, we will emit the 'exit' message
         if (self._pid_pid is not None) and self.is_connectedis_connected():
             response = self._stub_stub.Exit(ensight_pb2.ExitRequest(), metadata=self.metadatametadata())
         self._stub_stub = None
         self._dsg_stub_dsg_stub = None
         self._channel_channel.close()
         self._channel_channel = None
         self._pid_pid = None
         return response
  
     
     def is_connected(self):
         return self._channel_channel is not None
  
     
     def connect(self, timeout=15.0):
         if self._channel_channel is not None:
             return
         self._channel_channel = grpc.insecure_channel("{}:{}".format(self._host_host, self._port_port),
                                               options=[('grpc.max_receive_message_length', -1)])
         try:
             grpc.channel_ready_future(self._channel_channel).result(timeout=timeout)
         except grpc.FutureTimeoutError:
             self._channel_channel = None
             return
         self._stub_stub = ensight_pb2_grpc.EnSightServiceStub(self._channel_channel)
         self._dsg_stub_dsg_stub = dynamic_scene_graph_pb2_grpc.DynamicSceneGraphServiceStub(self._channel_channel)
  
     
     def metadata(self):
         ret = list()
         if self._security_token_security_token is not None:
             s = self._security_token_security_token
             if type(s) == str:
                  s = s.encode("utf-8")
             ret.append( (b'shared_secret', s) )
         return ret
  
     
     def render(self, width=640, height=480, aa=1, raw=False, highlighting=False):
         self.connectconnect()
         ret_type = ensight_pb2.RenderRequest.IMAGE_PNG
         if raw:
             ret_type = ensight_pb2.RenderRequest.IMAGE_RAW
         try:
             response = self._stub_stub.RenderImage(ensight_pb2.RenderRequest(type=ret_type, image_width=width,
                                                                         image_height=height, image_aa_passes=aa,
                                                                         include_highlighting=highlighting),
                                               metadata=self.metadatametadata())
         except:
             raise IOError("gRPC connection dropped")
         return response.value
  
     
     def geometry(self):
         self.connectconnect()
         try:
             response = self._stub_stub.GetGeometry(
                 ensight_pb2.GeometryRequest(type=ensight_pb2.GeometryRequest.GEOMETRY_GLB), metadata=self.metadatametadata())
         except:
             raise IOError("gRPC connection dropped")
         return response.value
  
     
     def command(self, command_string, do_eval=True, json=False):
         self.connectconnect()
         flags = ensight_pb2.PythonRequest.EXEC_RETURN_PYTHON
         if json:
             flags = ensight_pb2.PythonRequest.EXEC_RETURN_JSON
         if not do_eval:
             flags = ensight_pb2.PythonRequest.EXEC_NO_RESULT
         try:
             response = self._stub_stub.RunPython(ensight_pb2.PythonRequest(type=flags, command=command_string),
                                             metadata=self.metadatametadata())
         except:
             raise IOError("gRPC connection dropped")
         if response.error < 0:
             raise RuntimeError("Remote execution error")
         if flags == ensight_pb2.PythonRequest.EXEC_NO_RESULT:
             return None
         elif flags == ensight_pb2.PythonRequest.EXEC_RETURN_PYTHON:
             return eval(response.value)
         return response.value
  
     
     def image_stream_enable(self, flip_vertical=False):
         if self._image_stream_image_stream is not None:
             return
         self.connectconnect()
         self._image_stream_image_stream = self._stub_stub.GetImageStream(
             ensight_pb2.ImageStreamRequest(flip_vertical=flip_vertical, chunk=True), metadata=self.metadatametadata())
         self._image_thread_image_thread = threading.Thread(target=self.poll_imagespoll_images)
         self._image_thread_image_thread.daemon = True
         self._image_thread_image_thread.start()
  
     
     def get_image(self):
         return self._image_image, self._image_number_image_number
  
     
     def put_image(self, the_image):
         self._image_image = the_image
         self._image_number_image_number += 1
  
     
     def poll_images(self):
         try:
             while self._stub_stub is not None:
                 if self._shmem_client_shmem_client is not None:
                     img = ensight_grpc_shmem.stream_lock(self._shmem_client_shmem_client)
                     if type(img) is dict:
                         the_image = dict(pixels=img['pixeldata'], width=img['width'], height=img['height'])
                         self.put_imageput_image(the_image)
                         ensight_grpc_shmem.stream_unlock(self._shmem_client_shmem_client)
  
                 if self._image_stream_image_stream is not None:
                     img = self._image_stream_image_stream.next()
                     buffer = img.pixels
  
                     while not img.final:
                         img = self._image_stream_image_stream.next()
                         buffer += img.pixels
  
                     the_image = dict(pixels=buffer, width=img.width, height=img.height)
                     self.put_imageput_image(the_image)
         except:
             # signal that the gRPC connection has broken
             self._image_stream_image_stream = None
             self._image_thread_image_thread = None
             self._image_image = None
  
     
     def image_stream_is_enabled(self):
         return self._image_stream_image_stream is not None
  
     
     def prefix(self):
         # prefix URIs will have the format:  "grpc://{uuid}/{callbackname}?enum={}&uid={}"
         if self._prefix_prefix is None:
             self._prefix_prefix = "grpc://" + str(uuid.uuid1()) + "/"
         return self._prefix_prefix
  
     
     def event_stream_enable(self):
         if self._event_stream_event_stream is not None:
             return
         self.connectconnect()
         self._event_stream_event_stream = self._stub_stub.GetEventStream(ensight_pb2.EventStreamRequest(prefix=self.prefixprefix()),
                                                        metadata=self.metadatametadata())
         self._event_thread_event_thread = threading.Thread(target=self.poll_eventspoll_events)
         self._event_thread_event_thread.daemon = True
         self._event_thread_event_thread.start()
  
     
     def dynamic_scene_graph_stream(self, client_cmds):
         self.connectconnect()
         return self._dsg_stub_dsg_stub.GetSceneStream(client_cmds, metadata=self.metadatametadata())
  
     
     def event_stream_is_enabled(self):
         return self._event_stream_event_stream is not None
  
     
     def get_event(self):
         try:
             return self._events_events.pop(0)
         except IndexError:
             return None
  
     
     def put_event(self, evt):
         self._events_events.append(evt)
  
     
     def poll_events(self):
         try:
             while self._stub_stub is not None:
                 evt = self._event_stream_event_stream.next()
                 self.put_eventput_event(evt)
         except:
             # signal that the gRPC connection has broken
             self._event_stream_event_stream = None
             self._event_thread_event_thread = None
  
     
     def start_sub_service(self):
         try:
             if self._sub_service_sub_service is not None:
                 return
             self._sub_service_sub_service = EnSightSubServicer(parent=self)
             self._sub_service_sub_service.start()
         except:
             self._sub_service_sub_service = None
  
     
     def subscribe_events(self):
         self.start_sub_servicestart_sub_service()
         self.connectconnect()
         conn_type = ensight_pb2.SubscribeEventOptions.GRPC
         options = dict(uri=self._sub_service_sub_service._uri)
         event_options = ensight_pb2.SubscribeEventOptions(prefix=self.prefixprefix(), type=conn_type, options=options)
         _ = self._stub_stub.SubscribeEvents(event_options, metadata=self.metadatametadata())
  
     
     def subscribe_images(self, flip_vertical=False, use_shmem=True):
         self.connectconnect()
         if use_shmem:
             try:
                 # we need a shared memory file
                 self._shmem_filename_shmem_filename = self.find_filenamefind_filename()
                 if self._shmem_filename_shmem_filename is not None:
                     conn_type = ensight_pb2.SubscribeImageOptions.SHARED_MEM
                     options = dict(filename=self._shmem_filename_shmem_filename)
                     image_options = ensight_pb2.SubscribeImageOptions(prefix=self.prefixprefix(), type=conn_type,
                                                                       options=options, flip_vertical=flip_vertical,
                                                                       chunk=False)
                     test = self._stub_stub.SubscribeImages(image_options, metadata=self.metadatametadata())
                     # start the local server
                     self._shmem_client_shmem_client = ensight_grpc_shmem.stream_create(self._shmem_filename_shmem_filename)
                     # turn on the polling thread
                     self._image_thread_image_thread = threading.Thread(target=self.poll_imagespoll_images)
                     self._image_thread_image_thread.daemon = True
                     self._image_thread_image_thread.start()
                     return
             except Exception as e:
                 print("Unable to subscribe to an image stream via shared memory: {}".format(str(e)))
  
         self.start_sub_servicestart_sub_service()
         conn_type = ensight_pb2.SubscribeImageOptions.GRPC
         options = dict(uri=self._sub_service_sub_service._uri)
         image_options = ensight_pb2.SubscribeImageOptions(prefix=self.prefixprefix(), type=conn_type, options=options,
                                                           flip_vertical=flip_vertical, chunk=True)
         _ = self._stub_stub.SubscribeImages(image_options, metadata=self.metadatametadata())
  
     
     def unsubscribe(self):
         if self.is_connectedis_connected():
             if self._sub_service_sub_service is not None:
                 prefix = ensight_pb2.Prefix(prefix=self.prefixprefix())
                 self._stub_stub.Unsubscribe(prefix, metadata=self.metadatametadata())
                 self._sub_service_sub_service = None
  
     
     @classmethod
     def find_filename(cls, size=1024*1024*25):
         for i in range(100):
             filename = os.path.join(os.getcwd(), "shmem_{}.bin".format(os.getpid()+i))
             if not os.path.exists(filename):
                 try:
                     tmp = open(filename, "wb")
                     tmp.write(b'\0'*size)  # 25MB
                     tmp.close()
                     return filename
                 except:
                     pass
         return None
  
  
  
 class EnSightSubServicer(ensight_pb2_grpc.EnSightSubscriptionServicer):
  
     def __init__(self, parent=None):
         self._server = None
         self._uri = ""
         self._parent = parent
  
     def PublishEvent(self, request, context):
         if self._parent is not None:
             self._parent.put_event(request)
         return ensight_pb2.GenericResponse(str='Event Published')
  
     def PublishImage(self, request_iterator, context):
         img = request_iterator.next()
         buffer = img.pixels
         while not img.final:
             img = request_iterator.next()
             buffer += img.pixels
         the_image = dict(pixels=buffer, width=img.width, height=img.height)
         if self._parent is not None:
             self._parent.put_image(the_image)
         return ensight_pb2.GenericResponse(str='Image Published')
  
     def start(self):
         self._server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
         ensight_pb2_grpc.add_EnSightSubscriptionServicer_to_server(
             self, self._server)
         # Start the server on localhost with a random port
         port = self._server.add_insecure_port('localhost:0')
         self._uri = 'localhost:' + str(port)
         self._server.start()
  
  
Post-processing tools 2023 R2

Connect with Ansys
