# response_converter.py
from __future__ import annotations
import json
from io import BytesIO
from typing import Any
from PIL import Image
def pil_image_to_png_bytes(
image: Image.Image,
compress_level: int = 3,
) -> bytes:
"""
PIL.Image.ImageをPNGファイル形式のbytesへ変換する。
Parameters
----------
image:
PNGへ変換するPIL画像。
compress_level:
PNG圧縮レベル。0～9。
0は圧縮なしで高速・大容量。
9は高圧縮だが低速。
連続処理では1～3程度から試すのがおすすめ。
Returns
-------
bytes:
PNGファイル全体のバイト列。
"""
if not isinstance(image, Image.Image):
raise TypeError(
"imageはPIL.Image.Imageである必要があります。"
f" actual={type(image)}"
)
if not isinstance(compress_level, int):
raise TypeError(
"compress_levelはintである必要があります。"
)
if not 0 <= compress_level <= 9:
raise ValueError(
"compress_levelは0～9の範囲で指定してください。"
)
buffer = BytesIO()
image.save(
buffer,
format="PNG",
compress_level=compress_level,
)
return buffer.getvalue()
def create_response_data(
inference_results: list[dict[str, Any]],
request_id: int | str | None,
compress_level: int = 3,
) -> tuple[bytes, list[bytes]]:
"""
inference_processor.pyから返されたlist[dict]を、
C#送信用のJSONメタデータとPNGバイト列へ変換する。
想定する入力要素
----------------
{
"image": PIL.Image.Image,
"class_name": str,
"score": float,
"camera_number": int,
"result_index": int,
"request_id": int | str | None
}
Returns
-------
tuple[bytes, list[bytes]]:
1. UTF-8のJSONバイト列
2. PNG画像バイト列のリスト
"""
if not isinstance(inference_results, list):
raise TypeError(
"inference_resultsはlistである必要があります。"
f" actual={type(inference_results)}"
)
metadata_results: list[dict[str, Any]] = []
png_images: list[bytes] = []
for list_index, result in enumerate(inference_results):
if not isinstance(result, dict):
raise TypeError(
"inference_resultsの各要素はdictである必要があります。"
f" index={list_index}, actual={type(result)}"
)
image = _get_required_value(
result,
key="image",
expected_type=Image.Image,
result_index=list_index,
)
class_name = _get_required_value(
result,
key="class_name",
expected_type=str,
result_index=list_index,
)
score_value = _get_required_numeric_value(
result,
key="score",
result_index=list_index,
)
# dictにresult_indexがなければリスト位置を使用
result_index = int(
result.get(
"result_index",
list_index,
)
)
camera_number_value = result.get(
"camera_number",
None,
)
camera_number = (
int(camera_number_value)
if camera_number_value is not None
else None
)
png_bytes = pil_image_to_png_bytes(
image=image,
compress_level=compress_level,
)
png_images.append(png_bytes)
metadata_results.append(
{
"result_index": result_index,
"class_name": class_name,
"score": float(score_value),
"camera_number": camera_number,
# PNGバイナリ自体はJSONへ入れない
"image_index": list_index,
"image_format": "png",
"image_length": len(png_bytes),
# 参考情報。PNG内部にもサイズ情報は含まれる
"image_width": int(image.width),
"image_height": int(image.height),
"image_mode": str(image.mode),
}
)
metadata = {
"success": True,
"request_id": request_id,
"result_count": len(metadata_results),
"results": metadata_results,
"error": None,
}
json_bytes = json.dumps(
metadata,
ensure_ascii=False,
separators=(",", ":"),
).encode("utf-8")
return json_bytes, png_images
def create_error_response_data(
request_id: int | str | None,
error_code: str,
error_message: str,
) -> tuple[bytes, list[bytes]]:
"""
C#へ返すエラー応答を作成する。
エラー時は画像数0で返す。
"""
metadata = {
"success": False,
"request_id": request_id,
"result_count": 0,
"results": [],
"error": {
"code": str(error_code),
"message": str(error_message),
},
}
json_bytes = json.dumps(
metadata,
ensure_ascii=False,
separators=(",", ":"),
).encode("utf-8")
return json_bytes, []
def _get_required_value(
source: dict[str, Any],
key: str,
expected_type: type,
result_index: int,
) -> Any:
"""
dictから必須値を取得し、型を確認する。
"""
if key not in source:
raise KeyError(
f"推論結果に必須キー'{key}'がありません。"
f" result_index={result_index}"
)
value = source[key]
if not isinstance(value, expected_type):
raise TypeError(
f"推論結果の'{key}'の型が不正です。"
f" result_index={result_index},"
f" expected={expected_type},"
f" actual={type(value)}"
)
return value
def _get_required_numeric_value(
source: dict[str, Any],
key: str,
result_index: int,
) -> float:
"""
intまたはfloatの必須値を取得する。
NumPyのfloat32などが入る可能性がある場合でも、
float()で変換可能なら受け付ける。
"""
if key not in source:
raise KeyError(
f"推論結果に必須キー'{key}'がありません。"
f" result_index={result_index}"
)
value = source[key]
try:
return float(value)
except (TypeError, ValueError) as exception:
raise TypeError(
f"推論結果の'{key}'をfloatへ変換できません。"
f" result_index={result_index},"
f" actual={type(value)}"
) from exception
# communication_main.py
from __future__ import annotations
import json
import struct
import sys
import traceback
from io import BytesIO
from pathlib import Path
from typing import Any, BinaryIO
from PIL import Image, UnidentifiedImageError
from inference_processor import ImageInferenceProcessor
from response_converter import (
create_error_response_data,
create_response_data,
)
# 1つの受信画像に許可する最大サイズ。
# 異常値で巨大メモリを確保しないための上限。
MAX_INPUT_IMAGE_BYTES = 512 * 1024 * 1024
# JSONメタデータの最大サイズ。
MAX_JSON_BYTES = 10 * 1024 * 1024
# C#へ返すPNGの圧縮レベル。
# 連続処理ではまず1～3を推奨。
PNG_COMPRESS_LEVEL = 3
class EndOfInputError(Exception):
"""
C#側が標準入力を閉じた場合に使用する例外。
"""
class ProtocolError(Exception):
"""
通信プロトコルが不正な場合に使用する例外。
"""
def log(
message: str,
) -> None:
"""
ログはstdoutではなくstderrへ出力する。
stdoutはC#とのバイナリ通信専用なので、
print()を通常どおり使ってはいけない。
"""
print(
message,
file=sys.stderr,
flush=True,
)
def read_exact(
stream: BinaryIO,
byte_count: int,
) -> bytes:
"""
streamから指定されたバイト数を必ず読み取る。
read()は要求数より少ないデータを返す可能性があるため、
必要数に達するまで繰り返す。
"""
if byte_count < 0:
raise ValueError(
"byte_countは0以上である必要があります。"
)
if byte_count == 0:
return b""
buffer = bytearray()
while len(buffer) < byte_count:
chunk = stream.read(
byte_count - len(buffer)
)
if not chunk:
if len(buffer) == 0:
raise EndOfInputError(
"標準入力が閉じられました。"
)
raise EOFError(
"要求されたデータの途中で標準入力が閉じられました。"
f" expected={byte_count}, actual={len(buffer)}"
)
buffer.extend(chunk)
return bytes(buffer)
def read_uint64(
stream: BinaryIO,
) -> int:
"""
リトルエンディアンの符号なし64bit整数を読み取る。
"""
raw_bytes = read_exact(
stream,
8,
)
return struct.unpack(
"<Q",
raw_bytes,
)[0]
def write_uint64(
stream: BinaryIO,
value: int,
) -> None:
"""
リトルエンディアンの符号なし64bit整数を書き込む。
"""
if value < 0:
raise ValueError(
"valueは0以上である必要があります。"
)
stream.write(
struct.pack(
"<Q",
value,
)
)
def read_request() -> tuple[dict[str, Any], Image.Image | None]:
"""
C#から1要求を読み取る。
受信形式
--------
8 bytes : JSON長
N bytes : JSON本体
8 bytes : 入力画像長
M bytes : 入力画像本体
shutdown要求では画像長を0にできる。
Returns
-------
tuple:
request_metadata, PIL画像またはNone
"""
input_stream = sys.stdin.buffer
json_length = read_uint64(
input_stream
)
if json_length == 0:
raise ProtocolError(
"JSON長が0です。"
)
if json_length > MAX_JSON_BYTES:
raise ProtocolError(
"JSONメタデータが上限を超えています。"
f" length={json_length},"
f" max={MAX_JSON_BYTES}"
)
json_bytes = read_exact(
input_stream,
json_length,
)
try:
metadata = json.loads(
json_bytes.decode("utf-8")
)
except UnicodeDecodeError as exception:
raise ProtocolError(
"JSONメタデータをUTF-8として読み込めません。"
) from exception
except json.JSONDecodeError as exception:
raise ProtocolError(
"JSONメタデータの形式が不正です。"
) from exception
if not isinstance(metadata, dict):
raise ProtocolError(
"要求JSONのルートはobjectである必要があります。"
)
image_length = read_uint64(
input_stream
)
if image_length > MAX_INPUT_IMAGE_BYTES:
raise ProtocolError(
"入力画像が上限サイズを超えています。"
f" length={image_length},"
f" max={MAX_INPUT_IMAGE_BYTES}"
)
command = str(
metadata.get(
"command",
"process",
)
).lower()
if command == "shutdown":
# shutdown時は画像データを破棄する。
# 通常はimage_length=0を想定。
if image_length > 0:
read_exact(
input_stream,
image_length,
)
return metadata, None
if image_length == 0:
raise ProtocolError(
"process要求ですが入力画像長が0です。"
)
image_bytes = read_exact(
input_stream,
image_length,
)
try:
# BytesIOが閉じられても利用できるよう、
# load()後にcopy()して独立した画像にする。
with Image.open(
BytesIO(image_bytes)
) as opened_image:
opened_image.load()
input_image = opened_image.copy()
except UnidentifiedImageError as exception:
raise ProtocolError(
"受信データを画像として認識できません。"
) from exception
except OSError as exception:
raise ProtocolError(
"受信画像のデコードに失敗しました。"
) from exception
return metadata, input_image
def write_response(
json_bytes: bytes,
png_images: list[bytes],
) -> None:
"""
JSONメタデータと複数のPNGをC#へ送信する。
送信形式
--------
8 bytes : JSON長
N bytes : JSON本体
8 bytes : 画像枚数
画像枚数分：
8 bytes : PNG長
M bytes : PNG本体
"""
output_stream = sys.stdout.buffer
write_uint64(
output_stream,
len(json_bytes),
)
output_stream.write(
json_bytes
)
write_uint64(
output_stream,
len(png_images),
)
for png_bytes in png_images:
write_uint64(
output_stream,
len(png_bytes),
)
output_stream.write(
png_bytes
)
# C#側へすぐ届くよう必ずflushする
output_stream.flush()
def get_request_id(
metadata: dict[str, Any],
) -> int | str | None:
"""
request_idを取得する。
数値でも文字列でも許可する。
"""
return metadata.get(
"request_id",
None,
)
def get_camera_number(
metadata: dict[str, Any],
) -> int:
"""
JSONメタデータからcamera_numberを取得する。
"""
if "camera_number" not in metadata:
raise ProtocolError(
"要求JSONにcamera_numberがありません。"
)
camera_number_value = metadata["camera_number"]
# boolはintの派生型なので明示的に拒否する
if isinstance(camera_number_value, bool):
raise ProtocolError(
"camera_numberにboolは指定できません。"
)
try:
camera_number = int(
camera_number_value
)
except (TypeError, ValueError) as exception:
raise ProtocolError(
"camera_numberをintへ変換できません。"
) from exception
if camera_number < 0:
raise ProtocolError(
"camera_numberは0以上である必要があります。"
)
return camera_number
def run_server(
processor: ImageInferenceProcessor,
) -> None:
"""
C#からの要求を繰り返し処理する常駐ループ。
"""
log(
"Python画像推論プロセスの待受を開始しました。"
)
while True:
request_metadata: dict[str, Any] = {}
request_id: int | str | None = None
try:
request_metadata, input_image = (
read_request()
)
request_id = get_request_id(
request_metadata
)
command = str(
request_metadata.get(
"command",
"process",
)
).lower()
if command == "shutdown":
log(
"C#からshutdown要求を受信しました。"
)
shutdown_json = json.dumps(
{
"success": True,
"request_id": request_id,
"result_count": 0,
"results": [],
"shutdown": True,
"error": None,
},
ensure_ascii=False,
separators=(",", ":"),
).encode("utf-8")
write_response(
json_bytes=shutdown_json,
png_images=[],
)
break
if command != "process":
raise ProtocolError(
f"未対応のcommandです: {command}"
)
if input_image is None:
raise ProtocolError(
"process要求ですが画像がありません。"
)
camera_number = get_camera_number(
request_metadata
)
log(
"推論要求を受信しました。"
f" request_id={request_id},"
f" camera_number={camera_number},"
f" image_size={input_image.size},"
f" image_mode={input_image.mode}"
)
# inference_processor.pyの処理を呼び出す
inference_results = processor.process_image(
image=input_image,
camera_number=camera_number,
request_id=request_id,
)
# PIL.Image.ImageをPNGへ変換し、
# JSONメタデータと分離する
json_bytes, png_images = (
create_response_data(
inference_results=inference_results,
request_id=request_id,
compress_level=PNG_COMPRESS_LEVEL,
)
)
write_response(
json_bytes=json_bytes,
png_images=png_images,
)
log(
"推論結果を送信しました。"
f" request_id={request_id},"
f" result_count={len(png_images)}"
)
except EndOfInputError:
# C#が終了してstdinが閉じられた場合
log(
"標準入力が閉じられたため終了します。"
)
break
except Exception as exception:
# 詳細ログはstderrへ出す
traceback.print_exc(
file=sys.stderr
)
sys.stderr.flush()
error_code = (
"PROTOCOL_ERROR"
if isinstance(exception, ProtocolError)
else "PROCESSING_ERROR"
)
error_json, no_images = (
create_error_response_data(
request_id=request_id,
error_code=error_code,
error_message=str(exception),
)
)
try:
write_response(
json_bytes=error_json,
png_images=no_images,
)
except Exception:
# stdout自体が壊れている場合は継続できない
traceback.print_exc(
file=sys.stderr
)
sys.stderr.flush()
break
def main() -> int:
"""
Pythonプロセスのエントリーポイント。
"""
try:
script_directory = Path(
__file__
).resolve().parent
model_path = (
script_directory
/ "models"
/ "classifier.onnx"
)
log(
f"ONNXモデルを読み込みます: {model_path}"
)
# この時点でONNXモデルを1回だけ読み込む
processor = ImageInferenceProcessor(
model_path=model_path
)
log(
"ONNXモデルの読み込みが完了しました。"
)
run_server(
processor
)
log(
"Python画像推論プロセスを終了します。"
)
return 0
except Exception:
traceback.print_exc(
file=sys.stderr
)
sys.stderr.flush()
return 1
if __name__ == "__main__":
raise SystemExit(
main()
)