`io` -- 入出力ストリーム

このモジュールは、該当する CPython モジュールのサブセットを実装しています。詳しくはオリジナルの CPython ドキュメンテーションを参照してください: io.

このモジュールには stream (ファイルのような)オブジェクトの追加のデータ型とヘルパー関数が含まれています。

概念階層

CPython との違い

この章で説明するように、MicroPythonではストリームベースのクラスの概念的な階層が単純化されています。

(抽象)基底 stream クラスは、すべての具象クラスの動作の基礎となるものであり、CPython にある二分法(ペアワイズ分類)に固執しません。MicroPython では、これらは幾分単純化されていて、効率化とリソース節約のために暗黙的に役立ちます。

CPython の重要な二分法は、バッファされていないストリームとバッファされているストリームです。MicroPython では、すべてのストリームは現在バッファリングされていません。これは、すべての現代的な OS、さらには多くの RTOS やファイルシステムドライバーでさえも、すでにバッファリングが行われているからです。別のレイヤーにバッファリングを追加することは、生産性を低下させ("bufferbloat" という問題)、貴重なメモリを消費します。バッファリングが便利な場合もあるので、後でオプションのバッファリングサポートを導入するかもしれません。

しかし、CPython には別の重要な二分法もあり、それは「バッファされているか」と関連しています。ストリームが短い読込み/書出しを起こすかどうかです。短い読込みとは、たとえばストリームから10バイトを要求しても、それより少ないデータを得ることです。短い書出しについても同様です。CPython ではバッファリングされていないストリームは自動的に短い操作になる影響を受けますが、バッファされたストリームは要求したサイズを保証します。短くない読込み/書出しは重要な特性であり、より簡潔で効率的なプログラムを開発できるので、MicroPython にとっても非常に望ましいものです。したがって、MicroPython はバッファされたストリームをサポートしていませんが、短くない操作のストリームも提供します。短い操作の有無は、各クラスのニーズに応じて異なりますが、開発者は上記の理由により短くない操作を好むよう強く推奨します。たとえば、MicroPython のソケットは、短い読込み/書出しを避けることが保証されています。実際、現時点では、コアには短い操作のストリームクラスの例はなく、あるとすればポート固有のクラスとなります。そのような必要性はハードウェアの特性によって管理されます。

短くない操作の動作は、非ブロックストリームの場合はトリッキーであり、ブロックと非ブロックの動作は別の CPython での二分法です。この２つの動作は MicroPython で完全にサポートされています。非ブロックストリームは、データの到着または書出しを待つことは決してありません。可能な限りデータの読込み/書出しを行うか、データの欠乏(あるいはデータ書出し可能)の通知します。明らかに、これは「短くない操作」のポリシーと矛盾しています。実際、非ブロックバッファ(で短くない操作)のストリームは、CPython では面倒です。ものによって、そのような組み合わせは禁止されているか、未定義であり、文書化されていないものもあります。場合によっては冗長な例外が発生します。MicroPython では、非ブロックストリームが効率的な非同期操作にとって重要であるため、この特性は「短くない操作」に優先します。したがって、ストリームをブロックすると、可能な限り短い読込み/書出しが回避されます(ファイルの終わりに達した場合、またはエラーが発生した場合は短い読み込みを取得する唯一のケースですが、エラーは短いデータを返さず例外を発生させます)。非ブロックストリームは、動作をブロックしないように短いデータを生成する可能性があります。

残る二分法は、バイナリとテキストのストリームです。MicroPython はもちろんこれをサポートしていますが、CPython のテキストストリームが本質的にバッファされているのに対して MicroPython ではバッファしていません。(実際、これはバッファをサポートする可能正のある唯一のケースです)。

効率を上げるために、MicroPython は上記の階層に対応する抽象基底クラスを提供していないことに注意してください。ただし IOBase クラスをサブクラス化することで、純粋な Python でカスタムストリームオブジェクトを実装できます。

関数

io.open(name, mode='r', **kwargs): ファイルを開きます。組込みの open() 関数はこの関数の別名になります。すべての(ファイルシステムへのアクセスを提供する)ポートは、 mode パラメータをサポートする必要がありますが、他の引数のサポートはポートによって異なります。

クラス

class io.IOBase

Python でカスタムストリームオブジェクトを実装するための基底クラス。サブクラスは readinto, write, ioctl をオーバーライドして、 print(), json.dump(), select.poll(), ユーザーファイルシステム経由の open(), およびその他のストリームコンシューマーと連携するオブジェクトを作成できます。

CPython との違い

CPython で io.IOBase の API サーフェスははるかに広範です。一方、MicroPython の IOBase は最小限で、C ストリームインフラストラクチャが read(), readline(), seek(), close(), flush() などの標準メソッドを自動的に提供します。サブクラスは以下のメソッドのみを実装すれば済みます。

サブクラスは、以下のメソッドの一部または全部を実装します。Cストリーム層は、ユーザーコードが標準ストリーム関数を呼び出す際に、これらのメソッドを内部的に呼び出します。

