オシロスコープ上で動くゲームを作る方法

概要

オシロスコープを使って画像や動画を描いてみたい！
と思った方に向け、この記事を書いていきます。
最終的には以下の画像のように、絵をオシロスコープ上に描くことができます。

方法

1 用意するもの

• オシロスコープ
• Pythonの開発環境
• National Instruments (NI USB-6212)
• コントローラ(JC-U3613M)

開発環境

• Python3.8.6
• ライブラリ
  • nidaqmx
  • numpy
  • matplotlib
  • pygame
  • time
  • Image
• 使用したソフト
  • VScode editor

import numpy as np
import matplotlib.pyplot as plt
from PIL import Image
import time
from nidaqmx import constants
from nidaqmx.constants import LineGrouping, AcquisitionType, Edge
import nidaqmx
from nidaqmx import *
import pygame
from pygame.locals import *

2 手順

1 初期設定を行う

'''initiation(初期設定)'''
global tTask1
tTask1=nidaqmx.Task() #アナログ出力用のタスク
tTask1.__init__("main_VC_controller")#タスク名の割り当て

2 動かしたい画像を用意する（なるべく正方形のものが良い）

3 画像を白黒に変換し、０と１の配列にする

def convert_image_to_binary_matrix(size, image_path, threshold=128):
    # 画像を読み込み、グレースケールに変換
    image = Image.open(image_path).convert('L')
    # 画像を256x256にリサイズ
    image = image.resize((size, size))

    # 画像をNumPy配列に変換
    image_array = np.array(image)

    # 二値化：閾値以上を1、未満を0に設定
    binary_matrix = np.where(image_array < threshold, 1, 0)

    return binary_matrix

4 配列のうち、１の座標を読み込む

今回は飛行機のレースゲームを作ったため、飛行機の画像とコースの画像の二つを用意する

air1_matrix = convert_image_to_binary_matrix(64, r"飛行機画像へのパス")
road_matrix = convert_image_to_binary_matrix(256, r"コース画像へのパス")
plane1 = np.argwhere(air1_matrix == 1)
road = np.argwhere(road_matrix == 1)

5 固定画像のx-y座標をVx-Vyに変換する

Vx0 = []
Vy0 = []
for coordinate in road:
    Vx0.append(4*(0.5-coordinate[0] / 256 ))
    Vy0.append(4*(0.5-coordinate[1] / 256 ))
road0 = np.array([Vx0, Vy0])

6 動く画像の関数を作成する

def points(coordinates, theta, x=0, y=0):
    Vx = []
    Vy = []
    for coordinate in coordinates:
        Vx.append(6*((0.5-coordinate[1] / 64 ) * np.cos(theta*np.pi/180) - (0.5-coordinate[0] / 64 ) * np.sin(theta*np.pi/180)) - x)
        Vy.append(8*((0.5-coordinate[1] / 64 ) * np.sin(theta*np.pi/180) + (0.5-coordinate[0] / 64 ) * np.cos(theta*np.pi/180)) - y)
    D2 = np.array([Vx, Vy])
    return D2

coordinateは今回の場合plane1、thetaは画像の向き、x=0, y=0は画像の中心の座標

7 信号出力のコードを書く

''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''
'''サンプル波形のパラメータ'''''''''''''''''''''''''''''''''''''''''''''
samps_rate = 200000 #サンプリングレート
N = 10000 #バッファサイズ
n = 1 #三角関数の周期を変えるための整数
data_size = 1000 #1サンプルの配列データ要素数(データサイズ)
required_value=100 #コールバック関数をトリガするための端末読み取りデータ数

#初期波形の定義
t= np.linspace( -2*np.pi, 2*np.pi, num=data_size, endpoint=True )
Vx = 3 * np.cos(t) / (1 + np.sin(t)**2)
Vy = 3 * np.cos(t) * np.sin(t) / (1 + np.sin(t)**2)


'''出力するサンプル波形*(numpyの1次元ndarray配列によって指定)'''


sample_2D__ = np.array([ Vx, Vy ])
on_off=True
global samp_Writer
k = 0
j = 0

'''コールバック関数の定義'''''''''''''''''''''''''''''''''''''''''''''
def askUser():
    global on_off
    input("Press return to stop")
    on_off = False

def callback1(task_handle, status, callback_data=1):
    print('コールバックしました')
    return 0

