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 = os.read(master, 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 = f.read()
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/manage.py 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/manage.py 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': node.pk, 'alias': node.name})
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)
