Source code for festivalgrid.views

import os
import pty
import json
import pathlib
import requests
import subprocess

from django.shortcuts import render, redirect
from django.contrib.auth.decorators import login_required
from django.conf import settings
from django.http import HttpResponse, StreamingHttpResponse
from django.views.decorators.clickjacking import xframe_options_exempt
from django.template.loader import render_to_string

from ansi2html import Ansi2HTMLConverter
from grafana_api.grafana_face import GrafanaFace

from .models import Node

[docs]def subprocess_generator(command): yield render_to_string('festivalgrid/console_start.html', {}) (master, slave) = pty.openpty() subprocess.Popen(command, shell=True, stdin=slave, stdout=slave, stderr=slave, close_fds=True) os.close(slave) a2h = Ansi2HTMLConverter(scheme="osx", inline=True) rest = "" while True: try: data =, 1024) except OSError: break if not data: break try: d = data.decode() except: d = "" index = d.rfind("\n") if index < 0: rest += d yield "" else: lines = rest + d[:index] b = a2h.convert(lines, full=False) rest = d[index:] yield b yield render_to_string('festivalgrid/console_end.html', {})
[docs]def grafana(request, template_name, title, context): context['uid'] = 'festivalgrid' context['panel_title'] = title html = render_to_string('festivalgrid/grafana/%s.html' % template_name, context) model_string = html[html.find('\n') + 1:html.rfind('\n')] # remove first and last lines model = json.loads(model_string) grafana_api = GrafanaFace(auth=settings.GRAFANA_TOKEN, host="grafana:3000") grafana_api.dashboard.update_dashboard(dashboard={'dashboard': model, 'folderId': 0, 'overwrite': True}) return "http%s://%s:%s/d-solo/festivalgrid/%s?orgId=1&refresh=5s&panelId=1&fullscreen&from=now-15m&to=now&theme=dark" % ("s" if request.is_secure() else "", settings.GRAFANA_HOST, settings.GRAFANA_PORT, template_name)
[docs]@login_required def goldenlayout(request, layout): return render(request, 'festivalgrid/layout.html', { 'layout': layout, 'ip4net': settings.DEVICES_SUBNET })
[docs]@login_required def tilecache(request): url = request.path.replace("/tilecache/", "").replace("https://", "").strip("/") filepath = os.path.join(settings.BASE_DIR, "../web/tilecache/" + url) try: f = open(filepath, "rb") file = except FileNotFoundError: pathlib.Path(os.path.dirname(filepath)).mkdir(parents=True, exist_ok=True) headers = {'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_10_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/39.0.2171.95 Safari/537.36'} r = requests.get("https://" + url, headers=headers) file = r.content with open(filepath, 'wb') as f: f.write(file) return HttpResponse(file, content_type="image")
[docs]@login_required def update_devices(request): ip4range = request.GET.get('ip4range') timeout = int(request.GET.get('timeout', 10)) return StreamingHttpResponse(subprocess_generator('python /app/django/ update_devices %s --timeout %d' % (ip4range, timeout)))
[docs]@login_required def provision_devices(request): devices = request.GET.get('devices') return StreamingHttpResponse(subprocess_generator('python /app/django/ provision_devices %s' % devices))
[docs]@login_required @xframe_options_exempt def embed_grafana(request): targets = [] for node in Node.objects.all(): if hasattr(node, 'device') and node.device is not None: targets.append({'measurement':, 'alias':}) try: url = grafana(request, 'overview', 'Overview', {'targets': targets, 'style': "default", 'legend_right_side': "false"}) except (requests.exceptions.ConnectTimeout, requests.exceptions.ConnectionError) as e: return HttpResponse("<h3>Error connection to Grafana server at %s:%s</h3><p>Make sure Grafana is running and connection details are properly configured in .env file." % (settings.GRAFANA_HOST, settings.GRAFANA_PORT)) return redirect(url)