Bug and compatibility fixes

.vscode/settings.json

@@ -0,0 +1,5 @@
+{
+    "githubPullRequests.ignoredPullRequestBranches": [
+        "master"
+    ]
+}

Plan/DEMO-EvArea-cv2-mp-np.py

@@ -1,94 +1,114 @@
-import cv2
-import time
-import multiprocessing as mp
-import numpy as np
-import numpy.random as rd
-import itertools
+"""Simple multiprocessing life-area simulation with OpenCV display."""
 
-'''Yonv1943 2018-05-20 10:10:02'''
+import itertools
+import multiprocessing as mp
+import time
 
+import cv2
+import numpy as np
 
-class Information(object):
-    room = 4  # seed room
-    seat = 3  # area seat
-    side = 48  # area side
-    size = 4  # vida multi-processing threading number
 
-    explore = {  # vida, area (x, y)
-        0: np.array([+1, +0]),
-        1: np.array([+0, +1]),
-        2: np.array([-1, +0]),
-        3: np.array([+0, -1]),
-    }
-    '''index'''
-    ib, ig, ir, di = 0, 1, 2, 3  # color: [blue, green, red]
-    li = 0  # life
-    tu = 3  # turn
+ROOM = 4
+SEAT = 3
+SIDE = 48
+QUEUE_SIZE = 4
 
+INDEX_BLUE = 0
+INDEX_GREEN = 1
+INDEX_RED = 2
+INDEX_END = 3
+INDEX_LIFE = 0
 
-i = Information()
-rd.seed(int(time.time()))
+EXPLORE = {
+    0: np.array([1, 0]),
+    1: np.array([0, 1]),
+    2: np.array([-1, 0]),
+    3: np.array([0, -1]),
+}
 
 
-def seed(m1, m2):
-    m2 = m1
-    return m1, m2
+def spread_seed(source, target):
+    """Return updated source and target values."""
+    target = source.copy()
+    return source, target
 
 
-def vida(area):
-    for (x1, y1) in itertools.product(range(1, i.side - 1), range(1, i.side - 1)):
-        if area[x1, y1][i.li] <= 0.0:  # life == 0, continue
+def update_area(area, rng):
+    """Update the simulation area by spreading values to neighbors."""
+    for x_pos, y_pos in itertools.product(range(1, SIDE - 1), range(1, SIDE - 1)):
+        if area[x_pos, y_pos, INDEX_LIFE] <= 0.0:
             continue
 
-        (x2, y2) = i.explore[rd.randint(4)] + (x1, y1)
-        m1, m2 = area[x1, y1], area[x2, y2]
-        m1, m2 = seed(m1, m2)
+        next_x, next_y = EXPLORE[rng.integers(4)] + np.array([x_pos, y_pos])
+        source = area[x_pos, y_pos].copy()
+        target = area[next_x, next_y].copy()
+
+        source, target = spread_seed(source, target)
+        area[x_pos, y_pos] = source
+        area[next_x, next_y] = target
 
-        area[x1, y1], area[x2, y2] = m1, m2
     return area
 
 
-def p_vida(q_img, side):
-    area = np.zeros((side, side, i.seat), dtype=np.float)
+def producer(queue_img, side):
+    """Generate simulation frames and put them into the queue."""
+    rng = np.random.default_rng()
+    area = np.zeros((side, side, SEAT), dtype=float)
 
-    '''plant'''
-    seed_num = int(side ** 2 / i.room ** 2)
-    for (x, y) in rd.randint(low=0, high=i.side, size=(seed_num, 2)):
-        area[x, y] = rd.uniform(low=0.0, high=1.0, size=i.seat)
+    seed_count = int(side ** 2 / ROOM ** 2)
+    seed_positions = rng.integers(low=0, high=side, size=(seed_count, 2))
+
+    for x_pos, y_pos in seed_positions:
+        area[x_pos, y_pos] = rng.uniform(low=0.0, high=1.0, size=SEAT)
 
     while True:
-        area = vida(area)
-        q_img.put(area)
+        area = update_area(area, rng)
+        if not queue_img.full():
+            queue_img.put(area.copy())
 
 
-def p_view(q_img, window_name):
+def viewer(queue_img, window_name):
+    """Display simulation frames with OpenCV."""
     cv2.namedWindow(window_name, flags=cv2.WINDOW_KEEPRATIO)
-    timer0 = time.time()
+    previous_time = time.time()
 
     while True:
