Source code for tacoma.interactive

# -*- coding: utf-8 -*-
This module provides the necessary functions to start up a local
HTTP server and open an interactive d3-visualization of one or
several temporal networks.
from __future__ import print_function

import os
import sys

import http.server
import webbrowser
import time
import threading

import copy

import wget
import shutil

import json
from distutils.dir_util import copy_tree, mkpath

import tacoma as tc
from tacoma.data_io import mkdirp_customdir

ht09_config = {
    "temporal_network_files": [
    "edges_coordinate_files": [
    "plot_width": 320,
    "titles": [
    "network_plot_height": 250,
    "edges_plot_height": 100,
    "padding": 10,
    "start_it": 0,
    "node_radius": 2.5,
    "link_distance": 10,
    "node_charge": -8,
    "edge_line_width": 1,
    "font_size_in_px": 14,
    "link_width": 1,
    "d3_format_string": ".3f",

dtu_config = {
    "plot_width" : 320 ,
    "network_plot_height" : 250,
    "edges_plot_height" : 100,
    "padding" : 10,
    "start_it" : 0,
    "node_radius" : 1.5,
    "link_distance" : 7,
    "node_charge": -3,
    "edge_line_width" : 0.5,
    "node_edge_width" : 1,
    "font_size_in_px" : 14,
    "link_width" : 0.8

hs13_config = {
    "plot_width" : 320 ,
    "network_plot_height" : 350,
    "edges_plot_height" : 100,
    "padding" : 10,
    "start_it" : 0,
    "node_radius" : 2.5,
    "link_distance" : 8,
    "node_charge": -5,
    "edge_line_width" : 1,
    "font_size_in_px" : 14,
    "link_width" : 1,
    "node_edge_width": 1

standard_config = ht09_config

html_source_path = os.path.join(tc.__path__[0], 'interactive')

def _make_and_get_directory(path):
    """Simulate ``mkdir -p`` and return the path of the repository"""
    directory, _ = os.path.split(
    return directory

[docs]def download_d3(): """Download `d3.v4.min.js` and save it in `~/.tacoma/d3`, if the file does not exist yet.""" url = "" filename = "~/.tacoma/web/d3.v4/d3.v4.min.js" target_name = "d3.min.js" if not os.path.exists(os.path.abspath(os.path.expanduser(filename))): # get directory name for download directory = _make_and_get_directory(filename) # download print("downloading d3 ..."), out=directory) # move to better name os.rename(os.path.join(directory, target_name), os.path.abspath(os.path.expanduser(filename)))
[docs]def prepare_visualization_directory(): """Move all files from the tacoma/interactive directory to ~/.tacoma/web""" src = html_source_path dst = os.path.abspath(os.path.expanduser("~/.tacoma/web/")) #if not os.path.exists(os.path.join(dst, "index.html")): # always copy source files to the subdirectory copy_tree(src, dst)
[docs]def prepare_export_directory(d,subdir): """Move all files from the tacoma/interactive directory to directory ``d``""" src = html_source_path dst = os.path.abspath(os.path.expanduser(d)) # make directory and subdirectory mkpath(os.path.join(dst, subdir)) # always copy source files to the subdirectory copy_tree(src, dst) d3file = os.path.abspath(os.path.expanduser("~/.tacoma/web/d3.v4/d3.v4.min.js")) new_dst = os.path.join(dst,"d3.v4") if not os.path.exists(d3file): raise FileNotFoundError('D3 has not been downloaded yet. Please call `tacoma.interactive.download_d3()`.') # copy d3 from the tacoma directory src = os.path.abspath(os.path.expanduser("~/.tacoma/web/d3.v4/")) copy_tree(src, new_dst)
[docs]class StoppableHTTPServer(http.server.HTTPServer): """Taken from """ def __init__(self, server_address, handler, subfolder): http.server.HTTPServer.__init__(self, server_address, handler) self.subfolder = subfolder while subfolder.endswith('/'): subfolder = subfolder[:-1] self.subjson = subfolder + '_config.json'
[docs] def run(self): try: self.serve_forever() except OSError: pass
[docs] def stop_this(self): # Clean-up server (close socket, etc.) print('was asked to stop the server') self.server_close() # try: if os.path.exists(self.subjson): os.remove(self.subjson) # except if os.path.exists(self.subfolder): try: # os.rmdir(self.subfolder) shutil.rmtree(self.subfolder) except FileNotFoundError as e: raise e # os.chdir(self.cwd) print('deleted all files')
# def __del__(self): # self.stop_this() def _get_prepared_network(tn, dt, time_unit, time_normalization_factor): """Prepare the provided network (i.e. bin it in discrete time)""" tn_b = tc.bin(tn, dt=dt) # rebin the network # rescale the network's time tn_b.t = [t * time_normalization_factor for t in tn_b.t] tn_b.tmax *= time_normalization_factor if time_unit is None: time_unit = "" tn_b.time_unit = time_unit # set time unit return tn_b
[docs]def visualize(temporal_networks, frame_dt, time_normalization_factor=1, time_unit=None, titles=None, config=None, port=8226, export_path=None, ): """ Visualize a temporal network or a list of temporal networks interactively. This routine starts up an HTTP server, bins the networks according to the time step ``frame_dt`` and copies them to ``~/.tacoma/web``. Subsequently, a the interaction is started in the standard browser. The visualization is stopped with KeyboardInterrupt. The temporary temporal network files will subsequently be deleted. Parameters ---------- temporal_networks : an instance of :class:`_tacoma.edge_changes`, :class:`_tacoma.edge_lists`, :class:`_tacoma.edge_lists_with_histograms`, :class:`_tacoma.edge_changes_with_histograms` or a list containing those. The temporal networks to visualize. If a list is provided, all networks need to have the same `t0` and `tmax`. frame_dt : float The duration of a frame in the visualization. .. note:: This has to be given in the original time units of the temporal network, disregarding any value of ``time_normalization_factor``. time_normalization_factor : float, default : 1.0 Rescale time with this factor. time_unit : string, default : None, Unit of time of the visualization. titles : string or list of strings, default : None Titles to put on the figures of the corresponding temporal networks. config : dict or str Configuration values for the JavaScript visualization. If this is a string, it can be either ``hs13``, ``dtu``, or ``ht09`` and the appropriate configuration is loaded. port : int, default : 8226 Port of the started HTTP server. export_path : string, default : None path to a directory to which the whole visualization is copied. Use ``os.get_cwd()+'/export_dir/'`` for the current working directory (after ``import os``). .. warning:: No subdirectory will be made for the export. All visualization files will be exported to ``export_path`` directly. Notes ----- The configuration dictionary is filled with values to control the appearance of the visualizations. The standard configuration is .. code:: python config = { "plot_width" : 320 , "network_plot_height" : 250, "edges_plot_height" : 100, "padding" : 10, "start_it" : 0, "node_radius" : 2.5, "link_distance" : 10, "node_charge": -8, "edge_line_width" : 1, "font_size_in_px" : 14, "link_width" : 1, "d3_format_string": ".3f", } """ if not hasattr(temporal_networks, '__len__'): temporal_networks = [temporal_networks] if titles is None: titles = ["" for _ in temporal_networks] elif type(titles) == str or not hasattr(titles, '__len__'): titles = [titles] this_config = copy.deepcopy(standard_config) if isinstance(config, str): if config == 'dtu': config = dict(dtu_config) elif config == 'hs13': config = dict(hs13_config) elif config == 'ht09': config = dict(ht09_config) else: raise ValueError("config", config, "is unknown.") if config is not None: this_config.update(config) # print(titles) # define the server address # server_address = ('', port) path = '~/.tacoma/web/' web_dir = os.path.abspath(os.path.expanduser(path)) # download d3 if that did not happen yet download_d3() # copy the html and js files for the visualizations prepare_visualization_directory() # create a subfolder based on the current time subdir = "tmp_{:x}".format(int(time.time()*1000)) mkdirp_customdir(directory=web_dir) subdir_path = os.path.join(web_dir, subdir) mkdirp_customdir(directory=subdir_path) # in case an export is demanded, prepare the export directory if export_path is not None: export_path = os.path.abspath(os.path.expanduser(export_path)) prepare_export_directory(export_path, subdir) # change directory to this directory print("changing directory to", web_dir) print("starting server here ...", web_dir) cwd = os.getcwd() os.chdir(web_dir) server = StoppableHTTPServer(("", port), http.server.SimpleHTTPRequestHandler, subdir_path, ) for itn, tn in enumerate(temporal_networks): print("preparing network", titles[itn]) tn_b = _get_prepared_network( tn, frame_dt, time_unit, time_normalization_factor) taco_fname = os.path.join(subdir, subdir+'_'+str(itn)+'.taco') edge_fname = os.path.join(subdir, subdir+'_'+str(itn)+'.json') tc.write_edge_trajectory_coordinates(tn_b, os.path.join(web_dir, edge_fname), filter_for_duration=frame_dt * time_normalization_factor) tc.write_json_taco(tn_b, os.path.join(web_dir, taco_fname)) this_config['temporal_network_files'].append(taco_fname) this_config['edges_coordinate_files'].append(edge_fname) this_config['titles'].append(titles[itn]) with open(os.path.join(web_dir, subdir+'_config.json'), 'w') as f: json.dump(this_config, f) if export_path is not None: copy_tree(subdir_path, os.path.join(export_path, subdir)) with open(os.path.join(export_path, 'default_config.json'), 'w') as f: json.dump(this_config, f) # ========= start server ============ thread = threading.Thread(None, thread.start()"http://localhost:"+str(port)+"/?data=" + subdir) try: while True: time.sleep(2) except KeyboardInterrupt: # thread.join() print('stopping server ...') server.stop_this() thread.join() # time.sleep(1) print('changing directory back to', cwd) os.chdir(cwd)
if __name__ == "__main__": # download_d3() a = tc.load_json_taco("~/.tacoma/ht09.taco") visualize(a, frame_dt=20, titles='HT09', time_unit='h', time_normalization_factor=1./3600.)