py/tools/chrome_debugger.py

#!/usr/bin/env python
# Copyright 2014 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""Chrome remote debugging protocol.

This module provides controlling Chrome browser by the Remote Debugging
Protocol.

Example:

  To change current tab to Google,

  chrome = ChromeRemoteDebugger()
  while not chrome.IsReady():
    time.sleep(1)
  chrome.SetActivePage()
  chrome.PageNavigate('http://www.google.com')

See https://developer.chrome.com/devtools/docs/debugger-protocol for more
information."""


from __future__ import print_function

import json
import logging
import sys
import threading
import time
import urllib2

from ws4py.client.threadedclient import WebSocketClient


DEFAULT_CHROME_DEBUG_URL = 'http://127.0.0.1:9222'


class ChromeRemoteDebugger(object):
  """An interface to control Chrome brower by remote debugging protocol.

  Args:
    debug_url: An URL to connect to debug port (--remote-debugging-port) of
               running Chrome instance.

  Attributes:
    id: An incremental identifier for debugging protocol.
    lock: Internal lock.
    active_websocket: An websocket client to active page.
  """

  ANY = 'any value'

  def __init__(self, debug_url=DEFAULT_CHROME_DEBUG_URL):
    self.debug_url = debug_url
    self.id = 1
    self.active_websocket = None
    self.lock = threading.Lock()

  def IsReady(self):
    """Checks if a browser instance is available (with debugging ports enabled).

    Returns:
      True if an instance is ready, otherwise False.
    """
    with self.lock:
      if self.active_websocket:
        return True
    try:
      self.GetPages()
      return True
    except:
      return False

  def GetPages(self, page_type=ANY):
    """Returns current pages of running Chrome browser.

    Args:
      page_type: Filters result by given type, or ANY to return all types.

    Returns:
      A list representing PageSet in Chrome remote debugging protocol.
    """
    page_set = json.load(urllib2.urlopen(self.debug_url + '/json'))
    if page_type is not self.ANY:
      return [page for page in page_set if page['type'] == page_type]
    return page_set

  def SetActivePage(self, page=ANY):
    """Sets a page as active session for sending commands.

    Args:
      page: A page instance (returned by GetPages), or ANY to activate the first
            normal webpage (type 'other').
    """
    if page is self.ANY:
      page = self.GetPages('other')[0]
    ws = None
    if page is not None:
      ws = WebSocketClient(page['webSocketDebuggerUrl'])
      ws.connect()
    with self.lock:
      if self.active_websocket:
        self.active_websocket.close()
      self.active_websocket = ws

  def SendCommand(self, command):
    """Sends a remote debugging websocket command to Chrome browser.

    Args:
      command: A dictionary of remote debugging command.
    """
    command = command.copy()
    with self.lock:
      command.update({'id': self.id})
      self.id += 1
      self.active_websocket.send(json.dumps(command))

  def PageNavigate(self, url):
    """Navigates current page to the given URL.

    Args:
      url: An URL to open in browser.
    """
    self.SendCommand({'method': 'Page.navigate',
                      'params': {'url': url}})


if __name__ == '__main__':
  if len(sys.argv) != 3:
    exit('Usage: %s method_name json_params' % sys.argv[0])
  chrome = ChromeRemoteDebugger()
  chrome.SetActivePage()
  chrome.SendCommand({'method': sys.argv[1],
                      'params': json.loads(sys.argv[2])})