def callback2(task_idx, event_type, num_samples, callback_data):
    global Vx
    global Vy
    global sample_2D__
    global k
    global j
    #波形の出力
    
    while True:
        print(f"コールバック関数{j}に移動しました。")
        for i in range(10):
            sample_2D__ = points(plane1, T, X, Y)
            samp_Writer.write_many_sample(sample_2D__)
        j = j + 1
        time.sleep(1)
  
    k += 1
    time.sleep(1)
    print(k)
    return 0

'''アナログ出力信号のタスクを構成(DAQ内の構成)'''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''
tTask1.ao_channels.add_ao_voltage_chan( physical_channel='Dev1/ao0:1',
                                        name_to_assign_to_channel="AO-code", 
                                        min_val=-10, max_val=10 )#タスク1
tTask1._out_stream.output_onbrd_buf_size = 4095
tTask1.output_buf_size = N #バッファサイズ
tTask1.timing.cfg_samp_clk_timing(rate=samps_rate,
                                  sample_mode=AcquisitionType.CONTINUOUS,
                                  samps_per_chan=N)#有限サンプルの出力タイミング
#tTask1.timing.cfg_samp_clk_timing(rate=samps_rate, sample_mode=AcquisitionType.CONTINUOUS)#連続サンプルの出力タイミング

#連続のサンプルの出力タイミングを設定。samps_per_chanによって、バッファサイズを決定する
#test_Task.timing.cfg_samp_clk_timing( rate=samps_rate, sample_mode=AcquisitionType.CONTINUOUS_SAMPLES, samps_per_chan=N )


'''波形の生成(PC→DAQの構成)'''''''''''''''''''''


#波形出力の自動スタート、つまり、垂れ流し状態。生成されたそばから出力している
samp_Writer = nidaqmx.stream_writers.AnalogMultiChannelWriter(tTask1.out_stream, auto_start=False)

#第一引数に指定したサンプル数がPCバッファ→DAQデバイスに書き込まれたときに受信するコールバック関数を登録する
tTask1.register_every_n_samples_transferred_from_buffer_event(required_value, callback2)
#tTask1.register_every_n_samples_transferred_from_buffer_event(1000, callback2)

#これを行うとき、明示的なタスクの開始でなくてはならず、auto_startは使用不可である
samp_Writer.write_many_sample(sample_2D__)

'''実行フェーズ'''

print('開始準備中...')
time.sleep(3)

print('タスクを開始')
tTask1.start()#タスク1の開始

8 コントローラで画像を動かせるようにする

ここで、XYは中心の座標、Tは角度を表す

X = 0
Y = 0
T = 0

try:
    while True:
        for e in pygame.event.get():
            if e.type == QUIT:
                active = False

            if e.type == pygame.locals.JOYHATMOTION:
                if e.value[0] == 0:
                    if e.value[1] == 1:
                        print("十字キー：上")
                        T = 0
                        for i in range (10):
                            Y -= 1/10
                            sample_2D__ = points(plane1, T, X, Y)
                            samp_Writer.write_many_sample(sample_2D__)
                        print(X, Y)

                    elif e.value[1] == 0:
                        print("十字キー：ー")
                    else:
                        print("十字キー：下")
                        T = 180
                        for i in range (10):
                            Y += 1/10
                            sample_2D__ = points(plane1, T, X, Y)
                            samp_Writer.write_many_sample(sample_2D__)
                        print(X, Y)
                elif e.value[0] == 1:
                    if e.value[1] == 0:
                        print("十字キー：右")
                        T = -90
                        for i in range (10):
                            X -= 1/10
                            sample_2D__ = points(plane1, T, X, Y)
                            samp_Writer.write_many_sample(sample_2D__)
                        print(X, Y)
                    elif e.value[1] == 1:
                        print("十字キー：右上")
                        T = -45
                        for i in range (10):
                            X -= 1/10
                            Y -= 1/10
                            sample_2D__ = points(plane1, T, X, Y)
                            samp_Writer.write_many_sample(sample_2D__)
                        print(X, Y)
                    else:
                        print("十字キー：右下")
                        T = -135
                        for i in range (10):
                            X -= 1/10
                            Y += 1/10
                            sample_2D__ = points(plane1, T, X, Y)
                            samp_Writer.write_many_sample(sample_2D__)
                        print(X, Y)
                else:
                    if e.value[1] == 0:
                        print("十字キー：左")
                        T = 90
                        for i in range (10):
                            X += 1/10
                            sample_2D__ = points(plane1, T, X, Y)
                            samp_Writer.write_many_sample(sample_2D__)
                        print(X, Y)
                    elif e.value[1] == 1:
                        print("十字キー：左上")
                        T = 45
                        for i in range (10):
                            X += 1/10
                            Y -= 1/10
                            sample_2D__ = points(plane1, T, X, Y)
                            samp_Writer.write_many_sample(sample_2D__)
                        print(X, Y)
                    else:
                        print("十字キー：左下")
                        T = 135
                        for i in range (10):
                            X += 1/10
                            Y += 1/10
                            sample_2D__ = points(plane1, T, X, Y)
                            samp_Writer.write_many_sample(sample_2D__)
                        print(X, Y)
                        

            elif e.type == pygame.locals.JOYAXISMOTION:
                if e.axis == 0 or e.axis == 1:
                    if e.value == 0.0 or e.value == 1.0 or e.value == -1.0:
                        print('ジョイスティック(左) ー {0} {1}'.format(e.axis, e.value))
                    else:
                        print('ジョイスティック(左)  {0} {1}'.format(e.axis, e.value))
                elif  e.axis == 2 or e.axis == 3:
                    if e.value == 0.0 or e.value == 1.0 or e.value == -1.0:
                        print('ジョイスティック(右) ー {0} {1}'.format(e.axis, e.value))
                    else:
                        print('ジョイスティック(右)  {0} {1}'.format(e.axis, e.value))
                else:
                    print('ジョイスティック(ー)')
            elif e.type == pygame.locals.JOYBUTTONDOWN:
                print('ボタン押す {0}'.format(e.button))
            elif e.type == pygame.locals.JOYBUTTONUP:
                print('ボタン離す {0}'.format(e.button))

