Avant-propos

• L'appel de la méthode save_edges() est souvent suivi de la persistance dans le pkl.
• Dans le but de simplifier le code en évitant des répétitions, on modifie cette méthode, en y ajoutant un paramètre (optionnel).
  CtrlScene.save_edges() :

      def save_edges(self, backup=False):
          """ Mise à jour de la clé 'edges' du super-dictionnaire pkl.
          La persistance est prise en compte si backup=True. """
          s_id_edges = set()
          for gredge in self.o_grscene.items():
              if gredge.__class__.__name__ == 'UiEdge':
                  s_id_edges.add(gredge.o_edge.get_tid())
          self.o_pkl.od_pkl.write('edges', s_id_edges)
          if backup:
              self.o_pkl.backup()
  



• Ajout de la méthode get_gredges().
  CtrlScene.get_gredges() :

      def get_gredges(self):
          """ Renvoie la liste de tous les o_gredges contenus dans la scène. """
          l_items = self.o_grscene.items()
          lo_gredges = list()
          for o_gredge in l_items:
              if o_gredge.__class__.__name__ == 'UiEdge':
                  lo_gredges.append(o_gredge)
          return lo_gredges
  

   

• Correction de bug dans Pkl.restore() : En mode test, l'attribut fic_pkl était parfois mal affecté.
  pkl.py > Pkl.restore() :

      def restore(self):
          """ Passage unique, à l'instantiation.
          - Tentative de lecture du fichier params.pkl, produisant le dictionnaire de paramètres.
          - Si échec (inexistant ou non conforme), on utilise un dictionnaire vide.
          - Transformation en super-dictionnaire (self.od_pkl), grace à la classe Dictionary.
          """
          bk_path = os.path.dirname(__file__).replace('\\pc', '\\backups')
          self.fic_pkl = os.path.abspath(f'{bk_path}/{self.o_scene.graph_name}/params.pkl')
          try:
              """ Lecture du fichier pkl. """
              with open(self.fic_pkl, 'rb') as pk:    # Pas d'encoding pour les fichiers binaires.
                  d_pkl = pickle.load(pk)
          except (Exception,):
              d_pkl = {}
          self.od_pkl = Dictionary(d_pkl)        # OrderedDict de tout le graphe (vue + nodes + edges + groupes + labels)
  

   

Description

Couleurs

Propagation de la couleur. La logique est la suivante :

• Seuls les sockets de sortie peuvent fixer leur couleur, modifiable dans le panneau des paramètres.

• Un edge prend la couleur de son socket_out de départ.
• Un edge impose sa couleur au socket_in d'arrivée.
• Toute modification de la couleur et immédiatement répercutée en cascade.

Pour satisfaire à ces conditions, on imagine quel sera le code :

1. Au lancement de l'application, lors des instanciations :
   1. Affecter sa couleur à chaque socket_out, dans son setup, depuis le super-dictionnaire od_pkl.
   2. Affecter sa couleur à chaque edge, dans son setup. color = o_socket_out.color
   3. Toujours dans le setup du edge, affecter sa couleur au socket_in d'arrivée. o_socket_in.color = o_edge.color
2. Lors de la modification de couleur dans le dockable des paramètres :
   1. Affecter la couleur au socket_out concerné.
   2. Propager cette couleur à tous les edges partant de ce socket_out.
   3. Propager cette couleur à tous les socket_in d'arrivée.

Réalisation :

• Modification du code de la méthode get_default() de la classe-mère CtrlNode.
  CtrlNode.get_default() :

      def get_default(self):
          """ Titre du node et couleurs des 'socket_out' pour chaque type de node. """
          d_default = {'Titre du node': 'Choisir un titre'}
          d_colors = dict()
          for d_output in self.ld_outputs:
              if isinstance(d_output, dict):
                  d_colors[d_output['label']] = self.default_color
          if d_colors:
              d_default['Couleurs'] = d_colors
          return {self.main_key: d_default}
  

  Oui mais ... l'attribut default_color doit être déclaré.
  Ajouter cette ligne dans le groupe """ Sockets """ de CtrlNode.__init__() :

          self.default_color = '#88bbff'  # Couleur par défaut des sockets de sortie.
  

   

Par exemple, signaux.py > Node.get_default() devient :

    def get_default(self):
        d_updated = super().get_default().copy()
        d_updated[self.main_key].update({
            'Sinus': True,
            'Cosinus': False,
            'Carré': False,
            'Triangle': False,
            'Dent de scie m': False,
            'Dent de scie d': False,
            'Amplitude': 20,
            'Points par période': 100,
        })
        return d_updated

• Ajout de l'attribut color à CtrlSocket.
• Dans CtrlSocket.__init__() :

          self.color = '#bbb'                     # Gris clair pour les socket_in libres (non connectés).
  

   

• Affectation de cet attribut dans le setup, seulement aux sockets de sortie.
  CtrlSocket.setup() :

      def setup(self):
          if not self.b_input:
              """ Socket de sortie. """
              l_keys = ['Couleurs', self.d_datas['label']]
              self.color = self.o_node.get_param(l_keys, '#666')
          self.o_grsocket = UiSocket(self)
  

   

