mast/tools/processor.py
def process_input_data(self, input_file: str) -> list:
        """
        Purpose of this method is to iterate over each .xml file
        within landing layer, load its content and fetch all
        coordinates in order to calculate total distance in Km.
        """

        total_distance = 0
        activity_duration = None
        activity_start = None
        try:
            input_file = os.path.join(self.LANDING_DIR, input_file)
            if Path(input_file).suffix in self.__ALLOWED_EXTENSIONS:
                with open(os.path.join(self.LANDING_DIR, input_file), mode='rb') as xml_inp:
                    tree = ET.parse(xml_inp)
                    root = tree.getroot()

                    # Find the child element of tracking point including __namespace
                    pom_elem = findall('{.*}.*', root[0].tag if len(root) > 1 else root[0].tag)
                    # Cut of namespace prefix
                    namespace = findall('{.*}', pom_elem[0])[0]
                    trk = root.find(namespace + self.__TRK_ELM)
                    # Find all trk segments
                    trk_seg = [seg for seg in trk.findall(namespace + self.__TRKSEG_ELM)]

                    # Create dict where key is the seg and its value is array of activity points
                    segments = {seg: seg.findall(namespace + self.__TRKPT_ELM) for seg in trk_seg}

                    track_time = True
                    try:
                        timestamp_start = segments[list(segments.keys())[0]][0].find(namespace + self.__TIME_ELM)
                        if timestamp_start is not None:
                            activity_start = parser.parse(timestamp_start.text)
                        else:
                            track_time = False
                    except Exception as e:
                        logging.info("No timestamp for activity start present.", e)
                        track_time = False

                    inter_segment = [None, None]
                    segments_distance = []
                    segments_time = []

                    # If there is no timestamp skip
                    for segment in segments.values():
                        if inter_segment[0] is not None:
                            inter_segment[1] = segment[0]
                            if track_time:
                                segments_time.append(self.__calculate_total_time(inter_segment, namespace))
                            segments_distance.append(self.__calculate_orthodromic_distance(inter_segment))

                        if track_time:
                            segments_time.append(self.__calculate_total_time(segment, namespace))
                        segments_distance.append(self.__calculate_orthodromic_distance(segment))

                        inter_segment[0] = segment[-1]
                    total_distance = round(sum(segments_distance), 1)
                    activity_duration = sum(segments_time, datetime.timedelta())

                    xml_inp.close()
            else:
                raise AssertionError("Forbidden file extension encountered!")
        except Exception as ex:
            logging.error("Processing of the landing directory was unsuccessful!\n", ex)
        return [total_distance, activity_duration, activity_start]

        Purpose of this method is to iterate over each .xml file
        within landing layer, load its content and fetch all
        coordinates in order to calculate total distance in Km.
        
Purpose of this method is to iterate over each .xml file
within landing layer, load its content and fetch all
coordinates in order to calculate total distance in Km.
Purpose of this method is to iterate over each . xml file within landing layer load its content and fetch all coordinates in order to calculate total distance in Km .
def process_input_data(self, input_file: str) -> list:
        total_distance = 0
        activity_duration = None
        activity_start = None
        try:
            input_file = os.path.join(self.LANDING_DIR, input_file)
            if Path(input_file).suffix in self.__ALLOWED_EXTENSIONS:
                with open(os.path.join(self.LANDING_DIR, input_file), mode='rb') as xml_inp:
                    tree = ET.parse(xml_inp)
                    root = tree.getroot()
                    pom_elem = findall('{.*}.*', root[0].tag if len(root) > 1 else root[0].tag)
                    namespace = findall('{.*}', pom_elem[0])[0]
                    trk = root.find(namespace + self.__TRK_ELM)
                    trk_seg = [seg for seg in trk.findall(namespace + self.__TRKSEG_ELM)]
                    segments = {seg: seg.findall(namespace + self.__TRKPT_ELM) for seg in trk_seg}
                    track_time = True
                    try:
                        timestamp_start = segments[list(segments.keys())[0]][0].find(namespace + self.__TIME_ELM)
                        if timestamp_start is not None:
                            activity_start = parser.parse(timestamp_start.text)
                        else:
                            track_time = False
                    except Exception as e:
                        logging.info("No timestamp for activity start present.", e)
                        track_time = False
                    inter_segment = [None, None]
                    segments_distance = []
                    segments_time = []
                    for segment in segments.values():
                        if inter_segment[0] is not None:
                            inter_segment[1] = segment[0]
                            if track_time:
                                segments_time.append(self.__calculate_total_time(inter_segment, namespace))
                            segments_distance.append(self.__calculate_orthodromic_distance(inter_segment))
                        if track_time:
                            segments_time.append(self.__calculate_total_time(segment, namespace))
                        segments_distance.append(self.__calculate_orthodromic_distance(segment))
                        inter_segment[0] = segment[-1]
                    total_distance = round(sum(segments_distance), 1)
                    activity_duration = sum(segments_time, datetime.timedelta())
                    xml_inp.close()
            else:
                raise AssertionError("Forbidden file extension encountered!")
        except Exception as ex:
            logging.error("Processing of the landing directory was unsuccessful!\n", ex)
        return [total_distance, activity_duration, activity_start]
