from django.shortcuts import render from common.models import * from alert_manager.StrategyBuilder import StrategyBuilder import socket import majordome from django.contrib.auth.decorators import login_required import alert_manager.tasks import os, time, shutil TEST_FILE = "unittest_voevent.xml" TEST_FILE_PATH = os.path.join(alert_manager.tasks.VOEVENTS_PATH, TEST_FILE) VOEVENTS_TO_SEND_PATH = "alert_manager/events_to_send" @login_required def start_simulation(request): ''' Called when the alert simulation button is pressed Deletes then copies the test voevent file from the events_received folder ''' if os.path.isfile(TEST_FILE_PATH): os.remove(TEST_FILE_PATH) print("================== DELETE FILE ==================") time.sleep(3) print("================== COPY FILE ==================") shutil.copyfile(os.path.join(VOEVENTS_TO_SEND_PATH, TEST_FILE), TEST_FILE_PATH) alerts = Alert.objects.order_by("-request__created") if StrategyObs.objects.filter(is_default=True).exists(): current_default = StrategyObs.objects.filter(is_default=True)[0].name btn_type = "btn-primary" else: current_default = "Warning: please choose a default strategy" btn_type = "btn-danger" strat_list = StrategyObs.objects.all() success = True message = "Simulation started successfully" return render(request, "alert_manager/alerts.html", locals()) @login_required def alerts_list(request): ''' Display the list of all alerts ''' alerts = Alert.objects.order_by("-request__created") if StrategyObs.objects.filter(is_default=True).exists(): current_default = StrategyObs.objects.filter(is_default=True)[0].name btn_type = "btn-primary" else: current_default = "Warning: please choose a default strategy" btn_type = "btn-danger" strat_list = StrategyObs.objects.all() return render(request, "alert_manager/alerts.html", locals()) @login_required def change_default_strategy(request, strat_id): ''' Changes the strategy to be used when an alert is received ''' strategies = StrategyObs.objects.all() if strategies.filter(id=strat_id).exists(): for strat in strategies: strat.is_default = False strat.save() strat = strategies.get(id=strat_id) strat.is_default = True strat.save() success = True message = "Default strategy successfully changed" else: error = True message = "Could not find this strategy" alerts = Alert.objects.order_by("-request__created") if StrategyObs.objects.filter(is_default=True).exists(): current_default = StrategyObs.objects.filter(is_default=True)[0].name btn_type = "btn-primary" else: current_default = "Warning: please choose a default strategy" btn_type = "btn-danger" strat_list = StrategyObs.objects.all() return render(request, "alert_manager/alerts.html", locals()) @login_required def change_obs_strategy(request, alert_id): ''' Open a page for a given alert, listing all the other strategies ''' alert = Alert.objects.get(id=alert_id) if alert.strategyobs is not None: strategies = StrategyObs.objects.exclude(id=alert.strategyobs.id) else: strategies = StrategyObs.objects.all() return render(request, "alert_manager/strategy_change.html", locals()) @login_required def change_obs_strategy_validate(request, alert_id): ''' Creates a new request with the strategy given for the selected alert ''' try: alert = Alert.objects.get(id=alert_id) except Exception: alerts = Alert.objects.all() error = True message = "This alert doesn't exist" return render(request, "alert_manager/alerts.html", locals()) strategies = StrategyObs.objects.exclude(id=alert.strategyobs.id) if request.method == "POST": strategy_id = request.POST["strategy_choice"] try: strategy = StrategyObs.objects.get(id=strategy_id) except Exception: error = True message = "This strategy doesn't exist" return render(request, "alert_manager/strategy_change.html", locals()) if str(alert.strategyobs.id) == strategy_id: error = True message = "This is already the current strategy for this alert" return render(request, "alert_manager/strategy_change.html", locals()) sb = StrategyBuilder() file = strategy.xml_file pyros_user = alert.request.pyros_user scientific_program = alert.request.scientific_program name = alert.request.name sb.create_request_from_strategy(file, pyros_user, scientific_program, name) req = sb.validate() alert.pk = None alert.strategyobs = strategy alert.request_id = req alert.__dict__.update(req.__dict__) alert.save() strategies = StrategyObs.objects.exclude(id=alert.strategyobs.id) success = True message = "Strategy successfully changed. A new request was created." return render(request, "alert_manager/strategy_change.html", locals())