• Prise en compte pour l'affichage, dans l'UI du socket.
  UiSocket.state_brush() :

      @property
      def state_brush(self):
          """ Couleur de fond : paramétrée (on), gris (off), ou blanc (survol). """
          color = self.o_socket.color if self.o_socket.o_node.b_on else self.o_socket.o_node.d_display['col_brush']['off']
          color = '#fff' if self.b_hover else color
          return pg.mkBrush(color)
  

   

• Ajouter l'attribut color.
  Dans CtrlEdge.__init__() :

          self.color = '#bbb'
  

  Affectation dans CtrlEdge.setup() :

      def setup(self):
          self.o_gredge = UiEdge(self)                    # Dessin de l'edge dans l'UI.
          self.o_scene.o_grscene.addItem(self.o_gredge)   # Ajout de l'edge dans la scène.
  
          """ Couleurs, sauf si l'edge est en cours de construction. """
          if self.o_socket_in.__class__.__name__ == 'CtrlSocket':
              self.color = self.o_socket_out.color
              self.o_socket_in.color = self.color         # On impose sa couleur au socket_in d'arrivée.
  

  Affichage effectif dans l'UI. UiEdge.state_pen() :

      @property
      def state_pen(self):
          """ Couleur, épaisseur, style : Choix en fonction d'événements. """
          b_building = self.o_edge.o_socket_in.__class__.__name__ == 'QPointF'    # Edge en construction.
          color = '#faa' if self.isSelected() else self.o_edge.color
          color = '#ffa' if self.b_hover else color
          color = Qt.white if b_building else color
          line_type = Qt.DashLine if b_building else Qt.SolidLine
          line_width = 1 if b_building else 2
          return QPen(QColor(color), line_width, line_type)
  

• Toute modification d'un paramètre de node appelle la méthode refresh().
• En conséquence, celle-ci doit être augmentée pour prendre en compte les couleurs.
  CtrlNode.refresh() :

      def refresh(self, l_keys):
          """ Titre du node. """
          if l_keys[-1] == 'Titre du node':
              self.o_grnode.set_title()
          """ Couleur des sockets et des edges. """
          if l_keys[-2] == 'Couleurs':
              self.set_colors(l_keys[1:])
  

   

• Oui mais ... pour séparer le code, cette tâche est confiée à la méthode set_colors().
  CtrlNode.set_colors() :

      def set_colors(self, l_keys):
          """ - La couleur est appliquée au socket_out concerné.
              - Elle est également appliquée à tous ses éventuels edges connectés.
              - A leur tour, ceux-ci propagent cette couleur à leurs sockets d'arrivée. """
          for o_socket_out in self.lo_sockets_out:
              # à coder ... (Servez-vous des TDD).
              pass
  
          """ Rafraîchissement général. """
          self.o_grnode.update()
  

Faire un court-circuit

Exemple de court-circuit :

Avant                                                                                   Après

Points à respecter :

• Le court-circuit est exécuté en "secouant" un node (dans cet exemple, celui du milieu).
• Autorisation de court-circuit :
  • Dans le setup de chacun des types de node, la liste des court-circuits faisables est affectée.
  • Par défaut, c'est une liste vide.
  • Exemple de liste pour le macd : self.l_short_circuits = [(0, 0), (0, 2)]
    Qui signifie : "entrée 0 avec sortie 0" et "entrée 0 avec sortie 2".
• Dans cet exemple, les 2 edges ont été remplacés par un seul.
• Certains types de node possèdent plusieurs entrées et / ou plusieurs sorties :
  • Condition à respecter pour supprimer l'edge d'entrée :
    • Toutes les sorties doivent exister dans la liste self.l_short_circuits.
  • Contrainte technologique :
    • Interdiction d'avoir plusieurs entrées en court-circuit sur une même sortie.

1. Chaque type de node a sa liste de faisabilité, vérifiée au démarrage.
2. Détection des secousses.
3. Création du court-circuit.

• Ajouter les attributs l_short_circuits, l_shortables et b_removable dans CtrlNode.
  Dans CtrlNode.__init__() ajouter cette section :

          """ Secousses => Court-circuits. """
          self.l_short_circuits = list()  # Surchargé par les classes dérivées.
          self.l_shortables = list()
          self.b_removable = True  # Les edges d'entrée peuvent être détruits en cas de court-circuit.
  

   

• Pour chacun des types de nodes dont vous désirez gérer les court-circuits, surchargez l'attribut l_short_circuits dans leur setup.
  Exemple dans moy_mobile.py > Node.setup(), après les listes de sockets :

          self.l_short_circuits = [(0, 0)]    # Entrée N°0 'court-circuitable' avec la sortie N°0.
  

  
  Exemple dans macd.py > Node.setup(), après les listes de sockets :

          self.l_short_circuits = [(0, 0), (0, 2)]    # e0 -> s0 et e0 -> s2
  

   