readinto(buf): buf (呼び出し元がサイズが決められた bytearray)にデータを読み込みます。戻り値は読み込んだバイト数です。EOF の場合は 0、非ブロッキングストリームでデータが利用できない場合は None を返します。エラーが発生した場合は、負の errno 値(errno.EIO または 1 など)を返します。

write(buf): buf (bytearray)をストリームに書き出します。戻り値は書き出したバイト数です。非ブロッキングストリームがデータを受け入れられない場合は None を返します。エラーが発生した場合は、負の errno 値を返します。

ioctl(op, arg)

ストリームを制御し、そのプロパティを照会します。実行する操作は、以下の整数のいずれかである op で指定されます。

1 -- 書き込みバッファをフラッシュ(arg は未使用)

3 -- 準備完了をポーリング。 arg はチェックするイベントのビットマスクで、準備完了イベントのビットマスクを返します。ポーリングフラグは次のとおり:

0x0001 -- 読み取り可能なデータ

0x0004 -- ストリーム書き込み準備完了

0x0008 -- エラー状態

0x0010`` -- ハングアップ(例：接続が閉じた)

0x0020 -- 不正なリクエスト

4 -- ストリームを閉じる(arg は未使用)

11 -- 優先される読み取りバッファサイズ、または0を返します(arg は未使用)

ストリームのクローズをサポートするために、最低限 ioctl(4, ...) を処理する必要があります。ストリームを select.poll() または asyncio で使う場合、 ioctl(3, ...) を実装してください

高度な使用例向けに他の操作も存在します(2 = seek, 5 = timeout, 10 = fileno)。完全なリストについては py/stream.h を参照してください。

必ず整数を返してください。成功した場合は0を、サポートされていない操作の場合は1を返してください。(未処理の操作に対して0を返すと、Cレイヤーに対して操作が正常に処理されたと判断され、誤った動作を引き起こす可能性があります。)

class io.StringIO([string])

class io.BytesIO([string])

入出力のためのメモリ内の疑似ファイルオブジェクト。 StringIO はテキストモードの I/O に使用されます("t" 修飾子で開かれた通常のファイルと同様)。 BytesIO はバイナリモードの I/O に使用されます("b" 修飾子で開かれた通常のファイルと同様)。この疑似ファイルオブジェクトの初期の内容は string パラメータで指定できます(StringIO では通常の文字列、 BytesIO ではバイト列オブジェクトでなければなりません)。このオブジェクトについては、通常のファイルのメソッドのすべて read(), write(), seek(), flush(), close() が使え、加えて以下のメソッドが使えます。

getvalue(): データを保持するバッファの現在の内容を取得します。

class io.StringIO(alloc_size)

class io.BytesIO(alloc_size): alloc_size で指定したバイト数を保持するために事前に割り当てられた空の StringIO/BytesIO オブジェクトを作成します。つまり、その量のバイトを書き出してもバッファの再割り当てにはつながりません。したがって、メモリ不足の状況に遭遇したり、メモリの断片化につながることはありません。これらのコンストラクタは MicroPython の拡張であり、エンドユーザーアプリケーションではなく、特別な場合およびシステムレベルのライブラリーでのみ使用することをお勧めします。

CPython との違い

これらのコンストラクタは MicroPython の拡張です。

IOBase の使用例

出力をバッファに収集する最小限の書き込み専用ストリーム:

import io

class MyOutput(io.IOBase):
    def __init__(self):
        self.data = bytearray()

    def write(self, buf):
        self.data.extend(buf)
        return len(buf)

    def ioctl(self, op, arg):
        if op == 4:  # close
            return 0
        return -1

s = MyOutput()
print("hello", file=s)
print(s.data)  # bytearray(b'hello\n')

select.poll() で使用できる読み取り可能なストリーム:

import io, select
from micropython import const

_MP_STREAM_POLL = const(3)
_MP_STREAM_POLL_RD = const(0x0001)
_MP_STREAM_CLOSE = const(4)

class RingBuffer(io.IOBase):
    def __init__(self, size):
        self._buf = bytearray(size)
        self._size = size
        self._wpos = 0
        self._rpos = 0

    def _available(self):
        return (self._wpos - self._rpos) % self._size

    def put(self, data):
        for b in data:
            self._buf[self._wpos % self._size] = b
            self._wpos = (self._wpos + 1) % self._size

    def readinto(self, buf):
        n = min(len(buf), self._available())
        for i in range(n):
            buf[i] = self._buf[self._rpos % self._size]
            self._rpos = (self._rpos + 1) % self._size
        return n

    def ioctl(self, op, arg):
        if op == _MP_STREAM_POLL:
            if arg & _MP_STREAM_POLL_RD and self._available() > 0:
                return _MP_STREAM_POLL_RD
            return 0
        if op == _MP_STREAM_CLOSE:
            return 0
        return -1

rb = RingBuffer(64)
rb.put(b"test data")

poller = select.poll()
poller.register(rb, select.POLLIN)
for obj, flags in poller.poll(0):
    buf = bytearray(16)
    n = obj.readinto(buf)
    print(buf[:n])  # bytearray(b'test data')

io -- 入出力ストリーム

概念階層

関数

クラス

IOBase の使用例

`io` -- 入出力ストリーム