This is an automated email from the ASF dual-hosted git repository. zjffdu pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/zeppelin.git
The following commit(s) were added to refs/heads/master by this push: new a04bdb3 [ZEPPELIN-4265]. Support more types of output for IPythonInterpreter a04bdb3 is described below commit a04bdb349bc9ff8f3ccf68687f60f5938a88cec0 Author: Jeff Zhang <zjf...@apache.org> AuthorDate: Mon Aug 5 10:31:44 2019 +0800 [ZEPPELIN-4265]. Support more types of output for IPythonInterpreter ### What is this PR for? The final target of IPythonInterpreter is to make zeppelin IPythonInterpreter compatible with jupyter. I tried several popular python viz tools. And find some of them doesn't work due to IPythonInterpreter's limitation. This PR is trying to support more types of output for IPythonInterpreter. ### What type of PR is it? [Improvement] ### Todos * [ ] - Task ### What is the Jira issue? * Open an issue on Jira https://issues.apache.org/jira/browse/ZEPPELIN/ * Put link here, and add [ZEPPELIN-*Jira number*] in PR title, eg. [ZEPPELIN-533] ### How should this be tested? ### Screenshots (if appropriate) **Pands** ![image](https://user-images.githubusercontent.com/164491/62295599-fe414180-b49f-11e9-930c-461c83fb7285.png) **Altair** ![image](https://user-images.githubusercontent.com/164491/62284745-d1ccfb80-b486-11e9-9c97-01fc2dda5975.png) **HoloView** ![image](https://user-images.githubusercontent.com/164491/62285001-5ddf2300-b487-11e9-825f-0de4bc051012.png) **HvPlot** ![image](https://user-images.githubusercontent.com/164491/62285038-6d5e6c00-b487-11e9-8eed-39bd6a405545.png) **Pandas Bokeh** ![image](https://user-images.githubusercontent.com/164491/62285342-fd9cb100-b487-11e9-94c8-b4da33cfe672.png) **Plotnine** ![image](https://user-images.githubusercontent.com/164491/62304961-9a277900-b4b1-11e9-91d3-51687cab90be.png) ### Questions: * Does the licenses files need update? No * Is there breaking changes for older versions? No * Does this needs documentation? No Author: Jeff Zhang <zjf...@apache.org> Closes #3419 from zjffdu/ZEPPELIN-4265 and squashes the following commits: 475a91f16 [Jeff Zhang] revert changes in ipython_client.py 2ebac7568 [Jeff Zhang] [ZEPPELIN-4265]. Support more types of output for IPythonInterpreter --- .../org/apache/zeppelin/python/IPythonClient.java | 42 ++- python/src/main/proto/ipython.proto | 7 +- .../src/main/resources/grpc/python/ipython_pb2.py | 398 ++++++--------------- .../main/resources/grpc/python/ipython_server.py | 70 +++- .../zeppelin/python/IPythonInterpreterTest.java | 35 +- .../apache/zeppelin/dep/DependencyResolver.java | 6 +- .../zeppelin/notebook/repo/VFSNotebookRepo.java | 4 + 7 files changed, 239 insertions(+), 323 deletions(-) diff --git a/python/src/main/java/org/apache/zeppelin/python/IPythonClient.java b/python/src/main/java/org/apache/zeppelin/python/IPythonClient.java index c729898..ec0c052 100644 --- a/python/src/main/java/org/apache/zeppelin/python/IPythonClient.java +++ b/python/src/main/java/org/apache/zeppelin/python/IPythonClient.java @@ -88,35 +88,47 @@ public class IPythonClient { LOGGER.debug("stream_execute code:\n" + request.getCode()); asyncStub.execute(request, new StreamObserver<ExecuteResponse>() { int index = 0; - boolean isPreviousOutputImage = false; @Override public void onNext(ExecuteResponse executeResponse) { + LOGGER.debug("Interpreter Streaming Output: " + executeResponse.getType() + + "\t" + executeResponse.getOutput()); + if (index != 0) { + try { + // We need to add line separator first, because zeppelin only recoginize the % at + // the line beginning. + interpreterOutput.write("\n".getBytes()); + } catch (IOException e) { + LOGGER.error("Unexpected IOException", e); + } + } + if (executeResponse.getType() == OutputType.TEXT) { try { - LOGGER.debug("Interpreter Streaming Output: " + executeResponse.getOutput()); - if (isPreviousOutputImage) { - // add '\n' when switch from image to text - interpreterOutput.write("\n%text ".getBytes()); + if (executeResponse.getOutput().startsWith("%")) { + // the output from ipython kernel maybe specify format already. + interpreterOutput.write((executeResponse.getOutput()).getBytes()); + } else { + interpreterOutput.write(("%text " + executeResponse.getOutput()).getBytes()); } - isPreviousOutputImage = false; - interpreterOutput.write(executeResponse.getOutput().getBytes()); interpreterOutput.getInterpreterOutput().flush(); } catch (IOException e) { LOGGER.error("Unexpected IOException", e); } } - if (executeResponse.getType() == OutputType.IMAGE) { + if (executeResponse.getType() == OutputType.PNG || + executeResponse.getType() == OutputType.JPEG) { try { - LOGGER.debug("Interpreter Streaming Output: IMAGE_DATA"); - if (index != 0) { - // add '\n' if this is the not the first element. otherwise it would mix the image - // with the text - interpreterOutput.write("\n".getBytes()); - } interpreterOutput.write(("%img " + executeResponse.getOutput()).getBytes()); interpreterOutput.getInterpreterOutput().flush(); - isPreviousOutputImage = true; + } catch (IOException e) { + LOGGER.error("Unexpected IOException", e); + } + } + if (executeResponse.getType() == OutputType.HTML) { + try { + interpreterOutput.write(("%html\n" + executeResponse.getOutput()).getBytes()); + interpreterOutput.getInterpreterOutput().flush(); } catch (IOException e) { LOGGER.error("Unexpected IOException", e); } diff --git a/python/src/main/proto/ipython.proto b/python/src/main/proto/ipython.proto index a54f36d..16a7ebe 100644 --- a/python/src/main/proto/ipython.proto +++ b/python/src/main/proto/ipython.proto @@ -53,7 +53,12 @@ enum IPythonStatus { enum OutputType { TEXT = 0; - IMAGE = 1; + PNG = 1; + JPEG = 2; + HTML = 3; + SVG = 4; + JSON = 5; + LaTeX = 6; } // The request message containing the code diff --git a/python/src/main/resources/grpc/python/ipython_pb2.py b/python/src/main/resources/grpc/python/ipython_pb2.py index eca3dfe..0bad160 100644 --- a/python/src/main/resources/grpc/python/ipython_pb2.py +++ b/python/src/main/resources/grpc/python/ipython_pb2.py @@ -23,7 +23,6 @@ from google.protobuf import descriptor as _descriptor from google.protobuf import message as _message from google.protobuf import reflection as _reflection from google.protobuf import symbol_database as _symbol_database -from google.protobuf import descriptor_pb2 # @@protoc_insertion_point(imports) _sym_db = _symbol_database.Default() @@ -35,7 +34,8 @@ DESCRIPTOR = _descriptor.FileDescriptor( name='ipython.proto', package='ipython', syntax='proto3', - serialized_pb=_b('\n\ripython.proto\x12\x07ipython\"\x1e\n\x0e\x45xecuteRequest\x12\x0c\n\x04\x63ode\x18\x01 \x01(\t\"l\n\x0f\x45xecuteResponse\x12&\n\x06status\x18\x01 \x01(\x0e\x32\x16.ipython.ExecuteStatus\x12!\n\x04type\x18\x02 \x01(\x0e\x32\x13.ipython.OutputType\x12\x0e\n\x06output\x18\x03 \x01(\t\"\x0f\n\rCancelRequest\"\x10\n\x0e\x43\x61ncelResponse\"1\n\x11\x43ompletionRequest\x12\x0c\n\x04\x63ode\x18\x01 \x01(\t\x12\x0e\n\x06\x63ursor\x18\x02 \x01(\x05\"%\n\x12\x43ompletionRe [...] + serialized_options=_b('\n org.apache.zeppelin.python.protoB\014IPythonProtoP\001\242\002\007IPython'), + serialized_pb=_b('\n\ripython.proto\x12\x07ipython\"\x1e\n\x0e\x45xecuteRequest\x12\x0c\n\x04\x63ode\x18\x01 \x01(\t\"l\n\x0f\x45xecuteResponse\x12&\n\x06status\x18\x01 \x01(\x0e\x32\x16.ipython.ExecuteStatus\x12!\n\x04type\x18\x02 \x01(\x0e\x32\x13.ipython.OutputType\x12\x0e\n\x06output\x18\x03 \x01(\t\"\x0f\n\rCancelRequest\"\x10\n\x0e\x43\x61ncelResponse\"1\n\x11\x43ompletionRequest\x12\x0c\n\x04\x63ode\x18\x01 \x01(\t\x12\x0e\n\x06\x63ursor\x18\x02 \x01(\x05\"%\n\x12\x43ompletionRe [...] ) _EXECUTESTATUS = _descriptor.EnumDescriptor( @@ -46,15 +46,15 @@ _EXECUTESTATUS = _descriptor.EnumDescriptor( values=[ _descriptor.EnumValueDescriptor( name='SUCCESS', index=0, number=0, - options=None, + serialized_options=None, type=None), _descriptor.EnumValueDescriptor( name='ERROR', index=1, number=1, - options=None, + serialized_options=None, type=None), ], containing_type=None, - options=None, + serialized_options=None, serialized_start=399, serialized_end=438, ) @@ -69,15 +69,15 @@ _IPYTHONSTATUS = _descriptor.EnumDescriptor( values=[ _descriptor.EnumValueDescriptor( name='STARTING', index=0, number=0, - options=None, + serialized_options=None, type=None), _descriptor.EnumValueDescriptor( name='RUNNING', index=1, number=1, - options=None, + serialized_options=None, type=None), ], containing_type=None, - options=None, + serialized_options=None, serialized_start=440, serialized_end=482, ) @@ -92,17 +92,37 @@ _OUTPUTTYPE = _descriptor.EnumDescriptor( values=[ _descriptor.EnumValueDescriptor( name='TEXT', index=0, number=0, - options=None, + serialized_options=None, type=None), _descriptor.EnumValueDescriptor( - name='IMAGE', index=1, number=1, - options=None, + name='PNG', index=1, number=1, + serialized_options=None, + type=None), + _descriptor.EnumValueDescriptor( + name='JPEG', index=2, number=2, + serialized_options=None, + type=None), + _descriptor.EnumValueDescriptor( + name='HTML', index=3, number=3, + serialized_options=None, + type=None), + _descriptor.EnumValueDescriptor( + name='SVG', index=4, number=4, + serialized_options=None, + type=None), + _descriptor.EnumValueDescriptor( + name='JSON', index=5, number=5, + serialized_options=None, + type=None), + _descriptor.EnumValueDescriptor( + name='LaTeX', index=6, number=6, + serialized_options=None, type=None), ], containing_type=None, - options=None, + serialized_options=None, serialized_start=484, - serialized_end=517, + serialized_end=565, ) _sym_db.RegisterEnumDescriptor(_OUTPUTTYPE) @@ -112,7 +132,12 @@ ERROR = 1 STARTING = 0 RUNNING = 1 TEXT = 0 -IMAGE = 1 +PNG = 1 +JPEG = 2 +HTML = 3 +SVG = 4 +JSON = 5 +LaTeX = 6 @@ -129,14 +154,14 @@ _EXECUTEREQUEST = _descriptor.Descriptor( has_default_value=False, default_value=_b("").decode('utf-8'), message_type=None, enum_type=None, containing_type=None, is_extension=False, extension_scope=None, - options=None), + serialized_options=None, file=DESCRIPTOR), ], extensions=[ ], nested_types=[], enum_types=[ ], - options=None, + serialized_options=None, is_extendable=False, syntax='proto3', extension_ranges=[], @@ -160,28 +185,28 @@ _EXECUTERESPONSE = _descriptor.Descriptor( has_default_value=False, default_value=0, message_type=None, enum_type=None, containing_type=None, is_extension=False, extension_scope=None, - options=None), + serialized_options=None, file=DESCRIPTOR), _descriptor.FieldDescriptor( name='type', full_name='ipython.ExecuteResponse.type', index=1, number=2, type=14, cpp_type=8, label=1, has_default_value=False, default_value=0, message_type=None, enum_type=None, containing_type=None, is_extension=False, extension_scope=None, - options=None), + serialized_options=None, file=DESCRIPTOR), _descriptor.FieldDescriptor( name='output', full_name='ipython.ExecuteResponse.output', index=2, number=3, type=9, cpp_type=9, label=1, has_default_value=False, default_value=_b("").decode('utf-8'), message_type=None, enum_type=None, containing_type=None, is_extension=False, extension_scope=None, - options=None), + serialized_options=None, file=DESCRIPTOR), ], extensions=[ ], nested_types=[], enum_types=[ ], - options=None, + serialized_options=None, is_extendable=False, syntax='proto3', extension_ranges=[], @@ -205,7 +230,7 @@ _CANCELREQUEST = _descriptor.Descriptor( nested_types=[], enum_types=[ ], - options=None, + serialized_options=None, is_extendable=False, syntax='proto3', extension_ranges=[], @@ -229,7 +254,7 @@ _CANCELRESPONSE = _descriptor.Descriptor( nested_types=[], enum_types=[ ], - options=None, + serialized_options=None, is_extendable=False, syntax='proto3', extension_ranges=[], @@ -253,21 +278,21 @@ _COMPLETIONREQUEST = _descriptor.Descriptor( has_default_value=False, default_value=_b("").decode('utf-8'), message_type=None, enum_type=None, containing_type=None, is_extension=False, extension_scope=None, - options=None), + serialized_options=None, file=DESCRIPTOR), _descriptor.FieldDescriptor( name='cursor', full_name='ipython.CompletionRequest.cursor', index=1, number=2, type=5, cpp_type=1, label=1, has_default_value=False, default_value=0, message_type=None, enum_type=None, containing_type=None, is_extension=False, extension_scope=None, - options=None), + serialized_options=None, file=DESCRIPTOR), ], extensions=[ ], nested_types=[], enum_types=[ ], - options=None, + serialized_options=None, is_extendable=False, syntax='proto3', extension_ranges=[], @@ -291,14 +316,14 @@ _COMPLETIONRESPONSE = _descriptor.Descriptor( has_default_value=False, default_value=[], message_type=None, enum_type=None, containing_type=None, is_extension=False, extension_scope=None, - options=None), + serialized_options=None, file=DESCRIPTOR), ], extensions=[ ], nested_types=[], enum_types=[ ], - options=None, + serialized_options=None, is_extendable=False, syntax='proto3', extension_ranges=[], @@ -322,7 +347,7 @@ _STATUSREQUEST = _descriptor.Descriptor( nested_types=[], enum_types=[ ], - options=None, + serialized_options=None, is_extendable=False, syntax='proto3', extension_ranges=[], @@ -346,14 +371,14 @@ _STATUSRESPONSE = _descriptor.Descriptor( has_default_value=False, default_value=0, message_type=None, enum_type=None, containing_type=None, is_extension=False, extension_scope=None, - options=None), + serialized_options=None, file=DESCRIPTOR), ], extensions=[ ], nested_types=[], enum_types=[ ], - options=None, + serialized_options=None, is_extendable=False, syntax='proto3', extension_ranges=[], @@ -377,7 +402,7 @@ _STOPREQUEST = _descriptor.Descriptor( nested_types=[], enum_types=[ ], - options=None, + serialized_options=None, is_extendable=False, syntax='proto3', extension_ranges=[], @@ -401,7 +426,7 @@ _STOPRESPONSE = _descriptor.Descriptor( nested_types=[], enum_types=[ ], - options=None, + serialized_options=None, is_extendable=False, syntax='proto3', extension_ranges=[], @@ -500,252 +525,65 @@ StopResponse = _reflection.GeneratedProtocolMessageType('StopResponse', (_messag _sym_db.RegisterMessage(StopResponse) -DESCRIPTOR.has_options = True -DESCRIPTOR._options = _descriptor._ParseOptions(descriptor_pb2.FileOptions(), _b('\n org.apache.zeppelin.python.protoB\014IPythonProtoP\001\242\002\007IPython')) -try: - # THESE ELEMENTS WILL BE DEPRECATED. - # Please use the generated *_pb2_grpc.py files instead. - import grpc - from grpc.beta import implementations as beta_implementations - from grpc.beta import interfaces as beta_interfaces - from grpc.framework.common import cardinality - from grpc.framework.interfaces.face import utilities as face_utilities - - - class IPythonStub(object): - """The IPython service definition. - """ - - def __init__(self, channel): - """Constructor. - - Args: - channel: A grpc.Channel. - """ - self.execute = channel.unary_stream( - '/ipython.IPython/execute', - request_serializer=ExecuteRequest.SerializeToString, - response_deserializer=ExecuteResponse.FromString, - ) - self.complete = channel.unary_unary( - '/ipython.IPython/complete', - request_serializer=CompletionRequest.SerializeToString, - response_deserializer=CompletionResponse.FromString, - ) - self.cancel = channel.unary_unary( - '/ipython.IPython/cancel', - request_serializer=CancelRequest.SerializeToString, - response_deserializer=CancelResponse.FromString, - ) - self.status = channel.unary_unary( - '/ipython.IPython/status', - request_serializer=StatusRequest.SerializeToString, - response_deserializer=StatusResponse.FromString, - ) - self.stop = channel.unary_unary( - '/ipython.IPython/stop', - request_serializer=StopRequest.SerializeToString, - response_deserializer=StopResponse.FromString, - ) - - - class IPythonServicer(object): - """The IPython service definition. - """ - - def execute(self, request, context): - """Sends code - """ - context.set_code(grpc.StatusCode.UNIMPLEMENTED) - context.set_details('Method not implemented!') - raise NotImplementedError('Method not implemented!') - - def complete(self, request, context): - """Get completion - """ - context.set_code(grpc.StatusCode.UNIMPLEMENTED) - context.set_details('Method not implemented!') - raise NotImplementedError('Method not implemented!') - - def cancel(self, request, context): - """Cancel the running statement - """ - context.set_code(grpc.StatusCode.UNIMPLEMENTED) - context.set_details('Method not implemented!') - raise NotImplementedError('Method not implemented!') - - def status(self, request, context): - """Get ipython kernel status - """ - context.set_code(grpc.StatusCode.UNIMPLEMENTED) - context.set_details('Method not implemented!') - raise NotImplementedError('Method not implemented!') - - def stop(self, request, context): - # missing associated documentation comment in .proto file - pass - context.set_code(grpc.StatusCode.UNIMPLEMENTED) - context.set_details('Method not implemented!') - raise NotImplementedError('Method not implemented!') - - - def add_IPythonServicer_to_server(servicer, server): - rpc_method_handlers = { - 'execute': grpc.unary_stream_rpc_method_handler( - servicer.execute, - request_deserializer=ExecuteRequest.FromString, - response_serializer=ExecuteResponse.SerializeToString, - ), - 'complete': grpc.unary_unary_rpc_method_handler( - servicer.complete, - request_deserializer=CompletionRequest.FromString, - response_serializer=CompletionResponse.SerializeToString, - ), - 'cancel': grpc.unary_unary_rpc_method_handler( - servicer.cancel, - request_deserializer=CancelRequest.FromString, - response_serializer=CancelResponse.SerializeToString, - ), - 'status': grpc.unary_unary_rpc_method_handler( - servicer.status, - request_deserializer=StatusRequest.FromString, - response_serializer=StatusResponse.SerializeToString, - ), - 'stop': grpc.unary_unary_rpc_method_handler( - servicer.stop, - request_deserializer=StopRequest.FromString, - response_serializer=StopResponse.SerializeToString, - ), - } - generic_handler = grpc.method_handlers_generic_handler( - 'ipython.IPython', rpc_method_handlers) - server.add_generic_rpc_handlers((generic_handler,)) - - - class BetaIPythonServicer(object): - """The Beta API is deprecated for 0.15.0 and later. - - It is recommended to use the GA API (classes and functions in this - file not marked beta) for all further purposes. This class was generated - only to ease transition from grpcio<0.15.0 to grpcio>=0.15.0.""" - """The IPython service definition. - """ - def execute(self, request, context): - """Sends code - """ - context.code(beta_interfaces.StatusCode.UNIMPLEMENTED) - def complete(self, request, context): - """Get completion - """ - context.code(beta_interfaces.StatusCode.UNIMPLEMENTED) - def cancel(self, request, context): - """Cancel the running statement - """ - context.code(beta_interfaces.StatusCode.UNIMPLEMENTED) - def status(self, request, context): - """Get ipython kernel status - """ - context.code(beta_interfaces.StatusCode.UNIMPLEMENTED) - def stop(self, request, context): - # missing associated documentation comment in .proto file - pass - context.code(beta_interfaces.StatusCode.UNIMPLEMENTED) - - - class BetaIPythonStub(object): - """The Beta API is deprecated for 0.15.0 and later. - - It is recommended to use the GA API (classes and functions in this - file not marked beta) for all further purposes. This class was generated - only to ease transition from grpcio<0.15.0 to grpcio>=0.15.0.""" - """The IPython service definition. - """ - def execute(self, request, timeout, metadata=None, with_call=False, protocol_options=None): - """Sends code - """ - raise NotImplementedError() - def complete(self, request, timeout, metadata=None, with_call=False, protocol_options=None): - """Get completion - """ - raise NotImplementedError() - complete.future = None - def cancel(self, request, timeout, metadata=None, with_call=False, protocol_options=None): - """Cancel the running statement - """ - raise NotImplementedError() - cancel.future = None - def status(self, request, timeout, metadata=None, with_call=False, protocol_options=None): - """Get ipython kernel status - """ - raise NotImplementedError() - status.future = None - def stop(self, request, timeout, metadata=None, with_call=False, protocol_options=None): - # missing associated documentation comment in .proto file - pass - raise NotImplementedError() - stop.future = None - - - def beta_create_IPython_server(servicer, pool=None, pool_size=None, default_timeout=None, maximum_timeout=None): - """The Beta API is deprecated for 0.15.0 and later. - - It is recommended to use the GA API (classes and functions in this - file not marked beta) for all further purposes. This function was - generated only to ease transition from grpcio<0.15.0 to grpcio>=0.15.0""" - request_deserializers = { - ('ipython.IPython', 'cancel'): CancelRequest.FromString, - ('ipython.IPython', 'complete'): CompletionRequest.FromString, - ('ipython.IPython', 'execute'): ExecuteRequest.FromString, - ('ipython.IPython', 'status'): StatusRequest.FromString, - ('ipython.IPython', 'stop'): StopRequest.FromString, - } - response_serializers = { - ('ipython.IPython', 'cancel'): CancelResponse.SerializeToString, - ('ipython.IPython', 'complete'): CompletionResponse.SerializeToString, - ('ipython.IPython', 'execute'): ExecuteResponse.SerializeToString, - ('ipython.IPython', 'status'): StatusResponse.SerializeToString, - ('ipython.IPython', 'stop'): StopResponse.SerializeToString, - } - method_implementations = { - ('ipython.IPython', 'cancel'): face_utilities.unary_unary_inline(servicer.cancel), - ('ipython.IPython', 'complete'): face_utilities.unary_unary_inline(servicer.complete), - ('ipython.IPython', 'execute'): face_utilities.unary_stream_inline(servicer.execute), - ('ipython.IPython', 'status'): face_utilities.unary_unary_inline(servicer.status), - ('ipython.IPython', 'stop'): face_utilities.unary_unary_inline(servicer.stop), - } - server_options = beta_implementations.server_options(request_deserializers=request_deserializers, response_serializers=response_serializers, thread_pool=pool, thread_pool_size=pool_size, default_timeout=default_timeout, maximum_timeout=maximum_timeout) - return beta_implementations.server(method_implementations, options=server_options) - - - def beta_create_IPython_stub(channel, host=None, metadata_transformer=None, pool=None, pool_size=None): - """The Beta API is deprecated for 0.15.0 and later. - - It is recommended to use the GA API (classes and functions in this - file not marked beta) for all further purposes. This function was - generated only to ease transition from grpcio<0.15.0 to grpcio>=0.15.0""" - request_serializers = { - ('ipython.IPython', 'cancel'): CancelRequest.SerializeToString, - ('ipython.IPython', 'complete'): CompletionRequest.SerializeToString, - ('ipython.IPython', 'execute'): ExecuteRequest.SerializeToString, - ('ipython.IPython', 'status'): StatusRequest.SerializeToString, - ('ipython.IPython', 'stop'): StopRequest.SerializeToString, - } - response_deserializers = { - ('ipython.IPython', 'cancel'): CancelResponse.FromString, - ('ipython.IPython', 'complete'): CompletionResponse.FromString, - ('ipython.IPython', 'execute'): ExecuteResponse.FromString, - ('ipython.IPython', 'status'): StatusResponse.FromString, - ('ipython.IPython', 'stop'): StopResponse.FromString, - } - cardinalities = { - 'cancel': cardinality.Cardinality.UNARY_UNARY, - 'complete': cardinality.Cardinality.UNARY_UNARY, - 'execute': cardinality.Cardinality.UNARY_STREAM, - 'status': cardinality.Cardinality.UNARY_UNARY, - 'stop': cardinality.Cardinality.UNARY_UNARY, - } - stub_options = beta_implementations.stub_options(host=host, metadata_transformer=metadata_transformer, request_serializers=request_serializers, response_deserializers=response_deserializers, thread_pool=pool, thread_pool_size=pool_size) - return beta_implementations.dynamic_stub(channel, 'ipython.IPython', cardinalities, options=stub_options) -except ImportError: - pass +DESCRIPTOR._options = None + +_IPYTHON = _descriptor.ServiceDescriptor( + name='IPython', + full_name='ipython.IPython', + file=DESCRIPTOR, + index=0, + serialized_options=None, + serialized_start=568, + serialized_end=891, + methods=[ + _descriptor.MethodDescriptor( + name='execute', + full_name='ipython.IPython.execute', + index=0, + containing_service=None, + input_type=_EXECUTEREQUEST, + output_type=_EXECUTERESPONSE, + serialized_options=None, + ), + _descriptor.MethodDescriptor( + name='complete', + full_name='ipython.IPython.complete', + index=1, + containing_service=None, + input_type=_COMPLETIONREQUEST, + output_type=_COMPLETIONRESPONSE, + serialized_options=None, + ), + _descriptor.MethodDescriptor( + name='cancel', + full_name='ipython.IPython.cancel', + index=2, + containing_service=None, + input_type=_CANCELREQUEST, + output_type=_CANCELRESPONSE, + serialized_options=None, + ), + _descriptor.MethodDescriptor( + name='status', + full_name='ipython.IPython.status', + index=3, + containing_service=None, + input_type=_STATUSREQUEST, + output_type=_STATUSRESPONSE, + serialized_options=None, + ), + _descriptor.MethodDescriptor( + name='stop', + full_name='ipython.IPython.stop', + index=4, + containing_service=None, + input_type=_STOPREQUEST, + output_type=_STOPRESPONSE, + serialized_options=None, + ), +]) +_sym_db.RegisterServiceDescriptor(_IPYTHON) + +DESCRIPTOR.services_by_name['IPython'] = _IPYTHON + # @@protoc_insertion_point(module_scope) diff --git a/python/src/main/resources/grpc/python/ipython_server.py b/python/src/main/resources/grpc/python/ipython_server.py index 36e0a13..3fd0a8c 100644 --- a/python/src/main/resources/grpc/python/ipython_server.py +++ b/python/src/main/resources/grpc/python/ipython_server.py @@ -49,23 +49,37 @@ class IPython(ipython_pb2_grpc.IPythonServicer): print("execute code:\n") print(request.code.encode('utf-8')) sys.stdout.flush() - stdout_queue = queue.Queue(maxsize = 10) stderr_queue = queue.Queue(maxsize = 10) - image_queue = queue.Queue(maxsize = 5) + text_queue = queue.Queue(maxsize = 10) + png_queue = queue.Queue(maxsize = 5) + jpeg_queue = queue.Queue(maxsize = 5) + html_queue = queue.Queue(maxsize = 10) def _output_hook(msg): msg_type = msg['header']['msg_type'] content = msg['content'] + print("******************* CONTENT ******************") + print(str(content)[:400]) if msg_type == 'stream': - stdout_queue.put(content['text']) + text_queue.put(content['text']) elif msg_type in ('display_data', 'execute_result'): - stdout_queue.put(content['data'].get('text/plain', '')) - if 'image/png' in content['data']: - image_queue.put(content['data']['image/png']) + if 'text/html' in content['data']: + html_queue.put(content['data']['text/html']) + elif 'image/png' in content['data']: + png_queue.put(content['data']['image/png']) + elif 'image/jpeg' in content['data']: + jpeg_queue.put(content['data']['image/jpeg']) + elif 'text/plain' in content['data']: + text_queue.put(content['data']['text/plain']) + elif 'application/javascript' in content['data']: + print('add to html queue: ' + str(content)[:100]) + html_queue.put('<script> ' + content['data']['application/javascript'] + ' </script>\n') + elif 'application/vnd.holoviews_load.v0+json' in content['data']: + html_queue.put('<script> ' + content['data']['application/vnd.holoviews_load.v0+json'] + ' </script>\n') + elif msg_type == 'error': stderr_queue.put('\n'.join(content['traceback'])) - payload_reply = [] def execute_worker(): reply = self._kc.execute_interactive(request.code, @@ -80,22 +94,33 @@ class IPython(ipython_pb2_grpc.IPythonServicer): # Execution might be stuck there: # https://github.com/jupyter/jupyter_client/blob/master/jupyter_client/blocking/client.py#L32 while t.is_alive() and self.isKernelAlive(): - while not stdout_queue.empty(): - output = stdout_queue.get() + while not text_queue.empty(): + output = text_queue.get() yield ipython_pb2.ExecuteResponse(status=ipython_pb2.SUCCESS, type=ipython_pb2.TEXT, output=output) + while not html_queue.empty(): + output = html_queue.get() + yield ipython_pb2.ExecuteResponse(status=ipython_pb2.SUCCESS, + type=ipython_pb2.HTML, + output=output) while not stderr_queue.empty(): output = stderr_queue.get() yield ipython_pb2.ExecuteResponse(status=ipython_pb2.ERROR, type=ipython_pb2.TEXT, output=output) - while not image_queue.empty(): - output = image_queue.get() + while not png_queue.empty(): + output = png_queue.get() + yield ipython_pb2.ExecuteResponse(status=ipython_pb2.SUCCESS, + type=ipython_pb2.PNG, + output=output) + while not jpeg_queue.empty(): + output = jpeg_queue.get() yield ipython_pb2.ExecuteResponse(status=ipython_pb2.SUCCESS, - type=ipython_pb2.IMAGE, + type=ipython_pb2.JPEG, output=output) + # if kernel is not alive (should be same as thread is still alive), means that we face # an unexpected issue. if not self.isKernelAlive() or t.is_alive(): @@ -104,22 +129,31 @@ class IPython(ipython_pb2_grpc.IPythonServicer): output="Ipython kernel has been stopped. Please check logs. It might be because of an out of memory issue.") return - while not stdout_queue.empty(): - output = stdout_queue.get() + while not text_queue.empty(): + output = text_queue.get() yield ipython_pb2.ExecuteResponse(status=ipython_pb2.SUCCESS, type=ipython_pb2.TEXT, output=output) + while not html_queue.empty(): + output = html_queue.get() + yield ipython_pb2.ExecuteResponse(status=ipython_pb2.SUCCESS, + type=ipython_pb2.HTML, + output=output) while not stderr_queue.empty(): output = stderr_queue.get() yield ipython_pb2.ExecuteResponse(status=ipython_pb2.ERROR, type=ipython_pb2.TEXT, output=output) - while not image_queue.empty(): - output = image_queue.get() + while not png_queue.empty(): + output = png_queue.get() yield ipython_pb2.ExecuteResponse(status=ipython_pb2.SUCCESS, - type=ipython_pb2.IMAGE, + type=ipython_pb2.PNG, + output=output) + while not jpeg_queue.empty(): + output = jpeg_queue.get() + yield ipython_pb2.ExecuteResponse(status=ipython_pb2.SUCCESS, + type=ipython_pb2.JPEG, output=output) - if payload_reply: result = [] for payload in payload_reply[0]['content']['payload']: diff --git a/python/src/test/java/org/apache/zeppelin/python/IPythonInterpreterTest.java b/python/src/test/java/org/apache/zeppelin/python/IPythonInterpreterTest.java index 4a4c809..e084bfe 100644 --- a/python/src/test/java/org/apache/zeppelin/python/IPythonInterpreterTest.java +++ b/python/src/test/java/org/apache/zeppelin/python/IPythonInterpreterTest.java @@ -185,7 +185,6 @@ public class IPythonInterpreterTest extends BasePythonInterpreterTest { // check there must be one IMAGE output boolean hasImageOutput = false; boolean hasLineText = false; - boolean hasFigureText = false; for (InterpreterResultMessage msg : interpreterResultMessages) { if (msg.getType() == InterpreterResult.Type.IMG) { hasImageOutput = true; @@ -194,14 +193,9 @@ public class IPythonInterpreterTest extends BasePythonInterpreterTest { && msg.getData().contains("matplotlib.lines.Line2D")) { hasLineText = true; } - if (msg.getType() == InterpreterResult.Type.TEXT - && msg.getData().contains("matplotlib.figure.Figure")) { - hasFigureText = true; - } } assertTrue("No Image Output", hasImageOutput); assertTrue("No Line Text", hasLineText); - assertTrue("No Figure Text", hasFigureText); // bokeh // bokeh initialization @@ -256,6 +250,35 @@ public class IPythonInterpreterTest extends BasePythonInterpreterTest { assertTrue("No Image Output", hasImageOutput); } + + // TODO(zjffdu) Enable it after new altair is released with this PR. + // https://github.com/altair-viz/altair/pull/1620 + //@Test + public void testHtmlOutput() throws InterpreterException, IOException { + // html output + InterpreterContext context = getInterpreterContext(); + InterpreterResult result = interpreter.interpret( + " import altair as alt\n" + + " print(alt.renderers.active)\n" + + " alt.renderers.enable(\"colab\")\n" + + " import altair as alt\n" + + " # load a simple dataset as a pandas DataFrame\n" + + " from vega_datasets import data\n" + + " cars = data.cars()\n" + + " \n" + + " alt.Chart(cars).mark_point().encode(\n" + + " x='Horsepower',\n" + + " y='Miles_per_Gallon',\n" + + " color='Origin',\n" + + " ).interactive()", context); + assertEquals(InterpreterResult.Code.SUCCESS, result.code()); + assertEquals(2, context.out.size()); + assertEquals(InterpreterResult.Type.TEXT, + context.out.toInterpreterResultMessage().get(0).getType()); + assertEquals(InterpreterResult.Type.HTML, + context.out.toInterpreterResultMessage().get(1).getType()); + } + @Test public void testGrpcFrameSize() throws InterpreterException, IOException { tearDown(); diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/dep/DependencyResolver.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/dep/DependencyResolver.java index 0acfca9..495c69b 100644 --- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/dep/DependencyResolver.java +++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/dep/DependencyResolver.java @@ -96,7 +96,7 @@ public class DependencyResolver extends AbstractDependencyResolver { File destFile = new File(destPath, srcFile.getName()); if (!destFile.exists() || !FileUtils.contentEquals(srcFile, destFile)) { FileUtils.copyFile(srcFile, destFile); - logger.info("copy {} to {}", srcFile.getAbsolutePath(), destPath); + logger.debug("copy {} to {}", srcFile.getAbsolutePath(), destPath); } } } @@ -114,7 +114,7 @@ public class DependencyResolver extends AbstractDependencyResolver { if (!destFile.exists() || !FileUtils.contentEquals(srcFile, destFile)) { FileUtils.copyFile(srcFile, destFile); - logger.info("copy {} to {}", srcFile.getAbsolutePath(), destPath); + logger.debug("copy {} to {}", srcFile.getAbsolutePath(), destPath); } } @@ -142,7 +142,7 @@ public class DependencyResolver extends AbstractDependencyResolver { List<File> files = new LinkedList<>(); for (ArtifactResult artifactResult : listOfArtifact) { files.add(artifactResult.getArtifact().getFile()); - logger.info("load {}", artifactResult.getArtifact().getFile().getAbsolutePath()); + logger.debug("load {}", artifactResult.getArtifact().getFile().getAbsolutePath()); } return files; diff --git a/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/repo/VFSNotebookRepo.java b/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/repo/VFSNotebookRepo.java index 012f6a4..d8d8df0 100644 --- a/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/repo/VFSNotebookRepo.java +++ b/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/repo/VFSNotebookRepo.java @@ -102,6 +102,10 @@ public class VFSNotebookRepo implements NotebookRepo { private Map<String, NoteInfo> listFolder(FileObject fileObject) throws IOException { Map<String, NoteInfo> noteInfos = new HashMap<>(); if (fileObject.isFolder()) { + if (fileObject.getName().getBaseName().startsWith(".")) { + LOGGER.warn("Skip hidden folder: " + fileObject.getName().getPath()); + return noteInfos; + } for (FileObject child : fileObject.getChildren()) { noteInfos.putAll(listFolder(child)); }