import ast from src.core.pyros_django.obsconfig.obsconfig_class import OBSConfig from .forms import SequenceForm, AlbumForm, PlanForm from common.models import PyrosUser, ScientificProgram, Sequence, Album, Plan, Period import datetime import os, yaml def check_sequence_file_validity(yaml_content,request): sp_of_user = request.user.get_scientific_program() sp_list = ScientificProgram.objects.observable_programs().filter(id__in=sp_of_user) seq = Sequence.objects.create() seq.pyros_user = PyrosUser.objects.get(id=request.user.id) unit_name = os.environ["unit_name"] config = OBSConfig(os.environ["PATH_TO_OBSCONF_FILE"], unit_name) sequence_form = SequenceForm(instance=seq, data_from_config=config.getEditableAttributesOfMount(config.unit_name), layouts = config.get_layouts(config.unit_name), sp_list=sp_list) result = { "succeed": True, "errors": [], } index_value_sp = yaml_content["sequence"]["scientific_program"]["value"] values = yaml_content["sequence"]["scientific_program"]["values"] if index_value_sp < 0 or index_value_sp > len(yaml_content["sequence"]["scientific_program"]["values"]): result["errors"].append(f"Value of scientific program isn't valid, index out of bounds ({index_value_sp} > {len(values)})") index_value_sp = 0 chosen_sp = ScientificProgram.objects.get(name=yaml_content["sequence"]["scientific_program"]["values"][index_value_sp]) if chosen_sp in sp_list: seq.scientific_program = ScientificProgram.objects.get(name=yaml_content["sequence"]["scientific_program"]["values"][index_value_sp]) else: result["errors"].append(f"Scientific program {chosen_sp.name} is not assigned to that user ") seq.config_attributes = {} for field in sequence_form.fields.keys(): if sequence_form.fields[field].required == False or field == "scientific_program": continue value = yaml_content["sequence"][field]["value"] if field not in yaml_content["sequence"]: result["errors"].append(f"{field} not in yaml file") else: if yaml_content["sequence"][field].get("values"): index_value = yaml_content["sequence"][field]["value"] values = yaml_content["sequence"][field]["values"] if index_value < 0 or index_value > len(yaml_content["sequence"][field]["values"]): result["errors"].append(f"Value of {field} isn't valid, index out of bounds ({index_value} > {len(values)})") index_value = 0 value = yaml_content["sequence"][field]["values"][index_value] else: if field == "start_date": value = datetime.datetime.strptime(yaml_content["sequence"][field]["value"],'%d/%m/%Y %H:%M:%S') seq.__dict__[field] = value continue if field in seq.__dict__.keys(): seq.__dict__[field] = value else: seq.config_attributes[field] = value albums_from_file = yaml_content["sequence"]["ALBUMS"] choosen_layout = seq.config_attributes["layout"] albums_of_layout = config.getLayoutByName(unit_name=config.unit_name,name_of_layout=choosen_layout)["ALBUMS"] if len(albums_of_layout) == len(albums_from_file): for album in albums_from_file: album = album["Album"] if album["name"] not in albums_of_layout: result["errors"].append(f"Album {album['name']} is not the chosen layout") else: alb = Album.objects.create(name=album["name"], sequence=seq, complete=True) else: result["errors"].append(f"The number of albums doesn't correspond to the chosen layout") for album in yaml_content["sequence"]["ALBUMS"]: album = album["Album"] plans = album.get("Plans") if plans == None: result["errors"].append(f"Album {album['name']} has no plans. Please add at least one plan") else: for plan in plans: new_plan_object = Plan.objects.create(album=Album.objects.get(name=album["name"],sequence=seq),complete=True) new_plan_object.config_attributes = {} plan = plan["Plan"] config_attributes = {} plan_form = PlanForm(data_from_config=config.getEditableAttributesOfChannel(config.unit_name,list(config.get_channels(config.unit_name).keys())[0]),edited_plan=None) for field in plan_form.fields: min_value = None max_value = None value_type = None if field not in plan.keys(): result["errors"].append(f"Missing field : '{field}' for plan {plans.index(plan)}") else: # TODO : ajouter max_value, min_value, suppression plan et album si invalides if plan[field].get("value_type"): value_type = plan[field]["value_type"] if type(plan[field]["value"]) == str and ast.literal_eval(plan[field]["value"]) != value_type: result["errors"].append(f"Field {field} value doesn't correspond to the assigned type (type required : {value_type})") if plan[field].get("min_value"): if type(plan[field]["min_value"]) == str: min_value = ast.literal_eval(plan[field]["min_value"]) else: min_value = plan[field]["min_value"] if plan[field].get("max_value"): if type(plan[field].get("max_value")) == str: max_value = ast.literal_eval(plan[field]["max_value"]) else: max_value = plan[field]["max_value"] if field == "nb_images": new_plan_object.__dict__[field] = plan[field]["value"] else: if plan[field].get("values"): index_value = plan[field]["value"] values = plan[field]["values"] if index_value < 0 or index_value > len(plan[field]["values"]): result["errors"].append(f"Value of Plan field '{field}' isn't valid, index out of bounds ({index_value} > {len(values)})") index_value = 0 value = plan[field]["values"][index_value] try: # linked values splitted_values = value.split(";") config_attributes[field] = {} for splitted_value in splitted_values: subkey,subvalue = splitted_value.split(":") config_attributes[field][subkey] = ast.literal_eval(subvalue) except: # Do nothing, normal string config_attributes[field] = ast.literal_eval(value) new_plan_object.config_attributes[field] = config_attributes[field] else: if max_value and min_value: if plan[field]["value"] > max_value: result["errors"].append(f"Plan field {field} doesn't respect max value") if plan[field]["value"] < min_value: result["errors"].append(f"Plan field {field} doesn't respect min value") new_plan_object.config_attributes[field] = plan[field]["value"] new_plan_object.save() seq.status = Sequence.TOBEPLANNED seq.complete = True period = Period.objects.exploitation_period() if Period.objects.next_period() != None and Period.objects.next_period().start_date < seq.start_date.date(): period = Period.objects.next_period() seq.period = period seq.save() if len(result["errors"]) != 0: result["succeed"] = False seq.delete() else: result["sequence_id"] = seq.id return result