def process_input_data ( self , input_file : str ) -> list : total_distance = 0 activity_duration = None activity_start = None try : input_file = os . path . join ( self . LANDING_DIR , input_file ) if Path ( input_file ) . suffix in self . __ALLOWED_EXTENSIONS : with open ( os . path . join ( self . LANDING_DIR , input_file ) , mode = 'rb' ) as xml_inp : tree = ET . parse ( xml_inp ) root = tree . getroot ( ) pom_elem = findall ( '{.*}.*' , root [ 0 ] . tag if len ( root ) > 1 else root [ 0 ] . tag ) namespace = findall ( '{.*}' , pom_elem [ 0 ] ) [ 0 ] trk = root . find ( namespace + self . __TRK_ELM ) trk_seg = [ seg for seg in trk . findall ( namespace + self . __TRKSEG_ELM ) ] segments = { seg : seg . findall ( namespace + self . __TRKPT_ELM ) for seg in trk_seg } track_time = True try : timestamp_start = segments [ list ( segments . keys ( ) ) [ 0 ] ] [ 0 ] . find ( namespace + self . __TIME_ELM ) if timestamp_start is not None : activity_start = parser . parse ( timestamp_start . text ) else : track_time = False except Exception as e : logging . info ( "No timestamp for activity start present." , e ) track_time = False inter_segment = [ None , None ] segments_distance = [ ] segments_time = [ ] for segment in segments . values ( ) : if inter_segment [ 0 ] is not None : inter_segment [ 1 ] = segment [ 0 ] if track_time : segments_time . append ( self . __calculate_total_time ( inter_segment , namespace ) ) segments_distance . append ( self . __calculate_orthodromic_distance ( inter_segment ) ) if track_time : segments_time . append ( self . __calculate_total_time ( segment , namespace ) ) segments_distance . append ( self . __calculate_orthodromic_distance ( segment ) ) inter_segment [ 0 ] = segment [ - 1 ] total_distance = round ( sum ( segments_distance ) , 1 ) activity_duration = sum ( segments_time , datetime . timedelta ( ) ) xml_inp . close ( ) else : raise AssertionError ( "Forbidden file extension encountered!" ) except Exception as ex : logging . error ( "Processing of the landing directory was unsuccessful!\n" , ex ) return [ total_distance , activity_duration , activity_start ]
Purpose of this method is to iterate over each .xml file
within landing layer, load its content and fetch all
coordinates in order to calculate total distance in Km.
Purpose of this method is to iterate over each . xml file within landing layer load its content and fetch all coordinates in order to calculate total distance in Km .
"""
        Purpose of this method is to iterate over each .xml file
        within landing layer, load its content and fetch all
        coordinates in order to calculate total distance in Km.
        """ # Find the child element of tracking point including __namespace # Cut of namespace prefix # Find all trk segments # Create dict where key is the seg and its value is array of activity points # If there is no timestamp skip
self
input_file
str