drlhp.pref_interface.PrefInterface 测试

from multiprocessing import Queue

import numpy as np
import termcolor

from drlhp.pref_db import Segment
from drlhp.pref_interface import PrefInterface

def send_segments(n_segments, seg_pipe):
    frame_stack = np.zeros((84, 84, 4))
    for i in range(n_segments):
        segment = Segment()
        for _ in range(25):
            segment.append(frame=frame_stack, reward=0)
        segment.finalise(seg_id=i)
        seg_pipe.put(segment)
        

---------------------------------------------------------------------------
ModuleNotFoundError                       Traceback (most recent call last)
Cell In[1], line 7
      4 import termcolor
      6 from drlhp.pref_db import Segment
----> 7 from drlhp.pref_interface import PrefInterface
      9 def send_segments(n_segments, seg_pipe):
     10     frame_stack = np.zeros((84, 84, 4))

File ~/work/pybook/pybook/doc/libs/drlhp/drlhp/pref_interface.py:14
     12 from select import select
     13 import numpy as np
---> 14 from .utils import VideoRenderer
     15 import logging
     16 from functools import partial

File ~/work/pybook/pybook/doc/libs/drlhp/drlhp/utils.py:7
      4 import time
      6 import numpy as np
----> 7 import pyglet
      8 import pdb
      9 import sys

ModuleNotFoundError: No module named 'pyglet'

p = PrefInterface(synthetic_prefs=True, max_segs=1000)
termcolor.cprint(p, 'red')

---------------------------------------------------------------------------
NameError                                 Traceback (most recent call last)
Cell In[2], line 1
----> 1 p = PrefInterface(synthetic_prefs=True, max_segs=1000)
      2 termcolor.cprint(p, 'red')

NameError: name 'PrefInterface' is not defined

检查片段是否正确存储在循环缓冲区中。

pi = PrefInterface(synthetic_prefs=True, max_segs=5)
pipe = Queue()
for i in range(5):
    pipe.put(i)
    pi.recv_segments(pipe)
np.testing.assert_array_equal(pi.segments, [0, 1, 2, 3, 4])
for i in range(5, 8):
    pipe.put(i)
    pi.recv_segments(pipe)
np.testing.assert_array_equal(pi.segments, [5, 6, 7, 3, 4])
for i in range(8, 11):
    pipe.put(i)
    pi.recv_segments(pipe)
np.testing.assert_array_equal(pi.segments, [10, 6, 7, 8, 9])

---------------------------------------------------------------------------
NameError                                 Traceback (most recent call last)
Cell In[3], line 1
----> 1 pi = PrefInterface(synthetic_prefs=True, max_segs=5)
      2 pipe = Queue()
      3 for i in range(5):

NameError: name 'PrefInterface' is not defined