9 タスク停止のためのコードを書く

except KeyboardInterrupt:
    # Ctrl+Cが押された場合、ループを終了
    '''停止フェーズ'''

    #print(tTask1.is_task_done())#タスクが終わっているかどうかをTure or Falseで値を返す
    print('タスクの停止まで待機しています...')
    tTask1.wait_until_done(timeout=5)#タスクの停止までの待機時間(有限サンプルモードのときにして指定する)


    #print(tTask1.is_task_done())
    print('タスクの停止')
    tTask1.stop()#タスクを停止

    print('タスク1をクリアしました')
    tTask1.close()#生成したタスクをクリア

以上の工程により、オシロスコープ上で自由に画像を動かすことができるようになる

まとめ

最後にコード全体を記しておく

import numpy as np
import matplotlib.pyplot as plt
from PIL import Image
import time
from nidaqmx import constants
from nidaqmx.constants import LineGrouping, AcquisitionType, Edge
import nidaqmx
from nidaqmx import *
import pygame
from pygame.locals import *

pygame.init()
pygame.joystick.init()
try:
   joystick = pygame.joystick.Joystick(0)
   joystick.init()
   print('ジョイスティックの名前:', joystick.get_name())
   print('ボタン数 :', joystick.get_numbuttons())
except pygame.error:
   print('ジョイスティックが接続されていません')

'''initiation(初期設定)'''
global tTask1
tTask1=nidaqmx.Task() #アナログ出力用のタスク
tTask1.__init__("main_VC_controller")#タスク名の割り当て


def convert_image_to_binary_matrix(size, image_path, threshold=128):
    # 画像を読み込み、グレースケールに変換
    image = Image.open(image_path).convert('L')
    # 画像を256x256にリサイズ
    image = image.resize((size, size))

    # 画像をNumPy配列に変換
    image_array = np.array(image)

    # 二値化：閾値以上を1、未満を0に設定
    binary_matrix = np.where(image_array < threshold, 1, 0)

    return binary_matrix


air1_matrix = convert_image_to_binary_matrix(64, r"C:\python_venv\test1\air-1.png")
road_matrix = convert_image_to_binary_matrix(256, r"C:\python_venv\test1\コース2.png")
plane1 = np.argwhere(air1_matrix == 1)
road = np.argwhere(road_matrix == 1)

Vx0 = []
Vy0 = []
for coordinate in road:
    Vx0.append(4*(0.5-coordinate[0] / 256 ))
    Vy0.append(4*(0.5-coordinate[1] / 256 ))
road0 = np.array([Vx0, Vy0])


