はじめに
前回の投稿でMSX0からファイルを指定してダウンロードするPythonプログラムを作成しました。
チューニングの結果、427バイトのテキストのダウンロードに42秒要していたのが12秒に短縮し、私(えぬしち)のチューニング欲はいったん満たされました。めでたしめでたし。
めでたしではない。
実際にMSX0からファイルをダウンロードするシーンを考えてみると、プログラムで生成された拡張子「DAT」のファイルを全部とってきたいとか、ファイル名が「02311」で始まるファイルを全部とってきたいという要求が追加になりますよね、きっと。
そういうわけで、今回はプログラムのワイルドカード対応に挑戦してみます。
※【免責】十分な試験を行っていませんので、まだまだバグが含まれていると思います。その点を理解いただいての使用をお願いします。実行した結果についての責任は一切負えません。
実装
前回までに作ったダウンロードプログラムをうまいこと活用しながら、ワイルドカード対応を実装していきます。
前回のプログラムは、ファイル名を指定してダウンロードする関数を用意していました。しかし関数名がいけていません。改名します(execute_full_process_telnetlib ⇒get_file_from_MSX0)。
また、複数回繰り返し実行されることを想定していません。関数を実行するたびにBASICプログラムをMSX0に送信していては資源の無駄遣いです。これはいけません。初回だけ実行されるように工夫を入れます。BASICプログラムに日付を付与して、初回チェックしましょう。
あとは細かい見直しを少々いれまして、改修版がこちら↓になります。
msx0babaget改修版
msx0babaget.py
import sys
import telnetlib
import time
import re
import base64
import os
from datetime import datetime
def is_valid_filename(filename):
# ファイル名が8文字+拡張子3文字であるかどうかをチェックする正規表現
pattern = r'^[A-Za-z0-9_]{1,8}\.[A-Za-z0-9_]{3}$'
return re.match(pattern, filename)
def create_new_file(file_name):
open(file_name, 'wb').close()
def decode_and_write_to_file(encoded_text, file_name):
decoded_data = base64.b64decode(encoded_text)
with open(file_name, 'ab') as file:
file.write(decoded_data)
def get_file_from_MSX0(host, program, program_ver, message_file_path,local_file_path):
start_time = time.time()
message_file_name = os.path.basename(message_file_path)
if not is_valid_filename(message_file_name):
return "Invalid file name format.", time.time() - start_time
try:
# Telnet接続
print("msx0babaput started connecting to msx0")
tn = telnetlib.Telnet(host, 2223)
tn.write(b'\r\n')
# DOSモードで起動している場合、BASICインタープリタモードに切り替える
response = tn.read_until(b'A>', timeout=1) # 1秒以内にプロンプトが表>示されない場合はタイムアウト
if b'A>' in response:
tn.write(b'basic\r\n') # BASICインタープリタモードに切り替え
tn.read_until(b'Ok')
# ファイルサイズを取得
oneline='OPEN"' + message_file_name + '"AS#1:PRINT LOF(1):CLOSE#1\r\n'
tn.write(oneline.encode('ascii'))
output = tn.read_until(b"Ok\r\n").decode('ascii').strip()
match = re.search(r'\s+(\d+)\s+', output)
message_file_size = 0
if match:
message_file_size = int(match.group(1))
else:
print("Can not get file size!")
return "Error", time.time() - start_time
print("Target file name:"+ message_file_name +" size:" + str(message_file_size))
# MSX0上のプログラムバージョン確認
tn.write(b'list 1\r\n') #行番号1を表示
response = tn.read_until(b'Ok', timeout=1) # 1秒以内にプロンプトが表>示されない場合はタイムアウト
if program_ver.encode() in response:
sys.stdout.write("already initializing.")
sys.stdout.flush()
print()
# BASICプログラムをBASICインタープリタに送信する(ファイル名のみ)
sys.stdout.write("Start building base64encoder on msx0 basic.")
sys.stdout.flush()
program = '20 LET F$ = "' + message_file_name + '"\r\n'#ファイル名を上書き
tn.write(program.encode('ascii') + b"\r\n") # BASICプログラムをサーバに送信
result = tn.read_until(b'\r\n\r\n')
sys.stdout.write("\rStart building base64encoder on msx0 basic..Done")
sys.stdout.flush()
print()
else:
# newコマンドで初期化
sys.stdout.write("Start initializing.")
sys.stdout.flush()
tn.write(b'new\r\n')
tn.read_until(b'Ok\r\n')
sys.stdout.write("\rStart initializing..Done.")
sys.stdout.flush()
print()
# BASICプログラムをBASICインタープリタに送信する
sys.stdout.write("Start building base64encoder on msx0 basic.")
sys.stdout.flush()
program = program + '20 LET F$ = "' + message_file_name + '"\r\n'#ファイル名を上書き
program = program + '1 \'' + program_ver + '\r\n'#プログラムバージョン記入
tn.write(program.encode('ascii') + b"\r\n") # BASICプログラムをサーバに送信
result = tn.read_until(b'\r\n\r\n')
sys.stdout.write("\rStart building base64encoder on msx0 basic..Done")
sys.stdout.flush()
print()
# BASICプログラムを実行する
print("Start transfer a target file on msx0 basic to local PC.")
create_new_file(local_file_path)
tn.write(b'run\r\n')
tn.read_until(b'run\r\n')
chunk_counter=0
total_chunks= message_file_size // 57 +1
# BASICプログラムの結果を1行ずつ読み取り、base64デコードしファイルに書き込む
while True:
sys.stdout.write("\rTransferring: {}/{} chunks".format(chunk_counter, total_chunks))
sys.stdout.flush()
result = tn.read_until(b'\n')
if "Ok\r\n" == result.decode('ascii'):
break
decode_and_write_to_file(result.decode('ascii').strip(),local_file_path)
chunk_counter += 1
result="Done"
print()
print(result)
print("Execution time: {} seconds".format(time.time() - start_time))
return result, time.time() - start_time
finally:
# Telnet接続を閉じる
tn.close()
if __name__ == "__main__":
if len(sys.argv) != 4:
print("Usage: python msx0babaget.py <host> <message_file_path> <local_file_path>")
sys.exit(1)
host = sys.argv[1]
message_file_path = sys.argv[2]
local_file_path = sys.argv[3]
# BASICプログラム
encoder_program = '''
5 CALL IOTPUT("msx/u0/pm/cpu/percent",200)\r\n
10 DEFINT A-Z:DIM B$,I$,X%,Y%,Z%,O%,P%,Q%,R%,OB%(4) :CLEAR 1000\r\n
20 F$ = "messages.txt"\r\n
30 B$ = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"\r\n
100 S = 3:OPEN F$ AS #1 LEN=S\r\n
110 L = LOF(1): IF L=0 THEN 450 ELSE C = (L-1) \ S\r\n
120 FIELD #1, 3 AS I$\r\n
130 FOR H=1 TO C\r\n
150 GET #1\r\n
160 X% = ASC(MID$(I$, 1, 1))\r\n
170 Y% = ASC(MID$(I$, 2, 1))\r\n
180 Z% = ASC(MID$(I$, 3, 1))\r\n
190 PRINT MID$(B$, X% \ 4 + 1, 1) + MID$(B$, (X% AND 3) * 16 + Y% \ 16 + 1, 1);\r\n
200 PRINT MID$(B$, (Y% AND 15) * 4 + Z% \ 64 + 1, 1) + MID$(B$, (Z% AND 63) + 1, 1);\r\n
210 IF H MOD 19 = 0 THEN PRINT \r\n
220 NEXT\r\n
300 S = (L - 1) MOD 3\r\n
320 GET #1\r\n
340 X% = ASC(MID$(I$, 1, 1))\r\n
350 Y% = ASC(MID$(I$, 2, 1))*(1+S=1)\r\n
360 Z% = ASC(MID$(I$, 3, 1))*(1+(S=1 OR S=2))\r\n
370 OB%(1) = X% \ 4\r\n
380 OB%(2) = (X% AND 3) * 16 + Y% \ 16\r\n
390 OB%(3) = (Y% AND 15) * 4 + Z% \ 64\r\n
400 OB%(4) = Z% AND 63\r\n
420 FOR J = 1 TO 4\r\n
430 IF J >= 3 AND S= 1 THEN PRINT"=";
ELSE IF J = 4 AND S= 2 THEN PRINT"="; ELSE PRINT MID$(B$, OB%(J) + 1, 1);\r\n
440 NEXT\r\n
450 PRINT \r\n
460 CLOSE #1\r\n
500 CALL IOTPUT("msx/u0/pm/cpu/percent",100)\r\n
999 END\r\n'''
# 全プロセスを実行
program_ver= datetime.now().strftime("%Y%m%d_%H%M%S")
get_file_from_MSX0(host, encoder_program, program_ver, message_file_path,local_file_path)
get_file_from_MSX0関数にワイルドカード指定で取得されたファイルを1つずつ渡して実行するイメージとなります。
msx0babaRget
ではget_file_from_MSX0を呼び出す部分を実装していきましょう。流れとしてはこうです。
１．実行時の引数で、ワイルドカード指定と保管先のフォルダ名を取得
２．MSX0からワイルドカード指定でファイル名リストを取得する
３．プログラム投入制御用にタイムスタンプ付きのバージョン文字を生成
４．リストからひとつづつファイル名を指定してダウンロード関数を実行し、ファイルを保管先のフォルダに保存する。
以上、簡単です。
msx0babaRget.py
import telnetlib
import sys
import os
import time
from datetime import datetime
from msx0babaget import get_file_from_MSX0
def get_file_list(host,target_file):
# Telnet接続
tn = telnetlib.Telnet(host ,2223)
tn.write(b'\r\n')
# DOSモードで起動している場合、BASICインタープリタモードに切り替える
response = tn.read_until(b'A>', timeout=1) # 1秒以内にプロンプトが表>示されない場合はタイムアウト
if b'A>' in response:
tn.write(b'basic\r\n') # BASICインタープリタモードに切り替え
tn.read_until(b'Ok')
# "FILES"コマンドを送信し、結果を取得
tn.write(b'FILES "' + target_file.encode() + b'"\r\n')
result = tn.read_until(b"Ok").decode("utf-8")
# ファイルリストを抽出
lines = result.split("\r\n")[2:-1]
file_list = [line.strip() for line in lines if line.strip()]
# Telnet接続を閉じる
tn.close()
return file_list
def parse_files(file_list):
parsed_files = []
for line in file_list:
line_length = len(line)
num_files = line_length // 13 + 1
for i in range(num_files):
file_name = line[i*13 : i*13 + 8].strip()
extension = line[i*13 + 9 : i*13 + 12].strip()
parsed_files.append((file_name, extension))
return parsed_files
def sort_and_print_files(parsed_files):
# ファイルリストをソート
sorted_files = sorted(parsed_files, key=lambda x: (x[0], x[1]))
# ソートされたファイルリストを出力
for file_name, extension in sorted_files:
print("{}.{}".format(file_name, extension))
if __name__ == "__main__":
if len(sys.argv) < 5 or sys.argv[3] != "-o":
print("Usage: python my_script.py host target_file -o output_path")
sys.exit(1)
host = sys.argv[1]
target_file = sys.argv[2]
output_path = sys.argv[4]
if not os.path.exists(output_path):
print("Error: Output path {} does not exist.".format(output_path))
sys.exit(1)
file_list = get_file_list(host, target_file)
parsed_files = parse_files(file_list)
# BASICプログラム
encoder_program = '''
5 CALL IOTPUT("msx/u0/pm/cpu/percent",200)\r\n
10 DEFINT A-Z:DIM B$,I$,X%,Y%,Z%,O%,P%,Q%,R%,OB%(4) :CLEAR 1000\r\n
20 F$ = "messages.txt"\r\n
30 B$ = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"\r\n
100 S = 3:OPEN F$ AS #1 LEN=S\r\n
110 L = LOF(1): IF L=0 THEN 450 ELSE C = (L-1) \ S\r\n
120 FIELD #1, 3 AS I$\r\n
130 FOR H=1 TO C\r\n
150 GET #1\r\n
160 X% = ASC(MID$(I$, 1, 1))\r\n
170 Y% = ASC(MID$(I$, 2, 1))\r\n
180 Z% = ASC(MID$(I$, 3, 1))\r\n
190 PRINT MID$(B$, X% \ 4 + 1, 1) + MID$(B$, (X% AND 3) * 16 + Y% \ 16 + 1, 1);\r\n
200 PRINT MID$(B$, (Y% AND 15) * 4 + Z% \ 64 + 1, 1) + MID$(B$, (Z% AND 63) + 1, 1);\r\n
210 IF H MOD 19 = 0 THEN PRINT \r\n
220 NEXT\r\n
300 S = (L - 1) MOD 3\r\n
320 GET #1\r\n
340 X% = ASC(MID$(I$, 1, 1))\r\n
350 Y% = ASC(MID$(I$, 2, 1))*(1+S=1)\r\n
360 Z% = ASC(MID$(I$, 3, 1))*(1+(S=1 OR S=2))\r\n
370 OB%(1) = X% \ 4\r\n
380 OB%(2) = (X% AND 3) * 16 + Y% \ 16\r\n
390 OB%(3) = (Y% AND 15) * 4 + Z% \ 64\r\n
400 OB%(4) = Z% AND 63\r\n
420 FOR J = 1 TO 4\r\n
430 IF J >= 3 AND S= 1 THEN PRINT"=";
ELSE IF J = 4 AND S= 2 THEN PRINT"="; ELSE PRINT MID$(B$, OB%(J) + 1, 1);\r\n
440 NEXT\r\n
450 PRINT \r\n
460 CLOSE #1\r\n
500 CALL IOTPUT("msx/u0/pm/cpu/percent",100)\r\n
999 END\r\n'''
# 全プロセスを実行
program_ver= datetime.now().strftime("%Y%m%d_%H%M%S")
time.sleep(1)
for file_name, extension in parsed_files:
result, execution_time = get_file_from_MSX0(host, encoder_program, program_ver,
"{}.{}".format(file_name, extension),"{}/{}.{}".format(output_path,file_name,extension))
time.sleep(1)
冗長な記載箇所がありますが、そこは追々見直しますのでご容赦を。
実行結果
nshichi@DESKTOP-BML9KQH:~/msx0-repo$ python3 msx0babaRget.py 192.168.0.2 "*.txt" -o temp
msx0babaput started connecting to msx0
Target file name:RET_TEST.TXT size:1076
Start initializing..Done.
Start building base64encoder on msx0 basic..Done
Start transfer a target file on msx0 basic to local PC.
Transferring: 19/19 chunks
Done
Execution time: 22.811453104019165 seconds
msx0babaput started connecting to msx0
Target file name:MESSAGES.TXT size:428
already initializing.
Start building base64encoder on msx0 basic..Done
Start transfer a target file on msx0 basic to local PC.
Transferring: 8/8 chunks
Done
Execution time: 8.420207738876343 seconds
msx0babaput started connecting to msx0
Target file name:ABCDEFG.TXT size:8
already initializing.
Start building base64encoder on msx0 basic..Done
Start transfer a target file on msx0 basic to local PC.
Transferring: 1/1 chunks
Done
Execution time: 2.5178210735321045 seconds
msx0babaput started connecting to msx0
Target file name:HONJITSU.TXT size:15
already initializing.
Start building base64encoder on msx0 basic..Done
Start transfer a target file on msx0 basic to local PC.
Transferring: 1/1 chunks
Done
Execution time: 2.7326531410217285 seconds
msx0babaput started connecting to msx0
Target file name:DDDD.TXT size:0
already initializing.
Start building base64encoder on msx0 basic..Done
Start transfer a target file on msx0 basic to local PC.
Transferring: 1/1 chunks
Done
Execution time: 2.4057939052581787 seconds
Done
無事、複数ファイルダウンロードできました。
と、ここまで書いた時点で記事は保存しておいたんですが、その後PCがお亡くなりになりまして、試験用ファイルの説明と実行結果の説明を記載しようと考えていたのですが、追記ができなくなってしまいました。
さよなら、愛機。新しいPCの環境整えたら、記事アップデートしてきます！
現場からは以上です！