Add a register to count the number of times an observation is seen by the RL agent.

Ensure that the quadmesh and trimesh environments are functional with SB3 PPO and personal PPO.

Author: ArzhelaR

environment/environment_config.json

@@ -2,8 +2,8 @@
     "env_name": "Quadmesh-v0",
     "mesh_size": 16,
     "max_episode_steps": 30,
-    "n_darts_selected": 25,
-    "deep": 6,
+    "n_darts_selected": 10,
+    "deep": 8,
     "action_restriction": false,
     "with_degree_observation": false
 }

environment/gymnasium_envs/quadmesh_env/envs/quadmesh.py

@@ -9,6 +9,7 @@
 from mesh_model.mesh_analysis.quadmesh_analysis import global_score, isTruncated
 from environment.gymnasium_envs.quadmesh_env.envs.mesh_conv import get_x
 from environment.actions.quadrangular_actions import flip_edge, split_edge, collapse_edge, cleanup_edge
+from environment.observation_register import ObservationRegistry
 
 
 class Actions(Enum):
@@ -21,7 +22,7 @@ class Actions(Enum):
 class QuadMeshEnv(gym.Env):
     metadata = {"render_modes": ["human", "rgb_array"], "render_fps": 60}
 
-    def __init__(self, mesh=None, n_darts_selected=20, deep=6, with_degree_obs=True, action_restriction=False, render_mode=None):
+    def __init__(self, mesh=None, max_episode_steps=30, n_darts_selected=20, deep=6, with_degree_obs=True, action_restriction=False, render_mode=None):
         if mesh is not None:
             self.config = {"mesh": mesh}
             self.mesh = copy.deepcopy(mesh)
@@ -33,21 +34,22 @@ def __init__(self, mesh=None, n_darts_selected=20, deep=6, with_degree_obs=True,
         self._nodes_scores, self._mesh_score, self._ideal_score, self._nodes_adjacency = global_score(self.mesh)
         self._ideal_rewards = (self._mesh_score - self._ideal_score)*10
         self.next_mesh_score = 0
-        self.deep = deep
         self.n_darts_selected = n_darts_selected
         self.restricted = action_restriction
         self.degree_observation = with_degree_obs
         self.window_size = 512  # The size of the PyGame window
         self.g = None
         self.nb_invalid_actions = 0
+        self.max_steps = max_episode_steps
         self.darts_selected = [] # darts id observed
-        deep = self.deep*2 if self.degree_observation else deep
+        self.deep = deep*2 if self.degree_observation else deep
         self.observation_space = spaces.Box(
-            low=-15,  # nodes min degree : -15
-            high=15,  # nodes max degree : 15
-            shape=(self.n_darts_selected, self.deep * 2 if self.degree_observation else self.deep),
+            low=-6,  # nodes min degree : -15
+            high=2,  # nodes max degree : 15
+            shape=(self.n_darts_selected, deep),
             dtype=np.int64
         )
+        self.observation_count = ObservationRegistry(self.n_darts_selected, self.deep, -6, 2)
 
         self.observation = None
 
@@ -102,6 +104,8 @@ def _get_info(self, terminated, valid_act, action, mesh_reward):
             "invalid_collapse": 1.0 if action[0]==Actions.COLLAPSE.value and not valid_action else 0.0,
             "invalid_cleanup": 1.0 if action[0]==Actions.CLEANUP.value and not valid_action else 0.0,
             "mesh" : self.mesh,
+            "darts_selected" : self.darts_selected,
+            "observation_count" : self.observation_count,
         }
 
     def _action_to_dart_id(self, action: np.ndarray) -> int:
@@ -131,6 +135,7 @@ def step(self, action: np.ndarray):
         else:
             raise ValueError("Action not defined")
 
+        self.observation_count.register_observation(self.observation)
         if valid_action:
             # An episode is done if the actual score is the same as the ideal
             next_nodes_score, self.next_mesh_score, _, next_nodes_adjacency = global_score(self.mesh)

environment/observation_register.py

@@ -0,0 +1,46 @@
+
+class ObservationRegistry:
+    def __init__(self, n_darts_selected, deep, lowest_value, highest_value):
+        self.n_darts = n_darts_selected
+        self.deep = deep
+        self.low = lowest_value
+        self.high = highest_value
+        self.counts = {}
+
+    def encode(self, observation):
+        """
+        Converts an observation into a tuple.
+        :param observation:
+        :return: the tuple ID of the observation
+        """
+        ID = tuple(tuple(dart_surrounding) for dart_surrounding in observation)
+        return ID
+
+    def decode(self, ID):
+        """
+        Reconstructs an observation from its ID.
+        :param ID: a tuple ID
+        :return: the observation matrix
+        """
+        return [list(dart_surrounding) for dart_surrounding in ID]
+
+    def register_observation(self, observation) -> None:
+        """
+        Adds an observation to the register and increments its counter.
+        :param observation:
+        """
+        ID = self.encode(observation)
+        if self.counts.get(ID) is None:
+            self.counts[ID] = 1
+        else:
+            self.counts[ID] += 1
+
+    def get_count(self, observation):
+        """
+        Returns the number of times an observation has been seen by the agent.
+        :param observation:
+        :return: counts of observations
+        """
+        ID = self.encode(observation)
+        return self.counts.get(ID, 0)
+

main.py

@@ -1,7 +1,7 @@
 import sys
 
 from user_game import user_game
-from training.train import train
+from training.train_quadmesh import train
 from training.exploit import exploit
 #from mesh_model.reader import read_gmsh
 #from mesh_display import MeshDisplay

