Désobfuscation du code de calcul des séquences

million/analyze/find_holes.py

@@ -9,44 +9,32 @@ import million.analyze.message_evaluation as msg_ev
 
 def compute_sequences(messages: List[Message], accepted_max: int = 1_000_000) -> List[Sequence]:
     sequences: List[Sequence] = []
-    current_sequence = Sequence(
-        start_message=messages[0],
-        end_message=messages[0]
-    )
-    for i in range(1, len(messages)):
-        message = messages[i]
-        message_value = msg_ev.compute(message)
-        if message_value > accepted_max:
-            continue
-        if message_value - current_sequence.end() == 1:
-            current_sequence.end_message = message
+    current = Sequence(start_message=messages[0])
+    
+    for message in messages[1:]:
+        if msg_ev.compute(message) > accepted_max: continue
+
+        if msg_ev.compute(message) == current.end() + 1:
+            current.end_message = message
         else:
-            sequences.append(current_sequence)
-            current_sequence = Sequence(
-                start_message=message,
-                end_message=message
-            )
+            sequences.append(current)
+            current = Sequence(start_message=message)
 
     # order the sequences by start
     sequences.sort(key=lambda s: s.start())
 
     merged_sequences: List[Sequence] = []
-    current_sequence = sequences[0]
-    for i in range(1, len(sequences)):
-        sequence = sequences[i]
-        sequence_start_is_in_current_sequence = current_sequence.start() <= sequence.start() and current_sequence.end() >= sequence.start()
-        sequence_end_is_further = sequence.end() > current_sequence.end()
-        sequence_start_is_current_end_or_next = sequence.start() == current_sequence.end() + 1
-
-        if sequence_start_is_in_current_sequence or sequence_start_is_current_end_or_next:
-            if sequence_end_is_further:
-                current_sequence.end_message = sequence.end_message
+    previous = sequences[0]
+    
+    for sequence in sequences[1:]:
+        if previous.overlaps(sequence):
+            previous.merge(sequence)
         else:
-            merged_sequences.append(current_sequence)
-            current_sequence = sequence
+            merged_sequences.append(previous)
+            previous = sequence
 
     # Having merged the sequences once, any sequence having start = end can be removed
-    return [s for s in merged_sequences if s.start() != s.end()]
+    return [s for s in merged_sequences if s.length() > 1]
 
 
 def find_holes(messages: List[Message], accepted_max: int = 1_000_000) -> List[Hole]:

million/model/sequence.py

@@ -1,5 +1,7 @@
+from __future__ import annotations
 
 from pydantic import BaseModel
+import pydantic
 
 from million.model.message import Message
 import million.analyze.message_evaluation as msg_ev
@@ -7,10 +9,25 @@ import million.analyze.message_evaluation as msg_ev
 
 class Sequence(BaseModel):
     start_message: Message
-    end_message: Message | None
+    end_message: Message | None = None
+
+    @pydantic.validator('end_message', pre=True, always=True)
+    def default_end_message(cls, v, *, values):
+        return v or values['start_message'] 
 
     def start(self) -> int:
         return msg_ev.compute(self.start_message)
     
     def end(self) -> int:
-        return msg_ev.compute(self.end_message)
+        return msg_ev.compute(self.end_message)
+    
+    def length(self) -> int:
+        return self.end() - self.start() + 1
+    
+    def merge(self, other: Sequence) -> None:
+        if other.start() < self.start(): self.start_message = other.start_message
+        if other.end() > self.end(): self.end_message = other.end_message
+    
+    def overlaps(self, other: Sequence) -> bool:
+        return self.start() <= other.end() + 1 and \
+            other.start() <= self.end() + 1