def points(coordinates, theta, x=0, y=0):
    Vx = []
    Vy = []
    for coordinate in coordinates:
        Vx.append(6*((0.5-coordinate[1] / 64 ) * np.cos(theta*np.pi/180) - (0.5-coordinate[0] / 64 ) * np.sin(theta*np.pi/180)) - x)
        Vy.append(8*((0.5-coordinate[1] / 64 ) * np.sin(theta*np.pi/180) + (0.5-coordinate[0] / 64 ) * np.cos(theta*np.pi/180)) - y)
    D2 = np.array([Vx, Vy])
    return D2



'''
''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''
image_path = 'C:/python_code_image/Lenna.bmp' 
threshold = 128#閾値
maxval = 255 #8ビット階調
bmp_array = np.array(Image.open(image_path).convert('L'))#指定したパスに対してグレースケール変換
bmp_array = (bmp_array > threshold) * maxval
'''

''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''
'''サンプル波形のパラメータ'''''''''''''''''''''''''''''''''''''''''''''
samps_rate = 200000 #サンプリングレート
N = 10000 #バッファサイズ
n = 1 #三角関数の周期を変えるための整数
data_size = 1000 #1サンプルの配列データ要素数(データサイズ)
required_value=100 #コールバック関数をトリガするための端末読み取りデータ数

#初期波形の定義
t= np.linspace( -2*np.pi, 2*np.pi, num=data_size, endpoint=True )
Vx = 3 * np.cos(t) / (1 + np.sin(t)**2)
Vy = 3 * np.cos(t) * np.sin(t) / (1 + np.sin(t)**2)


'''出力するサンプル波形*(numpyの1次元ndarray配列によって指定)'''


sample_2D__ = np.array([ Vx, Vy ])
on_off=True
global samp_Writer
k = 0
j = 0

'''コールバック関数の定義'''''''''''''''''''''''''''''''''''''''''''''
def askUser():
    global on_off
    input("Press return to stop")
    on_off = False

def callback1(task_handle, status, callback_data=1):
    print('コールバックしました')
    return 0

def callback2(task_idx, event_type, num_samples, callback_data):
    global Vx
    global Vy
    global sample_2D__
    global k
    global j
    #波形の出力
    
    while True:
        print(f"コールバック関数{j}に移動しました。")
        for i in range(10):
            sample_2D__ = points(plane1, T, X, Y)
            samp_Writer.write_many_sample(sample_2D__)
        j = j + 1
        time.sleep(1)
  
    k += 1
    time.sleep(1)
    print(k)
    return 0

'''アナログ出力信号のタスクを構成(DAQ内の構成)'''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''
tTask1.ao_channels.add_ao_voltage_chan( physical_channel='Dev1/ao0:1',
                                        name_to_assign_to_channel="AO-code", 
                                        min_val=-10, max_val=10 )#タスク1
tTask1._out_stream.output_onbrd_buf_size = 4095
tTask1.output_buf_size = N #バッファサイズ
tTask1.timing.cfg_samp_clk_timing(rate=samps_rate,
                                  sample_mode=AcquisitionType.CONTINUOUS,
                                  samps_per_chan=N)#有限サンプルの出力タイミング
#tTask1.timing.cfg_samp_clk_timing(rate=samps_rate, sample_mode=AcquisitionType.CONTINUOUS)#連続サンプルの出力タイミング

#連続のサンプルの出力タイミングを設定。samps_per_chanによって、バッファサイズを決定する
#test_Task.timing.cfg_samp_clk_timing( rate=samps_rate, sample_mode=AcquisitionType.CONTINUOUS_SAMPLES, samps_per_chan=N )


'''波形の生成(PC→DAQの構成)'''''''''''''''''''''


#波形出力の自動スタート、つまり、垂れ流し状態。生成されたそばから出力している
samp_Writer = nidaqmx.stream_writers.AnalogMultiChannelWriter(tTask1.out_stream, auto_start=False)

#第一引数に指定したサンプル数がPCバッファ→DAQデバイスに書き込まれたときに受信するコールバック関数を登録する
tTask1.register_every_n_samples_transferred_from_buffer_event(required_value, callback2)
#tTask1.register_every_n_samples_transferred_from_buffer_event(1000, callback2)

#これを行うとき、明示的なタスクの開始でなくてはならず、auto_startは使用不可である
samp_Writer.write_many_sample(sample_2D__)

