added task import from file functions

airscore/core/frontendUtils.py

@@ -2659,3 +2659,24 @@ def recheck_needed(task_id: int):
     from task import task_need_recheck
     return task_need_recheck(task_id)
 
+
+def get_task_from_file(task_id: int, file) -> bool:
+    """reads .xctsk and .tsk files, creates a Task object and adds it to the db"""
+    from sources import xctrack, taskplanner
+    from task import Task, write_map_json
+
+    ext = file.filename
+    if 'xctsk' in ext:
+        # xctrack file
+        data = xctrack.read_xctsk_file(file)
+        task_info = xctrack.read_task(data)
+    elif 'tsk' in ext:
+        # task planner file
+        data = taskplanner.read_tsk_file(file)
+        task_info = taskplanner.read_task(data)
+    else:
+        return False
+
+    task = Task.update_from_dict(task_id, task_info)
+    return True
+

airscore/core/sources/taskplanner.py

@@ -0,0 +1,77 @@
+"""
+Task Planner Library
+
+contains methods to import Task from Tonino Tarsi's TasK Creator .tsk file
+https://www.vololiberomontecucco.it/taskcreator/
+
+Use: from sources.taskplanner import TaskPlanner
+
+Antonio Golfari - 2024
+"""
+
+from . import utils
+from lxml import etree
+from pathlib import Path
+
+
+def read_tsk_file(file: Path) -> "etree | None":
+
+    return utils.read_xml_file(file)
+
+
+def read_task(root: etree) -> dict:
+    """creates a dict file with all task information to be imported from Airscore"""
+
+    task_info = {}
+
+    if len(root):
+        rte = root.find('rte')
+        # task info
+        task_info['task_type'] = rte.find('type').text  # 'race', 'elapsed_time'
+        # gates info
+        gates = int(rte.find('ngates').text)
+        if gates > 1:
+            task_info['start_iteration'] = gates - 1
+            task_info['SS_interval'] = int(rte.find('gateint').text) * 60  # in seconds
+        # comment
+        if rte.find('info').text:
+            task_info['comment'] = rte.find('info').text
+        # wpt info
+        task_info['route'] = []
+        for node in rte.iter('rtept'):
+            wpt = dict(
+                num=int(node.find('index').text),
+                name=node.find('id').text,
+                description=node.find('name').text,
+                lat=float(node.get('lat')), lon=float(node.get('lon')),
+                altitude=None if node.find('z') is None else int(node.find('z').text),
+                radius=int(node.find('radius').text),
+                shape='circle',
+                how='entry'
+            )
+            t = node.find('type').text.lower()
+            if t == 'takeoff':
+                wpt['type'] = 'launch'
+                wpt['how'] = 'exit'
+                task_info['window_open_time'] = utils.get_time(node.find('open').text)
+                task_info['window_close_time'] = utils.get_time(node.find('close').text)
+            elif t == 'start':
+                wpt['type'] = 'speed'
+                task_info['start_time'] = utils.get_time(node.find('open').text)
+                task_info['start_close_time'] = (
+                    task_info['start_time'] + 3600 if not node.find('close')
+                    else utils.get_time(node.find('close').text)
+                )
+            elif t == 'end-of-speed-section':
+                wpt['type'] = 'endspeed'
+            elif t == 'goal':
+                wpt['type'] = 'goal'
+                if node.find('goalType').text == 'line':
+                    wpt['shape'] = 'line'
+                task_info['task_deadline'] = utils.get_time(node.find('close').text)
+            else:
+                wpt['type'] = 'waypoint'
+
+            task_info['route'].append(wpt)
+
+    return task_info

airscore/core/sources/utils.py

@@ -0,0 +1,39 @@
+"""
+Source utilities functions
+
+Antonio Golfari - 2024
+"""
+
+from lxml import etree
+from pathlib import Path
+
+def read_xml_file(file: Path, clean_namespace: bool = True) -> "etree | None":
+    """read the xml file"""
+    try:
+        tree = etree.parse(file)
+    except TypeError:
+        tree = etree.parse(file.as_posix())
+    except etree.Error as e:
+        print(f"XML Read Error: {e}")
+        return None
+    finally:
+        root = tree.getroot()
+        if clean_namespace:
+            clean_xml_namespaces(root)
+        return root
+
+def clean_xml_namespaces(root):
+    for element in root.getiterator():
+        if isinstance(element, etree._Comment):
+            continue
+        element.tag = etree.QName(element).localname
+    etree.cleanup_namespaces(root)
+
+
+def get_time(string: str) -> int:
+    print(string)
+    if len(string) < 3:
+        # sometimes this incredibly happens
+        string += ':00'
+    h, m = string.replace(';', ':').split(':')[:2]
+    return int(h) * 3600 + int(m[:2]) * 60

airscore/core/sources/xctrack.py

