8: Python文字列パイプ（文字列をシリアルに運ぶ）

<このシリーズの前の記事 | このシリーズの目次 | このシリーズの次の記事>

シリアルなのは、さもなければ、メモリが使い果たされてしまうかもしれないから。'java.io.PipedWriter' + 'java.io.PipedReader'に対する、Python相当物。

---

話題

About: Pythonプログラミング言語

---

この記事の目次

• 開始コンテキスト
• ターゲットコンテキスト
• オリエンテーション
• 本体
• 1: 大きすぎる文字列の全体はメモリにロードできない
• 2: Javaには、'java.io.PipedWriter' + 'java.io.PipedReader'がある
• 3: Pythonにはない. . .
• 4: The Plan　計画
• 5: コード
• 6: 1つの使用例と1つの実行結果

---

開始コンテキスト

• 読者は、Pythonプログラミング言語の基本的知識を持っている。

---

ターゲットコンテキスト

• 読者は、任意の文字列をシリアルに運ぶ、ある文字列パイプを知る。

---

オリエンテーション

任意の文字列をシリアルに運ぶ、ある文字列パイプがここで紹介されます。それが必要なのは、メモリが使い果たされるのを防ぐ（コード構造を歪めることなく）ためです。

本文字列パイプは、前記事で紹介されたオブジェクト群パイプのサブクラスです。

Java版、C++版、C#版も以後、作成されます。

---

本体

---

1: 大きすぎる文字列の全体はメモリにロードできない

Hypothesizer 7
文字列は、小さければ、メモリに普通に文字列オブジェクトとしてロードできる。

それが大きすぎれば、その方法は、メモリを使い果たしてしまい、結果は、悪性の高いものになり（スラッシングと呼ばれる）、システム全体を立ち往生させてしまう。その状況はすでに、別のシリーズのある記事にて詳しく議論された。

文字列消費ロジックを文字列生成ロジックの中に織り込むことによってコード構造を歪めることを甘受するのであれば、メモリスペースの枯渇は防げるだろうが、それを私は甘受しない: 文字列生成ロジックは純粋に文字列を生成し、文字列消費ロジックは純粋に文字列を消費すべきである。

---

2: Javaには、'java.io.PipedWriter' + 'java.io.PipedReader'がある

Hypothesizer 7
実は、Javaには、'java.io.PipedWriter' + 'java.io.PipedReader'があり、それは、任意の文字列をシリアルに運ぶ、パイプのサイズだけのメモリスペースを使用して。

問題は解決した、Javaにとっては！

---

3: Pythonにはない. . .

Hypothesizer 7
私は、それに相当するものがPythonにもあるだろうと思っていた、. . .が、見つけることができなかった。ふーむ. . .

---

4: The Plan　計画

Hypothesizer 7
したがって、私には、自らの文字列パイプを作成するしか選択肢がない。

実際には、オブジェクト群パイプを既に私は持っていて、それを、文字列パイプとして使用できる、なぜなら、文字列は文字群のシーケンスであって、文字群はオブジェクト群だから。実際には、Pythonには文字タイプがないから（それは賢明でないと私は思う）、'str'タイプが各文字を表わさなければならないだろう、それは、非効率であるが、我慢しなければならないだろう。

しかしながら、いくつかの追加のメソッド（1行を読むもののような）が、文字列パイプには便利であろう、したがって、私は、当該オブジェクト群クラスを拡張しよう。

---

5: コード

Hypothesizer 7
以下がそのコードだ（mypyアノテーションを含んでいる（私のどのPython コードもそうであるように））。

theBiasPlanet/coreUtilities/pipes/StringPipe.py

@Python ソースコード

from typing import List
from typing import Optional
from typing import TextIO
import sys
from theBiasPlanet.coreUtilities.constantsGroups.GeneralConstantsConstantsGroup import GeneralConstantsConstantsGroup
from theBiasPlanet.coreUtilities.inputsHandling.NoMoreDataException import NoMoreDataException
from theBiasPlanet.coreUtilities.inputsHandling.NoMoreNeedsException import NoMoreNeedsException
from theBiasPlanet.coreUtilities.pipes.ObjectsPipe import ObjectsPipe
from theBiasPlanet.coreUtilities.timersHandling.TimeOutException import TimeOutException

