Reach.py

# -*- coding: utf-8 -*-

import numpy as np

from time import time
from typing import List
from copy import deepcopy
from operator import itemgetter

from tools import flatten

from Model.Geometry.Profile import Profile
from Model.Geometry.ProfileXYZ import ProfileXYZ

from Model.Except import FileFormatError, exception_message_box

class Reach:
    def __init__(self, parent):
        self._parent = parent
        self._profiles: List[Profile] = []

        self._guidelines_is_valid = False
        self._guidelines = {}

        # Copy/Paste
        self.__list_copied_profiles = []

    def profile(self, i):
        """Returns profile at index i

        Args:
            i: The index of profile

        Returns:
            The profile at index i
        """
        if i in self._profiles:
            return self._profiles[i]

        return None

    @property
    def name(self):
        return self._parent.name

    @property
    def profiles(self):
        return self._profiles.copy()

    @property
    def number_profiles(self):
        """
        Returns:
            Number of profiles
        """
        return len(self._profiles)

    def get_geometry(self) -> List[Profile]:
        """
        Returns:
            The profiles list.
        """
        return self._profiles

    def get_profile_i(self, i: int) -> Profile:
        """
        Args:
            i: Index

        Returns:
            The profile at index i.
        """
        try:
            return self._profiles[i]
        except IndexError:
            raise IndexError(f"Invalid profile index: {i}")

    def add_profile(self, index):
        """Add a new profile at the end of profiles list

        Returns:
            Nothing.
        """
        nb_profile = self.number_profiles
        profile = ProfileXYZ()
        profile.num = nb_profile + 1

        self._profiles.insert(profile, index + 1)
        self._update_profile_numbers()

    def _update_profile_numbers(self):
        """Update profiles index

        Returns:
            Nothing.
        """
        for ind, profile in enumerate(self.get_geometry()):
            profile.num = ind + 1

    def insert(self, index: int):
        """Insert new profile in list

        Args:
            index: The position of the new profile.

        Returns:
            Nothing.
        """
        profile = ProfileXYZ()

        self._profiles.insert(index, profile)
        self._update_profile_numbers()

    def delete(self, list_index: list):
        """Delete some elements in profile list

        Args:
            list_index: The list of element index

        Returns:
            Nothing.
        """
        try:
            if list_index:
                indices = sorted(list(set(list_index)), reverse=True)
                for idx in indices:
                    try:
                        self._profiles.pop(idx)
                        self._update_profile_numbers()
                    except IndexError:
                        print(f"Invalid profile index: {idx}")
        except TypeError:
            if isinstance(list_index, int):
                self._profiles.pop(list_index)
                self._update_profile_numbers()
            else:
                raise TypeError(f"{list_index} is instance of unexpected type '{type(list_index)}'")

    def _sort(self, is_reversed: bool = False):
        self._profiles = sorted(
            self._profiles,
            key=lambda profile: profile.kp(),
            reverse=is_reversed
        )

    def get_x(self):
        return [profile.x() for profile in self.profiles]

    def get_y(self):
        return [profile.y() for profile in self.profiles]

    def get_z(self):
        return [profile.z() for profile in self.profiles]

    def get_z_min(self):
        """List of z min for each profile

        Returns:
            List of z min for each profile
        """
        return [profile.z_min() for profile in self._data.profiles]

    def get_z_max(self):
        """List of z max for each profile

        Returns:
            List of z max for each profile
        """
        return [profile.z_max() for profile in self._data.profiles]

    def get_kp(self):
        """List of profiles kp

        Returns:
            List of profiles kp
        """
        return [profile.kp for profile in self._data.profiles]

    ##############
    # GUIDELINES #
    ##############

    def _compute_guidelines_cache(self, guide_set, named_points):
        # Reset guide lines cache
        self._guidelines = {}

        # Make a list of point for each guideline
        for guide in guide_set:
            self._guidelines[guide] = flatten(
                map(
                    lambda l: list(
                        # Filter point with name (we assume we have
                        # only one point by profile)
                        filter(
                            lambda p: p.name == guide,
                            l
                        )
                    ),
                    named_points
                )
            )

    def compute_guidelines(self):
        """Compute reach guideline and check if is valid for all profiles

        Returns:
            True if all guide line is valid
        """
        # Get all point contains into a guideline
        named_points = [profile.named_points() for profile in self._profiles]
        points_name = list(
            map(
                lambda lst: list(map(lambda p: p.name, lst)),
                named_points
            )
        )

        # Get all guide line name
        guide_set = reduce(
            lambda acc, x: set(x).union(acc),
            points_name
        )

        # All guide line is valid
        is_ok = reduce(
            bool.__and__,
            map(
                lambda l: len(set(l).symmetric_difference(guide_set)) == 0,
                points_name
            )
        )
        self._guidelines_is_valid = is_ok

        # Compute guideline and put data in cache
        self._compute_guidelines_cache(guide_set, named_points)

        return is_ok

    def _map_guidelines_points(self, func):
        return list(
            # Map for each guideline
            map(
                lambda k: list(
                    # Apply function FUNC on each point of guideline
                    map(
                        func,
                        self._guidelines[k],
                    )
                ),
                self._guidelines
            )
        )

    def get_guidelines_x(self):
        return self._map_guidelines_points(lambda p: p.x)

    def get_guidelines_y(self):
        return self._map_guidelines_points(lambda p: p.y)

    def get_guidelines_z(self):
        return self._map_guidelines_points(lambda p: p.z)

    # Sort

    def sort_ascending(self):
        """Sort profiles by increasing KP

        Returns:
            Nothing.
        """
        self._sort(is_reversed=False)

    def sort_descending(self):
        """Sort profiles by decreasing KP

        Returns:
            Nothing.
        """
        self._sort(is_reversed=True)

    def copy(self, index_list: List[int]):
        self.__list_copied_profiles.clear()
        index_list = list(set(index_list))  # delete duplicate index

        for index in index_list:
            try:
                self.__list_copied_profiles.append(deepcopy(self.get_profile_i(index)))
            except IndexError:
                raise IndexError(f"Invalid profile index: {index}")

    def paste(self):
        if self.__list_copied_profiles:
            for profile in self.__list_copied_profiles:
                self._profiles.append(profile)

    def import_geometry(self, file_path_name: str):
        """Import a geometry from file (.ST or .st)

        Args:
            file_path_name: The absolute path of geometry file (.ST or .st) to import.

        Returns:
            Nothing.
        """
        try:
            list_profile, list_header = self.read_file_st(str(file_path_name))

            if list_profile and list_header:
                for ind, profile in enumerate(list_profile):
                    prof = ProfileXYZ(*list_header[ind])
                    prof.import_points(profile)
                    self._profiles.append(prof)
                    self._update_profile_numbers()
        except FileNotFoundError as e:
            print(e)
            exception_message_box(e)
        except FileFormatError as e:
            print(e)
            e.alert()

    def read_file_st(self, filename):
        """Read the ST file

        Returns:
            List of profiles and list of headers.
        """
        t0 = time()
        line_is_header = True
        list_point_profile = []
        list_profile = []
        list_header = []
        stop_code = "999.999"

        with open(filename, encoding="utf-8") as file_st:
            for line in file_st:
                if not (line.startswith("#") or line.startswith("*")):
                    line = line.split()

                    if line_is_header:
                        if len(line) >= 6:
                            list_header.append(line[:6])
                        elif len(line) == 5:
                            line.append("")
                            list_header.append(line)
                        else:
                            print(f"Point {line} invalide ==> pas pris en compte")
                            line_is_header = False
                    else:
                        # Read until "999.9990     999.9990" as found
                        if len(line) == 3:
                            x, y, z = line
                            if stop_code in x and stop_code in y:
                                line_is_header = True
                                list_profile.append(list_point_profile)
                                list_point_profile = []
                            else:
                                line.append("")
                                list_point_profile.append(line)
                        elif len(line) == 4:
                            x, y, z, ld = line
                            if stop_code in x and stop_code in y:
                                list_profile.append(list_point_profile)
                                list_point_profile = []
                                line_is_header = True
                            else:
                                list_point_profile.append(line)
                        else:
                            pass

        if list_profile and list_header:
            raise FileFormatError(filename, f"{list_profile}, {list_header}")

        print("******  Fichier {} lu et traité en {} secondes *******".format(filename, time() - t0))
        return list_profile, list_header

    # TODO: Move this function to model reach
    def export_reach(self, filename):
        with open(f"{filename}", "w") as file_st:
            for index in range(len(self._profiles)):
                profile = self._profiles[index]

                for v in profile.header:
                    file_st.write(f"{v:>6}")
                file_st.write("\n")

                for point in self._data.profile[index_pro].points:
                    for i in [point.x, point.y, point.z, point.name]:
                        file_st.write(f"{i:>13.4f}")
                        file_st.write("\n")

                file_st.write("    999.9990     999.9990     999.9990")
                file_st.write("\n")