@@ -0,0 +1,75 @@
+"""
+XC Track Library
+
+contains methods to import Task from XcTrack .xctsk file
+
+Use: import sources.xctrack
+
+Antonio Golfari - 2024
+"""
+
+import json
+
+from . import utils
+from pathlib import Path
+
+
+def read_xctsk_file(file) -> "dict | None":
+
+    try:
+        return json.load(file)
+    except:
+        print("xctsk file is not a valid JSON object")
+        return None
+
+
+def read_task(data: dict) -> dict:
+
+    task_info = {}
+    if data:
+        task_info['task_type'] = 'elapsed time' if data.get('taskType') == 'elapsed_time' else 'race'
+        print(f"len gates: {len(data['sss']['timeGates'])}")
+        if len(data['sss']['timeGates']):
+            task_info['start_time'] = utils.get_time(data['sss']['timeGates'][0])
+            print(f"start: {task_info['start_time']}")
+            # xctrack file does not have launch window info
+            task_info['start_close_time'] = task_info['start_time'] + 3600
+            task_info['window_open_time'] = task_info['start_time'] - 3600
+            task_info['window_close_time'] = task_info['start_time']
+            print(f"s close: {task_info['start_close_time']}")
+            print(f"w open: {task_info['window_open_time']}")
+            print(f"w close: {task_info['window_close_time']}")
+
+        task_info['task_deadline'] = utils.get_time(data['goal']['deadline'])
+        task_info['route'] = []
+        for idx, el in enumerate(data['turnpoints']):
+            w = el['waypoint']
+            wpt = dict(
+                num=idx,
+                name=w['description'],
+                description=w['name'],
+                lat=w['lat'], lon=w['lon'],
+                altitude=int(w['altSmoothed']),
+                radius=int(el['radius']),
+                shape='circle',
+                how='entry'
+            )
+            t = None if el.get('type') is None else el['type'].lower()
+            if t == 'takeoff':
+                wpt['type'] = 'launch'
+                wpt['how'] = 'exit'
+            elif t == 'sss':
+                wpt['type'] = 'speed'
+            elif t == 'ess':
+                wpt['type'] = 'endspeed'
+            elif idx == len(data['turnpoints']) - 1:
+                wpt['type'] = 'goal'
+                if data['goal']['type'].lower() == 'line':
+                    wpt['shape'] = 'line'
+            else:
+                wpt['type'] = 'waypoint'
+
+            task_info['route'].append(wpt)
+
+    return task_info
+

airscore/core/task.py

@@ -950,142 +950,44 @@ def to_db(self):
             if self.turnpoints:
                 self.update_waypoints()
 
-    def update_from_xctrack_data(self, taskfile_data):
-        """processes XCTrack file that is already in memory as json data and updates the task defintion"""
-        from calcUtils import string_to_seconds
-        from compUtils import get_wpts
-
-        startopenzulu = taskfile_data['sss']['timeGates'][0]
-        deadlinezulu = taskfile_data['goal']['deadline']
-
-        self.start_time = string_to_seconds(startopenzulu)
-        self.task_deadline = string_to_seconds(deadlinezulu)
-
-        '''check task start and start close times are ok for new start time
-        we will check to be at least 1 hour before and after'''
-        if not self.window_open_time or self.start_time - self.window_open_time < 3600:
-            self.window_open_time = self.start_time - 3600
-        if not self.start_close_time or self.start_close_time - self.start_time < 3600:
-            self.start_close_time = self.start_time + 3600
-
-        if taskfile_data['sss']['type'] == 'ELAPSED-TIME':
-            self.task_type = 'elapsed time'
-        else:
-            self.task_type = 'race'
-            '''manage multi start'''
-            self.SS_interval = 0
-            if len(taskfile_data['sss']['timeGates']) > 1:
-                second_start = string_to_seconds(taskfile_data['sss']['timeGates'][1])
-                self.SS_interval = int((second_start - self.start_time) / 60)  # interval in minutes
-                self.start_close_time = (
-                    int(self.start_time + len(taskfile_data['sss']['timeGates']) * (second_start - self.start_time)) - 1
-                )
+    @staticmethod
+    def update_from_dict(task_id: int, data: dict):
+        """ Creates Task from dict
+            usually results from file imports"""
 