class StringPipe (ObjectsPipe [str]):
	def __init__ (a_this: "StringPipe", a_bufferLength: int, a_notificationIsDelayed: bool) -> None:
		
		super ().__init__ (a_bufferLength, a_notificationIsDelayed)
	
	def __del__ (a_this: "StringPipe") -> None:
		None
	
	def writeWholeString (a_this: "StringPipe", a_reader: TextIO) -> None:
		l_writtenCharacter: Optional [str] = None
		while True:
			l_writtenCharacter = a_reader.read (1)
			if l_writtenCharacter == "":
				break
			try:
				a_this.write (l_writtenCharacter)
			except (NoMoreNeedsException) as l_exception:
				break
	
	def readString (a_this: "StringPipe", a_maximumLength: int, a_timeOutPeriodInMilliseconds: int = -1) -> str:
		l_readCharacter: Optional [str] = None
		l_readStringBuilder: List [str] = []
		l_readStringLength: int = 0
		while l_readStringLength < a_maximumLength:
			try:
				if l_readStringLength == 0:
					l_readCharacter = a_this.read (a_timeOutPeriodInMilliseconds)
				else:
					l_readCharacter = a_this.read ()
			except (NoMoreDataException, TimeOutException) as l_exception:
				if l_readStringLength == 0:
					raise l_exception
				break
			if l_readCharacter is not None:
				l_readStringBuilder.append (l_readCharacter)
				l_readStringLength = l_readStringLength + 1
		return "".join (l_readStringBuilder)
	
	def readStringLine (a_this: "StringPipe", a_maximumLength: int, a_timeOutPeriodInMilliseconds: int = -1) -> str:
		l_readStringBuilder: List [str] = []
		l_readStringLength: int = 0
		while True:
			try:
				l_readCharacter: str = a_this.read (a_timeOutPeriodInMilliseconds)
				if l_readCharacter is not None:
					l_readStringBuilder.append (l_readCharacter)
					l_readStringLength = l_readStringLength + 1
					if l_readCharacter == GeneralConstantsConstantsGroup.c_newLineCharacter or l_readStringLength >= a_maximumLength:
						break
			except (NoMoreDataException, TimeOutException) as l_exception:
				if l_readStringLength == 0:
					raise l_exception
				break
		return "".join (l_readStringBuilder)
	
	def readWholeString (a_this: "StringPipe") -> str:
		l_readCharacter: Optional [str] = None
		l_readStringBuilder: List [str] = []
		while True:
			try:
				l_readCharacter = a_this.read ()
			except (NoMoreDataException) as l_exception:
				break
			if l_readCharacter is not None:
				l_readStringBuilder.append (l_readCharacter)
		return "".join (l_readStringBuilder)

theBiasPlanet/coreUtilities/constantsGroups/GeneralConstantsConstantsGroup.py

@Python ソースコード

~

class GeneralConstantsConstantsGroup:
	~
	c_newLineCharacter: str = '\n' # char
	~

そのコードを強固にテストしたわけではない（今のところは）ことをここに注釈しておこう、それをいくつかのケース（その内の1つが次セクションで示される）で使用はしているが。

---

6: 1つの使用例と1つの実行結果

Hypothesizer 7
以下は、1つの使用例だ。

@Python ソースコード

from typing import List
from io import StringIO
import sys
from threading import Thread
from theBiasPlanet.coreUtilities.inputsHandling.NoMoreDataException import NoMoreDataException
from theBiasPlanet.coreUtilities.inputsHandling.NoMoreNeedsException import NoMoreNeedsException
from theBiasPlanet.coreUtilities.pipes.StringPipe import StringPipe

class Test1Test:
	@staticmethod
	def main (a_arguments: List [str]) -> None:
		Test1Test.test3 ()
	
	@staticmethod
	def test3 () -> None:
		l_stringPipe: "StringPipe" = StringPipe (16, True)
		def l_subThreadFunction () -> None:
			try:
				Test1Test.writeString (l_stringPipe)
			except (Exception) as l_exception:
				None
			finally:
				try:
					l_stringPipe.finishWriting ()
				except (Exception) as l_exception:
					None
		l_subThread: Thread = Thread (target = l_subThreadFunction)
		l_subThread.start ()
		Test1Test.readString (l_stringPipe)
		l_subThread.join ()
	
	@staticmethod
	def writeString (a_writer: StringPipe) -> None:
		l_iterationIndex: int = 0
		for l_iterationIndex in range (0, 512, 1):
			l_string: str = str (l_iterationIndex) + "\n"
			try:
				a_writer.writeWholeString (StringIO (l_string));
			except (NoMoreNeedsException) as l_exception:
				break
			sys.stdout.write ("### written: {0:s}\n".format (l_string))
			sys.stdout.flush ()
	
	@staticmethod
	def readString (a_reader: "StringPipe") ->None:
		l_maximumReadStringLength: int = 10
		while True:
			try:
				l_string: str = a_reader.readStringLine (l_maximumReadStringLength)
			except (NoMoreDataException) as l_exception:
				break
			sys.stdout.write ("### read: {0:s}\n".format (l_string))
			sys.stdout.flush ()

if __name__ == "__main__":
	Test1Test.main (sys.argv)

以下が、1つのアウトプットだ。

@出力

### written: 0

### written: 1

### read: 0

### read: 1

### read: 2

### written: 2

### written: 3

### written: 4

### written: 5

### written: 6

### written: 7

### written: 8

### written: 9

### read: 3

### read: 4

### read: 5

### read: 6

### read: 7

### read: 8

### read: 9

### written: 10

### written: 11

### written: 12

### written: 13

### written: 14

### written: 15

### read: 10

### read: 11

### read: 12

### read: 13

### read: 14

### read: 15

~
### written: 496

### read: 493

### read: 494

### read: 495

### read: 496

### written: 497

### written: 498

### written: 499

### written: 500

### read: 497

### read: 498

### read: 499

### read: 500

### written: 501

### written: 502

### written: 503

### written: 504

### read: 501

### read: 502

### read: 503

### read: 504

### written: 505

### written: 506

### written: 507

### written: 508

### read: 505

### read: 506

### read: 507

### read: 508

### written: 509

### written: 510

### written: 511

### read: 509

### read: 510

### read: 511

---

参考資料

<このシリーズの前の記事 | このシリーズの目次 | このシリーズの次の記事>