mesh_model/mesh_analysis/global_mesh_analysis.py

@@ -39,10 +39,14 @@ def score_calculation(n: Node) -> (int, int):
     :return: the irregularity of the node
     """
     adjacency = degree(n)
-    m = n.mesh
     # Check if the mesh is triangular or quad
-    d = Dart(m, m.dart_info[0,0])
-    triangular = d.get_beta(1).get_beta(1).get_beta(1) == d
+    d = n.get_dart()
+    if d.id <0:
+        raise ValueError("No existing dart")
+    d1 = d.get_beta(1)
+    d11 = d1.get_beta(1)
+    d111 = d11.get_beta(1)
+    triangular = (d111.id == d.id)
     if on_boundary(n):
         angle = get_boundary_angle(n)
         if triangular:
@@ -85,6 +89,7 @@ def get_angle(d1: Dart, d2: Dart, n: Node) -> float:
     cos_theta = np.clip(cos_theta, -1, 1)
     angle = np.arccos(cos_theta)
     if np.isnan(angle):
+        plot_mesh(n.mesh)
         raise ValueError("Angle error")
     return degrees(angle)
 

mesh_model/mesh_analysis/quadmesh_analysis.py

@@ -104,15 +104,14 @@ def isFlipOk(d: Dart) -> (bool, bool):
         return topo, geo
     else:
         d2, d1, d11, d111, d21, d211, d2111, n1, n2, n3, n4, n5, n6 = mesh.active_quadrangles(d)
-
+    # if degree are
     if not test_degree(n5) or not test_degree(n3):
         topo = False
         return topo, geo
 
     if d211.get_node() == d111.get_node() or d11.get_node() == d2111.get_node():
         topo = False
         return topo, geo
-
     topo = isValidQuad(n5, n6, n2, n3) and isValidQuad(n1, n5, n3, n4)
 
     """
@@ -174,6 +173,8 @@ def isCollapseOk(d: Dart) -> (bool, bool):
     f3 = (d11.get_beta(2)).get_face()
     f4 = (d111.get_beta(2)).get_face()
     adjacent_faces_lst=[f1.id, f2.id, f3.id, f4.id]
+
+    # Check that there are no adjacent faces in common
     if len(adjacent_faces_lst) != len(set(adjacent_faces_lst)):
         topo = False
         return topo, geo
@@ -251,6 +252,13 @@ def isValidQuad(A: Node, B: Node, C: Node, D: Node):
     u3 = np.array([D.x() - C.x(), D.y() - C.y()]) # vect(CD)
     u4 = np.array([A.x() - D.x(), A.y() - D.y()]) # vect(DA)
 
+    # Checking for near-zero vectors (close to (0,0))
+    if (np.linalg.norm(u1) < 1e-5 or
+            np.linalg.norm(u2) < 1e-5 or
+            np.linalg.norm(u3) < 1e-5 or
+            np.linalg.norm(u4) < 1e-5):
+        return False  # Quad invalid because one side is almost zero
+
     cp_A = cross_product(-1*u4, u1)
     cp_B = cross_product(-1*u1, u2)
     cp_C = cross_product(-1*u2, u3)
@@ -263,13 +271,12 @@ def isValidQuad(A: Node, B: Node, C: Node, D: Node):
         return False
 
     """
-    if signe(cp_A)+signe(cp_B)+signe(cp_C)+signe(cp_D)<2:
+    if 0< signe(cp_A)+signe(cp_B)+signe(cp_C)+signe(cp_D) <2 :
         return True
     else:
         return False
 
 
-
 def orientation(p, q, r):
     """ Calcule l'orientation de trois points.
         Retourne :

mesh_model/mesh_struct/mesh_elements.py

@@ -1,5 +1,5 @@
 from __future__ import annotations
-
+import numpy as np
 
 class Dart:
     _mesh_type: type = None
@@ -19,6 +19,8 @@ def __init__(self, m: _mesh_type, dart_id: int):
         :param id: a index, that corresponds to the location of the dart data in the mesh_struct dart container
         """
         self.mesh = m
+        if not isinstance(dart_id, (int, np.integer)):
+            raise ValueError(f"The id must be an integer, {dart_id} is type {type(dart_id)}.")
         self.id = dart_id
 
     def __eq__(self, a_dart: Dart) -> bool:
@@ -40,7 +42,6 @@ def get_beta(self, i: int) -> Dart:
         """
         if i < 1 or i > 2:
             raise ValueError("Wrong alpha dimension")
-
         if self.mesh.dart_info[self.id, i] == -1:
             return None
 
@@ -135,7 +136,7 @@ def get_dart(self) -> Dart:
         Get the dart value associated with this node
         :return: a dart
         """
-        return Dart(self.mesh, self.mesh.nodes[self.id, 2])
+        return Dart(self.mesh, int(self.mesh.nodes[self.id, 2]))
 
     def x(self) -> float:
         """

mesh_model/random_quadmesh.py

@@ -15,8 +15,8 @@ def random_mesh() -> Mesh:
     :param num_nodes_max: number of nodes of the final mesh
     :return: a random mesh
     """
-    filename = os.path.join('../mesh_files/', 't1_quad.msh')
-    mesh = read_gmsh(filename)
+    #filename = os.path.join('../mesh_files/', 't1_quad.msh')
+    mesh = read_gmsh("/home/ropercha/PycharmProjects/tune/mesh_files/t1_quad.msh")
     mesh_shuffle(mesh, 10)
     return mesh