• Ensuite, on appelle la méthode de vérification short_circuit_check() au lancement.
  En cas d'erreur, celle-ci affiche un message explicatif et le programme se termine.
  Refaire la méthode CtrlNode.setup(), refactorée, qui appelle cette vérification en dernière ligne :

      def setup(self, child_file):
          """ Code appelant : classe dérivée, on connait donc ses attributs. """
          """ Paramètres. """
          self.o_params = Parameters(self)
  
          """ Hauteur automatique du node : elle dépend du nombre de sockets en entrée ou en sortie (le plus grand).  """
          nb_sockets_max = max(len(self.ld_inputs), len(self.ld_outputs))  # En fait : Sockets + séparateurs.
          h_title = self.d_display['geometry']['h_title']
          first_y = 16 * (1 + (h_title + 4) // 16)  # Multiple de 16.
          self.height = first_y + nb_sockets_max * 16 - 4
  
          """ Nouveau node par drag & drop. """
          b_new = self.pos != (0, 0)  # Nouveau node (node dropé).
          if b_new:
              """ - Cas d'un nouveau node (dropé).
                  - Son centre apparaît exactement à la position du curseur de la souris.
                  - On mémorise son path et sa position. """
              self.pos = self.pos[0] - self.width // 2, self.pos[1] - self.height // 2
              path = 'nodes' + os.path.relpath(child_file).split('nodes')[1][:-3].replace(os.sep, '.')
              self.od_pkl.write([self.s_id, 'path'], path)
          else:
              self.pos = self.od_pkl.read([self.s_id, 'position'], (0, 0))
  
          """ Instanciation de l'UI. """
          self.o_grnode = UiNode(self)  # Gestion de l'UI (Interface utilisateur) : dessin couleurs, ...
          if b_new:
              self.o_grnode.save_pos()  # Important ! à la fin (contient o_pkl.backup()).
          self.o_scene.o_grscene.addItem(self.o_grnode)  # Incorporation dans la scène.
  
          """ Sockets. """
          for i, d_input in enumerate(self.ld_inputs):
              """ On complète les dictionnaires de base. """
              if isinstance(d_input, dict):  # Si ce n'est pas un dictionnaire, c'est un séparateur.
                  d_input['id'] = True, self.id, i  # On ajoute l'identifiant. Tuple (in ?, id_node, num_socket)
                  self.lo_sockets_in.append(CtrlSocket(self, d_input))  # Socket instancié et ajouté à la liste.
              else:
                  """ Permet d'éviter les erreurs d'indice. """
                  self.lo_sockets_in.append('sep')  # Séparateur.
          for i, d_output in enumerate(self.ld_outputs):
              """ On complète les dictionnaires de base. """
              if isinstance(d_output, dict):  # Si ce n'est pas un dictionnaire, c'est un séparateur.
                  d_output['id'] = False, self.id, i  # On ajoute l'identifiant. Tuple (in ?, id_node, num_socket)
                  self.lo_sockets_out.append(CtrlSocket(self, d_output))  # Socket instancié et ajouté à la liste.
              else:
                  self.lo_sockets_out.append('sep')  # Séparateur.
  
          self.short_circuit_check()
  

• Oui mais ... la méthode short_circuit_check() n'existe pas. On la crée :
  CtrlNode.short_circuit_check() :

      def short_circuit_check(self):
          """ Passage unique dans cette méthode, au démarrage. Fin programme si erreur.
              - Peut-on supprimer les edges d'entrée en cas de court-circuit ?
              - Oui, à condition que TOUTES les sorties existent dans self.l_short_circuits. """
          l_indx_outputs = list()  # Liste des index de sortie.
          d_counter = dict()
          for short_circuit in self.l_short_circuits:  # Exemple -> self.l_short_circuits = [(0, 0), (0, 2)]
              num_input = short_circuit[0]
              num_output = short_circuit[1]
              if num_input >= len(self.ld_inputs):
                  """ Fin programme. """
                  raise SystemExit(f"La liste des court-circuits du type de node '{self.type}' est erronée."
                                   f"\nL'entrée N°{num_input} est hors limites."
                                   f"\nRevoir le contenu de self.l_short_circuits dans le setup de '{self.type}'.")
              if num_output >= len(self.ld_outputs):
                  """ Fin programme. """
                  raise SystemExit(f"La liste des court-circuits du type de node '{self.type}' est erronée."
                                   f"\nLa sortie N°{num_output} est hors limites."
                                   f"\nRevoir le contenu de self.l_short_circuits dans le setup de '{self.type}'.")
              if isinstance(self.lo_sockets_in[num_input], str):
                  """ Fin programme. """
                  raise SystemExit(f"La liste des court-circuits du type de node '{self.type}' est erronée."
                                   f"\nL'entrée N°{num_input} n'est pas un socket mais un séparateur."
                                   f"\nRevoir le contenu de self.l_short_circuits dans le setup de '{self.type}'.")
              if isinstance(self.lo_sockets_out[num_output], str):
                  """ Fin programme. """
                  raise SystemExit(f"La liste des court-circuits du type de node '{self.type}' est erronée."
                                   f"\nLa sortie N°{num_output} n'est pas un socket mais un séparateur."
                                   f"\nRevoir le contenu de self.l_short_circuits dans le setup de '{self.type}'.")
              l_indx_outputs.append(num_output)  # Dans cet exemple ->                   ( 0  ,    2 )
              if num_output in d_counter:
                  d_counter[num_output] += 1
              else:
                  d_counter[num_output] = 1
          for i, d_output in enumerate(self.ld_outputs):
              if isinstance(d_output, dict):
                  if i not in l_indx_outputs:
                      """ Condition non satisfaite. """
                      self.b_removable = False  # b_removable est à True dans __init__().
                      break
          """ Contrainte technologique : Plusieurs entrées sur une même sortie est interdit ! """
          for nb_inputs in d_counter.values():
              if nb_inputs > 1:
                  """ Fin programme. """
                  raise SystemExit("Plusieurs entrées sont court-circuitables sur une même sortie."
                                   f"\nRevoir le contenu de self.l_short_circuits dans le setup de '{self.type}'.")
  

  
   
• Vérifer, en créant des listes correctes mais aussi des listes incorrectes.

Retour au plan

• Condition A : Un court-circuit peut être réalisé si les conditions précédentes sont satisfaites : fixes, définies par le code.
• Condition B : De plus, évidemment, la présence d'edges est nécessaire : celle-ci dépend du dessin de votre graphe.
• Si l'une de ces conditions n'est pas satisfaite, le court-circuit n'aura pas lieu.
• Les secousses se font en agitant la souris, bouton gauche appuyé sur le node.
• Lorsque le bouton gauche de la souris est appuyé, on vérifie la condition B. Donc :
  Ajouter la méthode UiNode.mousePressEvent() :

      def mousePressEvent(self, ev):
          """ Crée la liste des court-circuits faisables sur ce node : o_node.l_shortables.
          Un court-circuit est un tuple de 2 objets o_gredge : (o_gredge en entrée, o_gredge en sortie). """
          self.o_node.set_shortables()
          super().mousePressEvent(ev)
  

   

• Oui mais ... cette méthode affecte la liste l_shortables en appelant la méthode set_shortables() qui n'existe pas. On la crée.
  CtrlNode.set_shortables() :

      def set_shortables(self):
          """ Méthode appelée lorsque le bouton gauche de la souris est appuyé (sur ce node).
              - Chaque type de node possède sa liste de court-circuits faisables.
              - Retourne une liste de tuples.
              - Chaque tuple est composé des 2 gredges pouvant être réunis pour n'en faire qu'un. """
          self.l_shortables = list()
          for short_circuit in self.l_short_circuits:  # Exemple -> self.l_short_circuits = [(0, 0), (0, 2)]
              lo_gredges_in = list(self.lo_sockets_in[short_circuit[0]].get_gredges())
              lo_gredges_out = list(self.lo_sockets_out[short_circuit[1]].get_gredges())
              if lo_gredges_in:  # Cette liste contient 0 ou 1 edge.
                  for o_gredge_out in lo_gredges_out:
                      self.l_shortables.append((lo_gredges_in[0], o_gredge_out))
  

   

• La méthode mouseMoveEvent() est automatiquement appelée lorsque la souris bouge, ce qui est le cas.
  UiNode.mouseMoveEvent() :

      def mouseMoveEvent(self, ev):
          self.o_node.shake()
          super().mouseMoveEvent(ev)
  

  Oui mais ... cette méthode appelle la méthode shake() qui n'existe pas. On la crée.
  CtrlNode.shake() :

      def shake(self):
          def shortcircuit():
              for to_gredge in self.l_shortables:                         # Tuples de gredges.
                  o_socket_out = to_gredge[0].o_edge.o_socket_out         # From, depuis.
                  o_socket_in = to_gredge[1].o_edge.o_socket_in           # To, jusqu'à.
  
                  """ Suppression des 2 edges du tuple. """
                  self.o_scene.o_grscene.removeItem(to_gredge[1])         # Edge de sortie.
                  if self.b_removable:
                      self.o_scene.o_grscene.removeItem(to_gredge[0])     # Edge d'entrée.
  
                  """ Création du nouvel edge de court-circuit. """
                  CtrlEdge(self.o_scene, o_socket_out, o_socket_in)
  
              """ Persistance. """
              self.o_scene.save_edges(backup=True)
  
          def raz_shakes():
              """ Après cette raz, un nouveau court-circuit est autorisé. """
              self.k_shake = 0
              self.b_shaking = False
  
          if not self.l_shortables:
              return
          self.dt.delay(raz_shakes, delay=100)
          x, y = self.o_grnode.pos().x(), self.o_grnode.pos().y()  # Positions.
          dx, dy = x - self.x, y - self.y  # Déplacements (delta x, delta y).
          if (dx * self.dx < 0) or (dy * self.dy < 0):
              """ Le mouvement a changé de sens => incrémentation du compteur. """
              self.k_shake += 1
              if self.k_shake > 5:  # Valeur à ajuster.
                  shortcircuit()
  
          """ Mémorisation. """
          self.x, self.y = x, y
          self.dx, self.dy = dx, dy
  

  Ne pas oublier d'importer la classe CtrlEdge.

• Comment ça marche ?
  • Lorsque la souris se déplace, cette méthode est exécutée de manière répétée. Elle peut être répété rapidement des centaines de fois !
  • A chaque fois, on note la position (x, y) de la souris.
  • On calcule les déplacements (ou variations delta) : (dx, dy).
  • Dans les dernières lignes, on mémorise ces valeurs -> (self.x, self.y) et (self.dx, self.dy).
  • Si le produit dx * self.dx est négatif, cela signifie que la souris a changé de sens dans l'axe des x. Idem pour les y.
    C'est une secousse, que l'on comptabilise.
  • Si, pendant 100ms il n'y a pas de secousse, le compteur retombe à zéro (raz_shakes())
  • Au bout de 5 secousses, on estime que le node a été suffisament secoué, on autorise la création du court-circuit.
    La création du court-circuit est réalisée dans la fonction interne shortcircuit().
  • Remarque importante : ces valeurs (100 ms et 5 secousses) sont à ajuster par vous-même.
  • Des attributs sont à ajouter à CtrlNode : x, y, dx, dy, le compteur, le temporisateur, etc.
    Le groupe """ Secousses => Court-circuits. """ de CtrlNode.__init__() devient :

            """ Secousses => Court-circuits. """
            self.dt = DateTime()
            self.l_short_circuits = list()  # Surchargé par les classes dérivées.
            self.l_shortables = list()
            self.b_removable = True  # Les edges d'entrée peuvent être détruits en cas de court-circuit.
            self.k_shake = 0  # Nombre de secousses.
            self.b_shaking = False  # Secousse en cours.
            self.x, self.y = 0, 0  # Position antérieure (x, y), mémorisée.
            self.dx, self.dy = 0, 0  # Deltas (variations) antérieurs, mémorisés.
    

    Ne pas oublier d'importer DateTime.

Retour au plan

• Cette fonction est interne à la méthode shake().
• Elle est appelée dès qu'une secousse est détectée.
• La liste des court-circuits à faire a été créée au moment où le bouton gauche de la souris a été appuyé.
• Il s'agit de la liste l_shortables, qui contient les paires d'edges à réunir (tuples de 2 gredges).
• Pour chaque paire, les opérations effectuées sont :
  • Mémorisation des sockets qui définiront le nouvel edge : o_socket_out et o_socket_in.
  • Suppression de l'edge de sortie.
  • Suppression conditionnelle de l'edge d'entrée.
    • Rappel : La condition est le flag b_removable, calculé au lancement, dans short_circuit_check().
  • Création d'un nouvel edge, reprenant les sockets notés précedemment.
• Mémorisation et persistance de l'ensemble (hors boucle).

Insertion d'un node dans un edge

Avant                                                                                   Après

• Il s'agit de l'opération inverse d'un court-circuit.
  • Un edge va être supprimé : A---B
  • Deux edges vont être créés. C---B  et A---C
• Des conditions doivent être réunies pour pouvoir effectuer cette action :
  • Edge en création C---B. Son point de départ est le socket de sortie de C, noté Sc.
  • Entrée du node B : Le socket d'arrivée du mi-edge est déjà occupé par un edge.
  • Node C : Le node de départ du mi-edge (node C) a au moins un socket d'entrée libre, noté Ec.
  • Node C : Le tuple formé par les 2 sockets concernés (Ec, Sc) existe dans sa liste l_short_circuits.
• Où placer le code ?
  Rappelez-vous : au moment où l'on termine la création du mi-edge, la méthode stop_edge est appelée, dans UiView.
  C'est donc cette méthode que l'on va (encore) compléter.
  UiView.stop_edge() :

      def stop_edge(self, ev):
          def break_edge(o_socket_out):
              """ Remplace un edge par 2 edges. Plusieurs conditions sont nécessaires :
                  - Le socket d'arrivée du mi-edge est déjà occupé par un edge.
                  - Le node de départ a au moins un socket d'entrée libre.
                  - Le tuple formé par ce socket d'entrée et le socket de sortie en cours est dans l_short_circuits.
              """
              o_node = self.o_mid_edge.o_socket_out.o_node            # Node de départ.
              num_socket_out = self.o_mid_edge.o_socket_out.t_id[1]   # N° du socket de sortie en cours de liaison.
              # à coder ... (sans TDD).
  
          if self.o_mid_edge.__class__.__name__ == 'CtrlEdge':
              """ Reset. """
              self.normal_size_sockets()          # Rétablit les tailles normales aux sockets.
              self.o_scene.o_grscene.removeItem(self.o_mid_edge.o_gredge)     # Effacement du mi_edge.
  
              """ Drop = fin du drag. """
              if self.is_dropable(ev):
                  gr_item = self.itemAt(ev.pos())     # UiSocket
  
                  """ Suppression d'un éventuel edge connecté à ce socket. """
                  for o_gredge in gr_item.o_socket.get_gredges():
                      break_edge(o_gredge.o_edge.o_socket_out)        # "Dé-court-circuit" : remplace un edge par 2 edges.
                      self.o_scene.o_grscene.removeItem(o_gredge)     # Effacement.
  
                  """ Création d'un edge complet. """
                  CtrlEdge(self.o_scene, self.o_mid_edge.o_socket_out, gr_item.o_socket)
  
                  """ Enregistrement. """
                  self.o_scene.save_edges(backup=True)
  
              """ Destruction de l'objet. """
              self.o_mid_edge = None
  

• Cette opération est effectuée par la fonction interne break_edge(), qui procède aux vérifications préalables.
  
   

Croisement de deux edges

• Cette opération est assurée par la méthode cross_edges() de la scène.
  CtrlScene.cross_edges() :

    def cross_edges(self):
        """ Croisement de 2 edges. Conditions :
            - 2 edges sélectionnés.
            - Appui sur la touche 'Tab'.
            - Les extrémités d'un edge à créer appartiennent à des nodes différents.
            - Si un cas est non autorisé (nb d'edges sélectionnés différent de 2 ou bouclage sur un même node) :
                |_ On emet un bip (importer la classe windsound). """
        # à coder ... (Servez-vous des TDD).

• Cette méthode est appelée lorsqu'on appuie sur la touche Tab du clavier.

• Assurez-vous que l'état est rétabli lorsqu'on appuie à nouveau sur la touche Tab.

Avant                                                                                   Après
 

Non autorisé car 2 extrémités d'un edge à créer appartiendraient au même node.

Vérification

from PyQt5.QtWidgets import QApplication
from PyQt5.QtCore import QTimer
from pc.main import UiMain
import pytest
import random
import os
import sys


class TestEdges:
    g_appli = QApplication(sys.argv)
    tempo = QTimer()
    tempo.setSingleShot(True)

    @pytest.fixture(autouse=True)  # scope=session (un seul appel), autouse=True (un appel par test)
    def setup(self):
        self.remove_files()
        bk_test = os.path.abspath(f"{os.path.dirname(__file__)}/../backups/tests")
        os.makedirs(bk_test, exist_ok=True)  # Création dossier.
        o_main = self.open_window()
        o_main.resize(1200, 600)
        o_main.open_graph('tests')
        o_scene = None
        for i in range(o_main.tabGraphs.count()):
            if o_main.tabGraphs.widget(i).graph_name == 'tests':
                o_scene = o_main.tabGraphs.widget(i)
                break
        assert o_scene is not None
        assert o_scene.__class__.__name__ == 'CtrlScene'
        return {
            'main': o_main,
            'scene': o_scene
        }

    @staticmethod
    def remove_files():
        """ Appel final pour effacer les fichiers temporaires créés par les tests. """
        tests_folder = os.path.dirname(__file__)
        graph_folder = tests_folder.replace('tests', f'backups{os.sep}tests')
        files = [f'{tests_folder}{os.sep}params.conf', f'{graph_folder}{os.sep}params.pkl']
        for file in files:
            if os.path.isfile(file):
                os.remove(file)
        if os.path.isdir(graph_folder):
            os.rmdir(graph_folder)

    @pytest.fixture(scope="session", autouse=True)
    def cleanup(self, request):
        request.addfinalizer(self.remove_files)

    @staticmethod
    def open_window():
        win = UiMain()
        win.show()
        return win

    # noinspection PyUnresolvedReferences
    def connect(self, function, delay):
        try:
            self.tempo.timeout.disconnect()
        except TypeError:
            pass
        self.tempo.timeout.connect(function)
        self.tempo.start(delay)

    @staticmethod
    def build_graph(setup, l_nodes=None, s_edges=None):
        o_main = setup['main']
        o_scene = setup['scene']
        o_scene.o_pkl.od_pkl.clear()

        """ Valeurs par défaut. """
        if l_nodes is None:
            l_nodes = [
                (90, 'nodes.indicateurs.macd', (-256, -32)),
                (91, 'nodes.operateurs.union', (148, -96)),
            ]
        if s_edges is None:
            s_edges = {
                ((90, 0), (91, 4)),
                ((90, 2), (91, 2))
            }

        """ Dans od_pkl puis affichage. """
        for node in l_nodes:
            o_scene.o_pkl.od_pkl.write([f'Node{node[0]}', 'path'], node[1])
            o_scene.o_pkl.od_pkl.write([f'Node{node[0]}', 'position'], node[2])
        o_scene.o_pkl.od_pkl.write('edges', s_edges)
        o_scene.show_nodes()
        o_scene.show_edges()

        l_items = o_scene.o_grscene.items()
        l_grnodes, l_grsockets, l_gredges = list(), list(), list()
        for o_gritem in l_items:
            if o_gritem.__class__.__name__ == 'UiNode':
                l_grnodes.append(o_gritem)
            elif o_gritem.__class__.__name__ == 'UiSocket':
                l_grsockets.append(o_gritem)
            elif o_gritem.__class__.__name__ == 'UiEdge':
                l_gredges.append(o_gritem)

        return o_main, o_scene, l_grnodes, l_grsockets, l_gredges

    # ***************************************** TESTS *****************************************
    # @pytest.mark.skip
    def test_tid(self, setup):
        o_main, o_scene, lo_grnodes, lo_grsockets, lo_gredges = self.build_graph(setup)
        for o_gredge in lo_gredges:
            assert o_gredge.o_edge.get_tid() == (o_gredge.o_edge.o_socket_out.t_id, o_gredge.o_edge.o_socket_in.t_id)
        self.connect(o_main.close, 1000)
        self.g_appli.exec()

    # @pytest.mark.skip
    def test_save_edge(self, setup):
        o_main, o_scene, lo_grnodes, lo_grsockets, lo_gredges = self.build_graph(setup)
        o_scene.o_grscene.removeItem(lo_gredges[0])
        o_scene.save_edges()
        assert o_scene.o_pkl.od_pkl.read('edges') == set(o_gredge.o_edge.get_tid() for o_gredge in lo_gredges[1:])

        self.connect(o_main.close, 1000)
        self.g_appli.exec()

    # @pytest.mark.skip
    def test_get_gredges(self, setup):
        s_edges = {
            ((90, 0), (91, 4)),
            ((90, 2), (91, 0)),
            ((90, 2), (91, 1)),
            ((90, 2), (91, 2)),
        }
        o_main, o_scene, lo_grnodes, lo_grsockets, lo_gredges = self.build_graph(setup, s_edges=s_edges)
        for o_grsocket in lo_grsockets:
            s_gredges = set()
            for o_gredge in o_grsocket.o_socket.o_node.o_scene.o_grscene.items():
                if o_gredge.__class__.__name__ == 'UiEdge':
                    """ Choix du socket accroché à la bonne extrémité du edge. """
                    o_edge = o_gredge.o_edge
                    o_socket = o_edge.o_socket_in if o_grsocket.o_socket.b_input else o_edge.o_socket_out
                    if o_socket == o_grsocket.o_socket:
                        s_gredges.add(o_gredge)
            assert o_grsocket.o_socket.get_gredges() == s_gredges   # Méthode testée : CtrlSocket.get_gredges()

        self.connect(o_main.close, 1000)
        self.g_appli.exec()

    # @pytest.mark.skip
    def test_get_grsockets(self, setup):
        o_main, o_scene, lo_grnodes, lo_grsockets, lo_gredges = self.build_graph(setup)
        l_grsockets_in, l_grsockets_out = list(), list()
        for o_grsocket in lo_grsockets:
            if o_grsocket.o_socket.b_input:
                l_grsockets_in.append(o_grsocket)
            else:
                l_grsockets_out.append(o_grsocket)

        t_grsockets = o_scene.get_grsockets()
        assert t_grsockets[0] == l_grsockets_in
        assert t_grsockets[1] == l_grsockets_out

        self.connect(o_main.close, 1000)
        self.g_appli.exec()

    # @pytest.mark.skip
    def test_set_colors(self, setup):
        """ 2 nodes reliés par un edge. Application de la couleur au socket de départ. """
        l_nodes = [
            (90, 'nodes.indicateurs.macd', (-256, -32)),
            (91, 'nodes.operateurs.union', (148, -96)),
        ]
        s_edges = {
            ((90, 0), (91, 4)),
        }
        o_main, o_scene, lo_grnodes, lo_grsockets, lo_gredges = self.build_graph(setup, l_nodes=l_nodes, s_edges=s_edges)
        o_edge = lo_gredges[0].o_edge
        o_socket_out = o_edge.o_socket_out
        o_socket_in = o_edge.o_socket_in
        o_node_from = o_socket_out.o_node

        """ Simulation de modification du paramètre 'Couleurs' -> Socket de sortie du macd. """
        key = ['Node90', "Paramètres du node 'Node90'", 'Couleurs', o_socket_out.label]
        colors = ('#000000', '#ff0000', '#00ff00', '#0000ff', '#ffff00', '#00ffff', '#ff00ff', '#ffffff')
        color = random.choice(colors)
        o_node_from.o_params.od_real.write(key, color)
        o_node_from.o_params.set_params()

        """ Méthode testée : CtrlNode.set_colors(). """
        o_node_from.set_colors(['Couleurs', o_socket_out.label])

        """ Vérification de l'application de la couleur au socket de départ. """
        assert o_socket_out.color == color, "La couleur du socket de départ n'a pas été prise en compte."

        """ Vérification de la propagation vers l'edge. """
        assert o_edge.color == color, "La couleur du socket de départ n'a pas été propagée vers l'edge."

        """ Vérification de la propagation vers le socket d'arrivée. """
        assert o_socket_in.color == color, "La couleur de l'edge n'a pas été propagée vers le socket d'arrivée."

        """ Fin. """
        self.connect(o_main.close, 1000)
        self.g_appli.exec()

    # @pytest.mark.skip
    def test_cross_edges(self, setup):
        """ 3 nodes reliés par des edges. Croisement des edges. """
        l_nodes = [
            (0, 'nodes.operateurs.labo', (-256, -32)),
            (1, 'nodes.operateurs.labo', (-40, -160)),
            (2, 'nodes.operateurs.labo', (180, -48)),
        ]
        s_edges = {
            ((0, 0), (1, 0)),
            ((0, 1), (1, 2)),
            ((0, 0), (2, 1)),
            ((1, 1), (2, 2)),
        }
        o_main, o_scene, lo_grnodes, _, lo_gredges = self.build_graph(setup, l_nodes=l_nodes, s_edges=s_edges)

        """ Test 1 : Vérification du node de type 'Labo'. """
        o_labo = lo_grnodes[0].o_node
        msg = "Pour pouvoir effectuer ces tests, le type 'Labo' doit avoir :" \
              "\n- au moins 3 entrées," \
              "\n- au moins 2 sorties," \
              "\n- aucun séparateur de sockets."
        for o_socket in o_labo.lo_sockets_in + o_labo.lo_sockets_out:
            assert o_socket.__class__.__name__ == 'CtrlSocket', msg

        """ Test 2 : Croisement normal : 2 edges sélectionnés. """
        for o_gredge in lo_gredges:
            t_id = o_gredge.o_edge.get_tid()
            if t_id == ((0, 0), (1, 0)) or t_id == ((0, 1), (1, 2)):
                o_gredge.setSelected(True)
            else:
                o_gredge.setSelected(False)

        o_scene.cross_edges()
        l_tid = sorted([o_gredge.o_edge.get_tid() for o_gredge in o_scene.get_gredges()])
        msg = "Les edges ((0, 0), (1, 0)) et ((0, 1), (1, 2)) auraient dû être croisés, ce n'est pas le cas."
        assert l_tid == [((0, 0), (1, 2)), ((0, 0), (2, 1)), ((0, 1), (1, 0)), ((1, 1), (2, 2))], msg

        """ Test 3 : 'Dé-croisement' des 2 mêmes edges. """
        o_scene.cross_edges()
        l_tid = sorted([o_gredge.o_edge.get_tid() for o_gredge in o_scene.get_gredges()])
        msg = "Les edges ((0, 0), (1, 0)) et ((0, 1), (1, 2)) auraient dû être rétablis, ce n'est pas le cas."
        assert l_tid == [((0, 0), (1, 0)), ((0, 0), (2, 1)), ((0, 1), (1, 2)), ((1, 1), (2, 2))], msg

        """ Test 4 : Un seul edge sélectionné. """
        for o_gredge in o_scene.get_gredges():
            t_id = o_gredge.o_edge.get_tid()
            if t_id == ((0, 0), (1, 0)):
                o_gredge.setSelected(True)
            else:
                o_gredge.setSelected(False)
        o_scene.cross_edges()
        l_tid = sorted([o_gredge.o_edge.get_tid() for o_gredge in o_scene.get_gredges()])
        msg = "Un seul edge sélectionné => Il ne doit y avoir aucun changement."
        assert l_tid == [((0, 0), (1, 0)), ((0, 0), (2, 1)), ((0, 1), (1, 2)), ((1, 1), (2, 2))], msg

        """ Test 5 : Tous les edges sélectionnés. """
        for o_gredge in o_scene.get_gredges():
            o_gredge.setSelected(True)
        o_scene.cross_edges()
        l_tid = sorted([o_gredge.o_edge.get_tid() for o_gredge in o_scene.get_gredges()])
        msg = "Plus de 2 edges sont sélectionnés => Il ne doit y avoir aucun changement."
        assert l_tid == [((0, 0), (1, 0)), ((0, 0), (2, 1)), ((0, 1), (1, 2)), ((1, 1), (2, 2))], msg

        """ Test 6 : Bouclage sur un même node => interdit. """
        for o_gredge in o_scene.get_gredges():
            t_id = o_gredge.o_edge.get_tid()
            if t_id == ((0, 0), (1, 0)) or t_id == ((1, 1), (2, 2)):
                o_gredge.setSelected(True)
            else:
                o_gredge.setSelected(False)
        o_scene.cross_edges()
        l_tid = sorted([o_gredge.o_edge.get_tid() for o_gredge in o_scene.get_gredges()])
        msg = "Le croisement produit un bouclage sur un même node => Cela n'est pas autorisé."
        assert l_tid == [((0, 0), (1, 0)), ((0, 0), (2, 1)), ((0, 1), (1, 2)), ((1, 1), (2, 2))], msg

        """ Fin. """
        self.connect(o_main.close, 1000)
        self.g_appli.exec()

Snippets

• Edges : Customisation

Bonjour les codeurs !

Commentaires & questions

Laissez vos commentaires et vos questions sur Youtube !