-        area = q_img.get()
-        show = np.array(area[:, :, i.ib:i.di])
-        show = np.array(show * (255.0 / np.max(show)), dtype=np.uint8)
+        area = queue_img.get()
+        show = area[:, :, INDEX_BLUE:INDEX_END]
+
+        max_value = np.max(show)
+        if max_value > 0:
+            show = (show * (255.0 / max_value)).astype(np.uint8)
+        else:
+            show = np.zeros_like(show, dtype=np.uint8)
+
         cv2.imshow(window_name, show)
-        cv2.waitKey(1)
+        if cv2.waitKey(1) & 0xFF == 27:
+            break
+
+        current_time = time.time()
+        print(f"||| Ave time: {current_time - previous_time:.6f}")
+        previous_time = current_time
+
+    cv2.destroyAllWindows()
 
-        timer1 = time.time()
-        print("||| Ave time:", timer1 - timer0)
-        timer0 = timer1
 
 def main():
-    mp.set_start_method('spawn')
-    queue_img = mp.Queue(maxsize=4)
+    """Start producer and viewer processes."""
+    queue_img = mp.Queue(maxsize=QUEUE_SIZE)
 
-    process_l = [
-        mp.Process(target=p_vida, args=(queue_img, i.side)),
-        mp.Process(target=p_view, args=(queue_img, 'EvArea')),
+    processes = [
+        mp.Process(target=producer, args=(queue_img, SIDE)),
+        mp.Process(target=viewer, args=(queue_img, "EvArea")),
     ]
 
-    [p.start() for p in process_l]
-    [p.join() for p in process_l]
+    for process in processes:
+        process.start()
+
+    for process in processes:
+        process.join()
 
 
-if __name__ == '__main__':
-    main()
-pass
+if __name__ == "__main__":
+    mp.set_start_method("spawn")
+    main()

Plan/Demo_show_dymanic_images_in_colab.py

@@ -1,40 +1,60 @@
-from IPython.display import clear_output
-import matplotlib.pyplot as plt
-import numpy.random as rd
+"""Render a random LunarLander episode frame by frame in a notebook."""
+
 from time import sleep
 
-"""Github: Yonv1943 2021-03-03
-Demo_show_dymanic_images_in_colab.py
-"""
+import gym
+import matplotlib.pyplot as plt
+from IPython.display import clear_output
+from pyvirtualdisplay import Display
 
 
-import gym  # pip3 install gym==0.17 pyglet==1.5.0  # env.render() bug in gym==0.18, pyglet==1.6
-# env = gym.make('CartPole-v0')
-env = gym.make('LunarLander-v2')
+MAX_STEPS = 1000
+DISPLAY_SIZE = (400, 300)
+FRAME_DELAY = 0.01
 
-from pyvirtualdisplay import Display
-display = Display(visible=0, size=(400, 300))
-display.start()
 
-observation = env.reset()
-cum_reward = 0
-frames = []
-for t in range(1000):
-    frames.append(env.render(mode='rgb_array'))
-    action = env.action_space.sample()
-    observation, reward, done, info = env.step(action)
-    if done:
-        break
+def play_random_lunar_lander():
+    """Run one random episode and display the captured frames."""
+    display = Display(visible=0, size=DISPLAY_SIZE)
+    display.start()
+
+    env = gym.make("LunarLander-v2")
+
+    try:
+        reset_result = env.reset()
+        if isinstance(reset_result, tuple):
+            observation, _ = reset_result
+        else:
+            observation = reset_result
+
+        frames = []
+
+        for _ in range(MAX_STEPS):
+            frames.append(env.render(mode="rgb_array"))
+
+            action = env.action_space.sample()
+            step_result = env.step(action)
+
+            if len(step_result) == 5:
+                observation, reward, terminated, truncated, info = step_result
+                done = terminated or truncated
+            else:
+                observation, reward, done, info = step_result
 
-for frame in frames:
-    # action = rd.uniform(-1, 1, size=action_dim)
-    # state, reward, done, _ = env.step(action)
+            if done:
+                break
 
-    # image = env.render(mode='rgb_array')
+        plt.figure(figsize=(6, 4))
+        for frame in frames:
+            plt.imshow(frame)
+            plt.axis("off")
+            plt.show()
+            sleep(FRAME_DELAY)
+            clear_output(wait=True)
 
-    plt.imshow(frame)
-    plt.show()
-    sleep(0.01)
-    clear_output(wait=True)
+    finally:
+        env.close()
+        display.stop()
 
 
+play_random_lunar_lander()