Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
:throws: :class:`~iterm2.RPCException` if something goes wrong
.. seealso:: Example ":ref:`targeted_input_example`"
"""
result = await iterm2.rpc.async_register_web_view_tool(
connection,
display_name,
identifier,
reveal_if_already_registered,
url)
status = result.register_tool_response.status
# pylint: disable=no-member
if status == iterm2.api_pb2.RegisterToolResponse.Status.Value("OK"):
return None
raise iterm2.rpc.RPCException(result.register_tool_response)
async def async_invoke_method(connection, receiver, invocation, timeout):
"""Convenience wrapper around async_invoke_function for methods."""
assert receiver
response = await iterm2.rpc.async_invoke_function(
connection,
invocation,
receiver=receiver,
timeout=timeout)
which = response.invoke_function_response.WhichOneof('disposition')
if which == 'error':
if (response.invoke_function_response.error.status ==
iterm2.api_pb2.InvokeFunctionResponse.Status.Value("TIMEOUT")):
raise iterm2.rpc.RPCException("Timeout")
raise iterm2.rpc.RPCException("{}: {}".format(
iterm2.api_pb2.InvokeFunctionResponse.Status.Name(
response.invoke_function_response.error.status),
response.invoke_function_response.error.error_reason))
return json.loads(response.invoke_function_response.success.json_result)
async def async_inject(self, data: bytes) -> None:
"""
Injects data as though it were program output.
:param data: A byte array to inject.
:throws: :class:`~iterm2.rpc.RPCException` if something goes wrong.
.. seealso:: Example ":ref:`cls_example`"
"""
response = await iterm2.rpc.async_inject(
self.connection, data, [self.__session_id])
status = response.inject_response.status[0]
# pylint: disable=no-member
if status != iterm2.api_pb2.InjectResponse.Status.Value("OK"):
raise iterm2.rpc.RPCException(
iterm2.api_pb2.InjectResponse.Status.Name(status))
future: asyncio.Future = asyncio.Future()
self.future = future
await self.future
self.future = None
if not self.want_contents:
return None
# pylint: disable=no-member
result = await iterm2.rpc.async_get_screen_contents(
self.connection,
self.session_id,
None)
if (result.get_buffer_response.status == iterm2.
api_pb2.GetBufferResponse.Status.Value("OK")):
return ScreenContents(result.get_buffer_response)
raise iterm2.rpc.RPCException(
iterm2.api_pb2.GetBufferResponse.Status.Name(
result.get_buffer_response.status))
:throws: :class:`~iterm2.rpc.RPCException` if something goes wrong.
"""
response = await iterm2.rpc.async_invoke_function(
self.connection,
invocation,
session_id=self.session_id,
timeout=timeout)
which = response.invoke_function_response.WhichOneof('disposition')
# pylint: disable=no-member
if which == 'error':
if (response.invoke_function_response.error.status ==
iterm2.api_pb2.InvokeFunctionResponse.Status.
Value("TIMEOUT")):
raise iterm2.rpc.RPCException("Timeout")
raise iterm2.rpc.RPCException("{}: {}".format(
iterm2.api_pb2.InvokeFunctionResponse.Status.Name(
response.invoke_function_response.error.status),
response.invoke_function_response.error.error_reason))
return json.loads(
response.invoke_function_response.success.json_result)
iterm2.api_pb2.SetProfilePropertyResponse.Status.Name(
status))
return
# Deprecated code path, in use by 3.3.0beta9 and earlier.
for key, json_value in write_only_profile.values.items():
response = await iterm2.rpc.async_set_profile_property_json(
self.connection,
self.session_id,
key,
json_value)
status = response.set_profile_property_response.status
# pylint: disable=no-member
if (status != iterm2.api_pb2.SetProfilePropertyResponse.Status.
Value("OK")):
raise iterm2.rpc.RPCException(
iterm2.api_pb2.SetProfilePropertyResponse.Status.Name(
status))
self.__windowed_coord_range)
# pylint: disable=no-member
if (result.get_buffer_response.status == iterm2.
api_pb2.GetBufferResponse.Status.Value("OK")):
screen_contents = iterm2.screen.ScreenContents(
result.get_buffer_response)
built_string = ""
i = 0
while i < screen_contents.number_of_lines:
line = screen_contents.line(i)
i += 1
built_string += line.string
if line.hard_eol:
built_string += "\n"
return built_string
raise iterm2.rpc.RPCException(
iterm2.api_pb2.GetBufferResponse.Status.Name(
result.get_buffer_response.status))
"""Sets the current set of broadcast domains.
:param connection: The connection to iTerm2.
:param broadcast_domains: The new collection of broadcast domains.
.. seealso:: Example ":ref:`enable_broadcasting_example`"
"""
response = await iterm2.rpc.async_set_broadcast_domains(connection, list(
map(lambda d: list(
map(lambda s: s.session_id,
d.sessions)),
broadcast_domains)))
# pylint: disable=no-member
if (response.set_broadcast_domains_response.status !=
iterm2.api_pb2.SetBroadcastDomainsResponse.Status.Value("OK")):
raise iterm2.rpc.RPCException(
iterm2.api_pb2.SetBroadcastDomainsResponse.Status.Name(
response.set_broadcast_domains_response.status))
"""
Closes the tab.
:param force: If True, the user will not be prompted for a
confirmation.
:throws: :class:`RPCException` if something goes wrong.
.. seealso:: Example ":ref:`close_to_the_right_example`"
"""
result = await iterm2.rpc.async_close(
self.connection, tabs=[self.__tab_id], force=force)
status = result.close_response.statuses[0]
# pylint: disable=no-member
if status != iterm2.api_pb2.CloseResponse.Status.Value("OK"):
raise iterm2.rpc.RPCException(
iterm2.api_pb2.CloseResponse.Status.Name(status))
:returns: The result of the invocation if successful.
:throws: :class:`~iterm2.rpc.RPCException` if something goes wrong.
"""
response = await iterm2.rpc.async_invoke_function(
connection,
invocation,
timeout=timeout)
which = response.invoke_function_response.WhichOneof('disposition')
if which == 'error':
# pylint: disable=no-member
if (response.invoke_function_response.error.status ==
iterm2.api_pb2.InvokeFunctionResponse.Status.Value("TIMEOUT")):
raise iterm2.rpc.RPCException("Timeout")
raise iterm2.rpc.RPCException("{}: {}".format(
iterm2.api_pb2.InvokeFunctionResponse.Status.Name(
response.invoke_function_response.error.status),
response.invoke_function_response.error.error_reason))
return json.loads(response.invoke_function_response.success.json_result)