'''実行フェーズ'''

print('開始準備中...')
time.sleep(3)

print('タスクを開始')
tTask1.start()#タスク1の開始

X = 0
Y = 0
T = 0

try:
    while True:
        for e in pygame.event.get():
            if e.type == QUIT:
                active = False

            if e.type == pygame.locals.JOYHATMOTION:
                if e.value[0] == 0:
                    if e.value[1] == 1:
                        print("十字キー：上")
                        T = 0
                        for i in range (10):
                            Y -= 1/10
                            sample_2D__ = points(plane1, T, X, Y)
                            samp_Writer.write_many_sample(sample_2D__)
                        print(X, Y)

                    elif e.value[1] == 0:
                        print("十字キー：ー")
                    else:
                        print("十字キー：下")
                        T = 180
                        for i in range (10):
                            Y += 1/10
                            sample_2D__ = points(plane1, T, X, Y)
                            samp_Writer.write_many_sample(sample_2D__)
                        print(X, Y)
                elif e.value[0] == 1:
                    if e.value[1] == 0:
                        print("十字キー：右")
                        T = -90
                        for i in range (10):
                            X -= 1/10
                            sample_2D__ = points(plane1, T, X, Y)
                            samp_Writer.write_many_sample(sample_2D__)
                        print(X, Y)
                    elif e.value[1] == 1:
                        print("十字キー：右上")
                        T = -45
                        for i in range (10):
                            X -= 1/10
                            Y -= 1/10
                            sample_2D__ = points(plane1, T, X, Y)
                            samp_Writer.write_many_sample(sample_2D__)
                        print(X, Y)
                    else:
                        print("十字キー：右下")
                        T = -135
                        for i in range (10):
                            X -= 1/10
                            Y += 1/10
                            sample_2D__ = points(plane1, T, X, Y)
                            samp_Writer.write_many_sample(sample_2D__)
                        print(X, Y)
                else:
                    if e.value[1] == 0:
                        print("十字キー：左")
                        T = 90
                        for i in range (10):
                            X += 1/10
                            sample_2D__ = points(plane1, T, X, Y)
                            samp_Writer.write_many_sample(sample_2D__)
                        print(X, Y)
                    elif e.value[1] == 1:
                        print("十字キー：左上")
                        T = 45
                        for i in range (10):
                            X += 1/10
                            Y -= 1/10
                            sample_2D__ = points(plane1, T, X, Y)
                            samp_Writer.write_many_sample(sample_2D__)
                        print(X, Y)
                    else:
                        print("十字キー：左下")
                        T = 135
                        for i in range (10):
                            X += 1/10
                            Y += 1/10
                            sample_2D__ = points(plane1, T, X, Y)
                            samp_Writer.write_many_sample(sample_2D__)
                        print(X, Y)
                        

            elif e.type == pygame.locals.JOYAXISMOTION:
                if e.axis == 0 or e.axis == 1:
                    if e.value == 0.0 or e.value == 1.0 or e.value == -1.0:
                        print('ジョイスティック(左) ー {0} {1}'.format(e.axis, e.value))
                    else:
                        print('ジョイスティック(左)  {0} {1}'.format(e.axis, e.value))
                elif  e.axis == 2 or e.axis == 3:
                    if e.value == 0.0 or e.value == 1.0 or e.value == -1.0:
                        print('ジョイスティック(右) ー {0} {1}'.format(e.axis, e.value))
                    else:
                        print('ジョイスティック(右)  {0} {1}'.format(e.axis, e.value))
                else:
                    print('ジョイスティック(ー)')
            elif e.type == pygame.locals.JOYBUTTONDOWN:
                print('ボタン押す {0}'.format(e.button))
            elif e.type == pygame.locals.JOYBUTTONUP:
                print('ボタン離す {0}'.format(e.button))

except KeyboardInterrupt:
    # Ctrl+Cが押された場合、ループを終了
    '''停止フェーズ'''

    #print(tTask1.is_task_done())#タスクが終わっているかどうかをTure or Falseで値を返す
    print('タスクの停止まで待機しています...')
    tTask1.wait_until_done(timeout=5)#タスクの停止までの待機時間(有限サンプルモードのときにして指定する)


    #print(tTask1.is_task_done())
    print('タスクの停止')
    tTask1.stop()#タスクを停止

    print('タスク1をクリアしました')
    tTask1.close()#生成したタスクをクリア