Refresh chart examples when chart source or chart example is modified

watch_examples.py

from selenium import webdriver
from selenium.webdriver.common.keys import Keys
from time import sleep
import os
import subprocess
from watchdog.observers import Observer
from watchdog.events import PatternMatchingEventHandler


CHARTS = {
    'bar': {
        'script': "bar_multi.py",
        'url': "bar_multi.html",
        'zoom_level': 4
    },
    'box': {
        'script': "boxplot_multi.py",
        'url': "boxplot_multi.html",
        'zoom_level': 4
    }
}

driver = webdriver.Firefox()

EXECUTE_PATH = '/Users/nroth/workspace/bokeh/examples/charts/file'
WATCH_PATH = '/Users/nroth/workspace/bokeh/bokeh/charts'

from os import listdir
from os.path import isfile, join

execute_files = [f for f in listdir(EXECUTE_PATH) if isfile(join(EXECUTE_PATH, f)) and
                 '.py' in f]


class ChartsChangeHandler(PatternMatchingEventHandler):

    def __init__(self, chart_browser, **kwargs):
        super(ChartsChangeHandler, self).__init__(**kwargs)
        self.chart_browser = chart_browser

    def on_modified(self, event):
        for chart in CHARTS.values():
            subprocess.Popen(['/Users/nroth/anaconda/envs/bokeh3/bin/python',
                              os.path.join(EXECUTE_PATH, chart['script'])],
                              cwd=EXECUTE_PATH)

        self.chart_browser.refresh_tabs()


class ChartViewer(object):

    def __init__(self, driver, charts, path_to_watch=WATCH_PATH,
                 path_to_scripts=EXECUTE_PATH,
                 delay=5):
        """

        Args:
            driver (selenium.webdriver.remote.webdriver.WebDriver):
            charts (dict):
            delay (int):
        """
        self.driver = driver
        self.charts_dir = charts
        self.delay = delay
        self.path_to_watch = path_to_watch
        self.path_to_scripts = path_to_scripts
        self.chart_watcher = ChartsChangeHandler(chart_browser=self, patterns='*.py')

        self.open_urls()
        self.watch_tabs()

    @property
    def charts(self):
        return list(self.charts_dir.values())

    def open_urls(self):
        charts = self.charts.copy()
        self.open_url(charts.pop(0))
        for chart in charts:
            self.open_new_tab()
            self.open_url(chart)

    def open_new_tab(self):
        self.driver.find_element_by_tag_name('body').send_keys(Keys.COMMAND + 't')

    def next_tab(self):
        self.driver.find_element_by_tag_name('body').send_keys(Keys.CONTROL + Keys.TAB)

    def open_url(self, chart):
        self.driver.get(url="file://" + os.path.join(EXECUTE_PATH, chart['url']))
        self.shrink_page(times=chart['zoom_level'])

    def watch_tabs(self):

        observer = Observer()
        observer.schedule(self.chart_watcher, self.path_to_watch, recursive=True)
        observer.schedule(self.chart_watcher, self.path_to_scripts, recursive=True)
        observer.start()

        try:
            while True:
                sleep(1)
        except KeyboardInterrupt:
            observer.stop()

        observer.join()

    def refresh_tabs(self):
        for chart in self.charts:
            self.next_tab()
            self.driver.refresh()

    def shrink_page(self, times=1):
        for time in range(times):
            self.driver.find_element_by_tag_name('body').send_keys(Keys.COMMAND + '-')


if __name__ == '__main__':
    viewer = ChartViewer(driver=driver, charts=CHARTS, delay=5)