Update rngtool.py

src/rngtool.py

@@ -1,4 +1,8 @@
-from typing import List, Tuple
+"""[summary]
+
+"""
+from typing import List
+from typing import Tuple
 import time
 import cv2
 from xorshift import Xorshift
@@ -12,7 +16,7 @@ def randrange(r,mi,ma):
     t = (r & 0x7fffff) / 8388607.0
     return t * mi + (1.0 - t) * ma
 
-def tracking_blink(img, roi_x, roi_y, roi_w, roi_h)->Tuple[List[int],List[int],float]:
+def tracking_blink(img, roi_x, roi_y, roi_w, roi_h, size = 40)->Tuple[List[int],List[int],float]:
     """measuring the type and interval of player's blinks
 
     Returns:
@@ -20,7 +24,7 @@ def tracking_blink(img, roi_x, roi_y, roi_w, roi_h)->Tuple[List[int],List[int],f
     """
 
     eye = img
-    
+
     video = cv2.VideoCapture(0,cv2.CAP_DSHOW)
     video.set(cv2.CAP_PROP_FRAME_WIDTH,1920)
     video.set(cv2.CAP_PROP_FRAME_HEIGHT,1080)
@@ -29,7 +33,6 @@ def tracking_blink(img, roi_x, roi_y, roi_w, roi_h)->Tuple[List[int],List[int],f
     state = IDLE
     blinks = []
     intervals = []
-    prev_match = 0
     prev_time = 0
 
     prev_roi = None
@@ -38,24 +41,25 @@ def tracking_blink(img, roi_x, roi_y, roi_w, roi_h)->Tuple[List[int],List[int],f
     offset_time = 0
 
     # 瞬きの観測
-    while len(blinks)<40 or state!=IDLE:
-        ret, frame = video.read()
+    while len(blinks)<size or state!=IDLE:
+        _, frame = video.read()
         time_counter = time.perf_counter()
-        unix_time = time.time()
         roi = cv2.cvtColor(frame[roi_y:roi_y+roi_h,roi_x:roi_x+roi_w],cv2.COLOR_RGB2GRAY)
-        if (roi==prev_roi).all():continue
-        prev_roi = roi        
+        if (roi==prev_roi).all():
+            continue
+
+        prev_roi = roi
         res = cv2.matchTemplate(roi,eye,cv2.TM_CCOEFF_NORMED)
         _, match, _, _ = cv2.minMaxLoc(res)
-        cv2.imshow("",roi)
-        cv2.waitKey(1)
+
         if 0.01<match<0.85:
             if state==IDLE:
                 blinks.append(0)
                 interval = (time_counter - prev_time)/1.018
                 interval_round = round(interval)
                 intervals.append(interval_round)
-                if len(intervals)==40:
+
+                if len(intervals)==size:
                     offset_time = time_counter
 
                 #check interval 
@@ -71,7 +75,7 @@ def tracking_blink(img, roi_x, roi_y, roi_w, roi_h)->Tuple[List[int],List[int],f
                     state = DOUBLE
             elif state==DOUBLE:
                 pass
-            
+
         if state!=IDLE and time_counter - prev_time>0.7:
             state = IDLE
             print(debug_txt)
@@ -98,23 +102,21 @@ def tracking_poke_blink(img, roi_x, roi_y, roi_w, roi_h, size = 60)->Tuple[List[
     prev_roi = None
     prev_time = time.perf_counter()
 
-    offset_time = 0
 
     # 瞬きの観測
     while len(intervals)<size:
-        ret, frame = video.read()
+        _, frame = video.read()
         time_counter = time.perf_counter()
 
         roi = cv2.cvtColor(frame[roi_y:roi_y+roi_h,roi_x:roi_x+roi_w],cv2.COLOR_RGB2GRAY)
-        if (roi==prev_roi).all():continue
+        if (roi==prev_roi).all():
+            continue
         prev_roi = roi
 
         res = cv2.matchTemplate(roi,eye,cv2.TM_CCOEFF_NORMED)
         _, match, _, _ = cv2.minMaxLoc(res)
 
         if 0.4<match<0.85:
-            if len(intervals)==1:
-                offset_time = unix_time
             if state==IDLE:
                 interval = (time_counter - prev_time)
                 print(interval)
@@ -134,8 +136,8 @@ def recov(blinks:List[int],rawintervals:List[int])->Xorshift:
     Recover the xorshift from the type and interval of blinks.
 
     Args:
-        blinks (List[int]): 
-        intervals (List[int]): 
+        blinks (List[int]):
+        intervals (List[int]):
 
     Returns:
         List[int]: internalstate
@@ -158,7 +160,59 @@ def recov(blinks:List[int],rawintervals:List[int])->Xorshift:
     result.getNextRandSequence(advanced_frame)
     return result
 
+def reidentifyByBlinks(rng:Xorshift, observed_blinks:List[int], search_min=0, search_max=10**6)->Xorshift:
+    if search_max<search_min:
+        search_min, search_max = search_max, search_min
+    search_range = search_max - search_min
+    observed_len = len(observed_blinks)
+    if 2**observed_len < search_range:
+        return None
+
+    identify_rng = Xorshift(*rng.getState())
+    rands = [(i, r&0xF) for i,r in enumerate(identify_rng.getNextRandSequence(search_max))]
+    blinkrands = [(i, r) for i,r in rands if r&0b1110==0]
+
+    #prepare
+    expected_blinks_lst = []
+    expected_blinks = 0
+    lastblink_idx = -1
+    mask = (1<<observed_len)-1
+    for idx, r in blinkrands[:observed_len]:
+        lastblink_idx = idx
+
+        expected_blinks <<= 1
+        expected_blinks |= r
+
+    expected_blinks_lst.append((lastblink_idx, expected_blinks))
+
+    for idx, r in blinkrands[observed_len:]:
+        lastblink_idx = idx
+
+        expected_blinks <<= 1
+        expected_blinks |= r
+        expected_blinks &= mask
+
+        expected_blinks_lst.append((lastblink_idx, expected_blinks))
+
+    #search
+    search_blinks = calc.list2bitvec(observed_blinks)
+    result = Xorshift(rng.getState())
+    for idx, blinks in expected_blinks_lst:
+        if search_blinks==blinks and search_min <= idx:
+            result.getNextRandSequence(idx)
+            return result
+
+    return None
+
 def recovByMunchlax(rawintervals:List[float])->Xorshift:
+    """Recover the xorshift from the interval of Munchlax blinks.
+
+    Args:
+        rawintervals (List[float]): [description]
+
+    Returns:
+        Xorshift: [description]
+    """
     advances = len(rawintervals)
     intervals = [interval+0.048 for interval in rawintervals]#観測結果のずれを補正
     intervals = intervals[1:]#最初の観測結果はノイズなのでそれ以降を使う