-        print('xct start:       {} '.format(self.start_time))
-        print('xct deadline:    {} '.format(self.task_deadline))
-
-        waypoint_list = get_wpts(self.id)
-        print('n. waypoints: {}'.format(len(taskfile_data['turnpoints'])))
-
-        for i, tp in enumerate(taskfile_data['turnpoints']):
-            waytype = "waypoint"
-            shape = "circle"
-            how = "entry"  # default entry .. looks like xctrack doesn't support exit cylinders apart from SSS
-            wpID = waypoint_list.get(tp["waypoint"]["name"]) or None
-            wpNum = i + 1
-
-            if i < len(taskfile_data['turnpoints']) - 1:
-                if 'type' in tp:
-                    if tp['type'] == 'TAKEOFF':
-                        waytype = "launch"  # live
-                        # waytype = "start"  # aws
-                        how = "exit"
-                    elif tp['type'] == 'SSS':
-                        waytype = "speed"
-                        if taskfile_data['sss']['direction'] == "EXIT":  # get the direction form the SSS section
-                            how = "exit"
-                    elif tp['type'] == 'ESS':
-                        waytype = "endspeed"
-            else:
-                waytype = "goal"
-                if taskfile_data['goal']['type'] == 'LINE':
-                    shape = "line"
-
-            turnpoint = Turnpoint(tp['waypoint']['lat'], tp['waypoint']['lon'], tp['radius'], waytype, shape, how)
-            turnpoint.name = tp["waypoint"]["name"]
-            turnpoint.rwp_id = wpID
-            turnpoint.num = wpNum
-            self.turnpoints.append(turnpoint)
-
-    def update_from_xctrack_file(self, filename):
-        """Updates Task from xctrack file, which is in json format."""
-        with open(filename, encoding='utf-8') as json_data:
-            # a bit more checking..
-            print("file: ", filename)
-            try:
-                task_data = json.load(json_data)
-            except:
-                print("file is not a valid JSON object")
-                exit()
-        self.update_from_xctrack_data(task_data)
+        task = Task.read(task_id=task_id)
 
-    @staticmethod
-    def create_from_xctrack_file(filename):
-        """Creates Task from xctrack file, which is in json format.
-        NEEDS UPDATING BUT WE CAN PROBABLY REMOVE THIS AS THE TASK SHOULD ALWAYS BE CREATED BEFORE IMPORT??
-        """
-        offset = 0
-        task_file = filename
-        turnpoints = []
-        with open(task_file, encoding='utf-8') as json_data:
-            # a bit more checking..
-            print("file: ", task_file)
-            try:
-                t = json.load(json_data)
-            except:
-                print("file is not a valid JSON object")
-                exit()
-
-        startopenzulu = t['sss']['timeGates'][0]
-        deadlinezulu = t['goal']['deadline']
-        task_type = 'race' if t['sss']['type'].lower() == 'race' else 'elapsed time'
-
-        startzulu_split = startopenzulu.split(":")  # separate hours, minutes and seconds.
-        deadlinezulu_split = deadlinezulu.split(":")  # separate hours, minutes and seconds.
-
-        start_time = (int(startzulu_split[0]) + offset) * 3600 + int(startzulu_split[1]) * 60
-        task_deadline = (int(deadlinezulu_split[0]) + offset) * 3600 + int(deadlinezulu_split[1]) * 60
-
-        for tp in t['turnpoints'][:-1]:  # loop through all waypoints except last one which is always goal
-            waytype = "waypoint"
-            shape = "circle"
-            how = "entry"  # default entry .. looks like xctrack doesn't support exit cylinders apart from SSS
-
-            if 'type' in tp:
-                if tp['type'] == 'TAKEOFF':
-                    waytype = "launch"  # live
-                    # waytype = "start"  #aws
-                    how = "exit"
-                if tp['type'] == 'SSS':
-                    waytype = "speed"
-                    if t['sss']['direction'] == "EXIT":  # get the direction form the SSS section
-                        how = "exit"
-                if tp['type'] == 'ESS':
-                    waytype = "endspeed"
-            turnpoint = Turnpoint(tp['waypoint']['lat'], tp['waypoint']['lon'], tp['radius'], waytype, shape, how)
-            turnpoints.append(turnpoint)
+        ''' get task info'''
+        for key, value in data.items():
+            if hasattr(task, key):
+                if task.time_offset != 0 and any(s in key for s in ('_time', '_deadline')):
+                    value -= task.time_offset
+                setattr(task, key, value)
 
-        # goal - last turnpoint
-        tp = t['turnpoints'][-1]
-        waytype = "goal"
-        if t['goal']['type'] == 'LINE':
-            shape = "line"
+        ''' get route'''
+        task.turnpoints = []
+        task.partial_distance = []
 
-        turnpoint = Turnpoint(tp['waypoint']['lat'], tp['waypoint']['lon'], tp['radius'], waytype, shape, how)
-        turnpoints.append(turnpoint)
+        for idx, tp in enumerate(data['route'], 1):
+            '''creating waypoints'''
+            # I could take them from database, but this is the only way to be sure it is the correct one
+            turnpoint = Turnpoint(tp['lat'], tp['lon'], tp['radius'], tp['type'], tp['shape'], tp['how'])
 
-        task = Task(turnpoints, start_time, task_deadline, task_type)
+            turnpoint.name = tp['name']
+            turnpoint.num = idx
+            turnpoint.description = tp['description']
+            turnpoint.altitude = tp['altitude']
+            task.turnpoints.append(turnpoint)
+
+        # calculate task distances
+        task.calculate_task_length()
         task.calculate_optimised_task_length()
+        # save to db
+        task.update_task_info()
+        task.to_db()
+
+        # update map
+        write_map_json(task_id)
 
         return task