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 functools import reduce

from tools import flatten, timer, trace

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, status=None, parent=None):
        self._status = status
        self._parent = parent
        self._profiles: List[Profile] = []

        self._guidelines_is_valid = False
        self._guidelines = {}

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

        Args:
            i: The index of profile

        Returns:
            The profile at index i
        """
        if i < len(self._profiles):
            return self._profiles[i]

        return None

    @property
    def name(self):
        if self._parent == None:
            return ""

        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 _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:
            The new profile.
        """
        profile = ProfileXYZ(reach=self, status=self._status)

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

        self._status.modified()

        return profile

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

        Args:
            index: The position of the new profile.

        Returns:
            Nothing.
        """
        self._profiles.insert(index, profile)
        self._update_profile_numbers()
        self._status.modified()


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

        Args:
            indexes: The list of index to delete

        Returns:
            Nothing.
        """
        profiles = set(
            map(
                lambda e: e[1],
                filter(
                    lambda e: e[0] in indexes,
                    enumerate(self.profiles)
                )
            )
        )

        self._profiles = list(
            filter(
                lambda p: p not in profiles,
                self.profiles
            )
        )
        self._update_profile_numbers()
        self._status.modified()

    def delete_profiles(self, profiles):
        """Delete some elements in profile list

        Args:
            profiles: The list of profile to delete

        Returns:
            Nothing.
        """
        self._profiles = list(
            filter(
                lambda p: p not in profiles,
                self.profiles
            )
        )
        self._update_profile_numbers()
        self._status.modified()


    def move_up_profile(self, index: int):
        if index < len(self.profiles):
            next = index - 1

            p = self._profiles
            p[index], p[next] = p[next], p[index]
            self._status.modified()

    def move_down_profile(self, index: int):
        if index >= 0:
            prev = index + 1

            p = self._profiles
            p[index], p[prev] = p[prev], p[index]
            self._status.modified()

    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.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.profiles]

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

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

    def get_kp_min(self):
        return min(self.get_kp())

    def get_kp_max(self):
        return max(self.get_kp())


    # Guidelines

    @timer
    def _compute_guidelines_cache(self, guide_set, named_points,
                                  complete, incomplete):
        # Reset guide lines cache
        self._guidelines = {}
        self._complete_guidelines = complete.copy()
        self._incomplete_guidelines = incomplete.copy()
        self._guidelines_is_valid = len(incomplete) == 0

        # 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
                )
            )

    @timer
    def compute_guidelines(self):
        """Compute reach guidelines

        Returns:
            Tuple of complete and incomplete guidelines name.
        """
        # 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
        )

        # Get incomplete guideline
        incomplete = set(
            reduce(
                lambda acc, h: acc + h,
                map(
                    lambda l: list(set(l).symmetric_difference(guide_set)),
                    points_name
                )
            )
        )

        complete = guide_set - incomplete

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

        self._status.modified()
        return (complete, incomplete)

    def _map_guidelines_points(self, func, full=False):
        if len(self._guidelines) == 0:
            _ = self.compute_guidelines()

        return list(
            # Map for each guideline
            map(
                lambda k: list(
                    # Apply function FUNC on each point of guideline
                    map(
                        func,
                        self._guidelines[k],
                    )
                ),
                # Get only guide lines if FULL is False
                self._guidelines if full else filter(
                    lambda x: x in self._complete_guidelines,
                    self._guidelines
                )
            )
        )

    def _complete_filter(self, gl):
        return gl in self._complete_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

    @timer
    def sort(self, is_reversed: bool = False):
        self._profiles = sorted(
            self._profiles,
            key=lambda profile: profile.kp,
            reverse=is_reversed
        )
        self._status.modified()

    @timer
    def sort_with_indexes(self, indexes: list):
        if len(self._profiles) != len(indexes):
            print("TODO: CRITICAL ERROR!")

        self._profiles = list(
            map(
                lambda x: x[1],
                sorted(
                    enumerate(self.profiles),
                    key=lambda x: indexes[x[0]]
                )
            )
        )
        self._status.modified()

    # Import/Export

    @timer
    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.
        """
        list_profile = []
        list_header = []

        try:
            list_profile, list_header = self.read_file_st(str(file_path_name))
            profile_header = ["num", "code1", "code2", "nb_point", "kp", "name"]

            if list_profile and list_header:
                for ind, profile in enumerate(list_profile):
                    d = {}
                    for i, data in enumerate(list_header[ind]):
                        d[profile_header[i]] = data

                    prof = ProfileXYZ(
                        **d, reach=self, status=self._status
                    )
                    prof.import_points(profile)
                    self._profiles.append(prof)
                    self._update_profile_numbers()

                self._status.modified()
        except FileNotFoundError as e:
            print(e)
            exception_message_box(e)
        except FileFormatError as e:
            print(e)
            e.alert()

    @timer
    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:
                            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

        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")