アップストリームとダウンストリーム¶
ここではiscp-pyを使用してアップストリームとダウンストリームを行うサンプルを示します
環境¶
python 3.10
事前準備¶
本サンプルを動作させるための準備をします。 まず、アップストリーム用のノードとダウンストリーム用のノードを作成します。 ノードはintdashのエッジとマッピングされます。intdashAPIや管理コンソールを使用して作成してください。 既に作成したノードがある場合はそのノードIDとシークレットを使用しても問題ありません。 作成したノードのIDとシークレットを環境に合わせて下記に示す値を環境変数としてセットします。
[1]:
import os
URL_API = os.getenv("URL_API", "https://example.com")
DOWN_NODE_ID = os.getenv("DOWN_NODE_ID", "downstream-node-id")
DOWN_NODE_SECRET = os.getenv("DOWN_NODE_SECRET", "xxxxxxx")
UP_NODE_ID = os.getenv("UP_NODE_ID", "upstream-node-id")
UP_NODE_SECRET = os.getenv("UP_NODE_SECRET", "xxxxxxx")
アクセストークンの発行¶
アップストリーム用のアクセストークンと、ダウンストリーム用のアクセストークンをそれぞれ発行します。 ノードのIDとシークレットはOAuth2のクライアントクレデンシャル方式によるトークン発行ができます。
[2]:
import string
import requests
from oauthlib.oauth2 import BackendApplicationClient
from requests_oauthlib import OAuth2Session
_URL_TOKEN_ENDPOINT = f"{URL_API}/api/auth/oauth2/token"
def token_client_credentials(edge_id: str, secret: str) -> str:
"""
トークンの払い出しコード
"""
oauth = OAuth2Session(client=BackendApplicationClient(client_id=edge_id))
oauth.fetch_token(
token_url=_URL_TOKEN_ENDPOINT,
client_id=edge_id,
client_secret=secret,
include_client_id=True,
)
return oauth.token["access_token"]
[3]:
down_token=token_client_credentials(DOWN_NODE_ID, DOWN_NODE_SECRET)
[4]:
up_token=token_client_credentials(UP_NODE_ID, UP_NODE_SECRET)
アップストリームを行うコードの定義¶
アップストリームを行うコードのサンプルです。 このサンプルでは、トランスポートにWebSocketを使用して基準時刻のメタデータと、文字列型のデータポイントをiSCPサーバーへ送信しています。
[5]:
import iscp
import asyncio
import urllib
_URL_API = urllib.parse.urlparse(URL_API)
_ENABLE_TLS=_URL_API.scheme == "https"
_ISCP_PORT = _URL_API.port if _URL_API.port else 443 if _URL_API.scheme == "https" else 80
_ADDRESS_WEBSOCKET= f"{_URL_API.hostname}:{_ISCP_PORT}"
started_downstream = asyncio.Event()
async def start_upstream():
async with await iscp.Conn.connect(address=_ADDRESS_WEBSOCKET, connector=iscp.WebSocketConnector(enable_tls=_ENABLE_TLS), token_source=lambda: up_token, node_id=UP_NODE_ID) as conn:
async with await conn.open_upstream(session_id="session_id") as up:
await started_downstream.wait()
start = iscp.DateTime.utcnow()
await conn.send_base_time(
iscp.BaseTime(
name="manual",
elapsed_time=1_000_000,
base_time=start,
priority=1000,
session_id="session_id",
)
)
for i in range(10):
await asyncio.sleep(1)
await up.write_data_points(
iscp.DataID(name="greeting", type="string"),
iscp.DataPoint(
elapsed_time=iscp.DateTime.utcnow().unix_nano() - start.unix_nano(),
payload=b"hello",
),
)
await up.write_data_points(
iscp.DataID(name="end", type="string"),
iscp.DataPoint(
elapsed_time=iscp.DateTime.utcnow().unix_nano() - start.unix_nano(),
payload=b"end",
),
)
print("closed upstream")
print("closed upstream connection")
ダウンストリームを行うコードの定義¶
前述のアップストリームで送信されたデータをダウンストリームで受信するコードのサンプルです。
このサンプルでは、アップストリーム開始のメタデータ、基準時刻のメタデータ、 文字列型のデータポイントを受信しています。
[6]:
import iscp
import json
import dataclasses
async def start_downstream():
async with await iscp.Conn.connect(address=_ADDRESS_WEBSOCKET, connector=iscp.WebSocketConnector(enable_tls=_ENABLE_TLS), token_source=lambda: down_token, node_id=DOWN_NODE_ID) as conn:
async with await conn.open_downstream(
filters=[
iscp.DownstreamFilter(
source_node_id=UP_NODE_ID,
data_filters=[iscp.DataFilter(name="#", type="#")],
)
]
) as down:
started_downstream.set()
metadata = await down.metadatas().__anext__() # 基準時刻の受信
print(f"Received Metadata: {metadata}")
async for msg in down.chunks(): # データポイントの受信
print(f"Received DownstreamChunk: {msg}")
for data_point_group in msg.data_point_groups:
if data_point_group.data_id.name == "end":
print("Received end message")
return
実行¶
[7]:
import asyncio
await asyncio.gather(start_downstream(), start_upstream())
# 本サンプルはJupyter Notebookから実行しています。
# 通常の利用シーンでは、下記のように非同期処理を実行する必要があることにご注意ください。
# import asyncio
# async def main():
# await asyncio.gather(start_downstream(), start_upstream())
# if __name__ == "__main__":
# asyncio.run(main())
Received Metadata: DownstreamMetadata(source_node_id='8366f4b1-dc22-4e6d-b22b-dbe2beb92518', metadata=UpstreamOpen(stream_id=UUID('67bbc25d-f4d2-421d-b049-85a73c9a358b'), session_id='session_id', qos=<QoS.UNRELIABLE: 'UNRELIABLE'>))
Received DownstreamChunk: DownstreamChunk(upstream_info=UpstreamInfo(session_id='session_id', stream_id=UUID('67bbc25d-f4d2-421d-b049-85a73c9a358b'), source_node_id='8366f4b1-dc22-4e6d-b22b-dbe2beb92518'), sequence_number=1, data_point_groups=[DataPointGroup(data_id=DataID(name='greeting', type='string'), data_points=[DataPoint(elapsed_time=1001316426, payload=b'hello')])])
Received DownstreamChunk: DownstreamChunk(upstream_info=UpstreamInfo(session_id='session_id', stream_id=UUID('67bbc25d-f4d2-421d-b049-85a73c9a358b'), source_node_id='8366f4b1-dc22-4e6d-b22b-dbe2beb92518'), sequence_number=2, data_point_groups=[DataPointGroup(data_id=DataID(name='greeting', type='string'), data_points=[DataPoint(elapsed_time=2002286993, payload=b'hello')])])
Received DownstreamChunk: DownstreamChunk(upstream_info=UpstreamInfo(session_id='session_id', stream_id=UUID('67bbc25d-f4d2-421d-b049-85a73c9a358b'), source_node_id='8366f4b1-dc22-4e6d-b22b-dbe2beb92518'), sequence_number=3, data_point_groups=[DataPointGroup(data_id=DataID(name='greeting', type='string'), data_points=[DataPoint(elapsed_time=3002982723, payload=b'hello')])])
Received DownstreamChunk: DownstreamChunk(upstream_info=UpstreamInfo(session_id='session_id', stream_id=UUID('67bbc25d-f4d2-421d-b049-85a73c9a358b'), source_node_id='8366f4b1-dc22-4e6d-b22b-dbe2beb92518'), sequence_number=4, data_point_groups=[DataPointGroup(data_id=DataID(name='greeting', type='string'), data_points=[DataPoint(elapsed_time=4003247398, payload=b'hello')])])
Received DownstreamChunk: DownstreamChunk(upstream_info=UpstreamInfo(session_id='session_id', stream_id=UUID('67bbc25d-f4d2-421d-b049-85a73c9a358b'), source_node_id='8366f4b1-dc22-4e6d-b22b-dbe2beb92518'), sequence_number=5, data_point_groups=[DataPointGroup(data_id=DataID(name='greeting', type='string'), data_points=[DataPoint(elapsed_time=5005035682, payload=b'hello')])])
Received DownstreamChunk: DownstreamChunk(upstream_info=UpstreamInfo(session_id='session_id', stream_id=UUID('67bbc25d-f4d2-421d-b049-85a73c9a358b'), source_node_id='8366f4b1-dc22-4e6d-b22b-dbe2beb92518'), sequence_number=6, data_point_groups=[DataPointGroup(data_id=DataID(name='greeting', type='string'), data_points=[DataPoint(elapsed_time=6007406553, payload=b'hello')])])
Received DownstreamChunk: DownstreamChunk(upstream_info=UpstreamInfo(session_id='session_id', stream_id=UUID('67bbc25d-f4d2-421d-b049-85a73c9a358b'), source_node_id='8366f4b1-dc22-4e6d-b22b-dbe2beb92518'), sequence_number=7, data_point_groups=[DataPointGroup(data_id=DataID(name='greeting', type='string'), data_points=[DataPoint(elapsed_time=7009003893, payload=b'hello')])])
Received DownstreamChunk: DownstreamChunk(upstream_info=UpstreamInfo(session_id='session_id', stream_id=UUID('67bbc25d-f4d2-421d-b049-85a73c9a358b'), source_node_id='8366f4b1-dc22-4e6d-b22b-dbe2beb92518'), sequence_number=8, data_point_groups=[DataPointGroup(data_id=DataID(name='greeting', type='string'), data_points=[DataPoint(elapsed_time=8009939474, payload=b'hello')])])
Received DownstreamChunk: DownstreamChunk(upstream_info=UpstreamInfo(session_id='session_id', stream_id=UUID('67bbc25d-f4d2-421d-b049-85a73c9a358b'), source_node_id='8366f4b1-dc22-4e6d-b22b-dbe2beb92518'), sequence_number=9, data_point_groups=[DataPointGroup(data_id=DataID(name='greeting', type='string'), data_points=[DataPoint(elapsed_time=9011216126, payload=b'hello')])])
Received DownstreamChunk: DownstreamChunk(upstream_info=UpstreamInfo(session_id='session_id', stream_id=UUID('67bbc25d-f4d2-421d-b049-85a73c9a358b'), source_node_id='8366f4b1-dc22-4e6d-b22b-dbe2beb92518'), sequence_number=10, data_point_groups=[DataPointGroup(data_id=DataID(name='greeting', type='string'), data_points=[DataPoint(elapsed_time=10012087941, payload=b'hello')]), DataPointGroup(data_id=DataID(name='end', type='string'), data_points=[DataPoint(elapsed_time=10012241252, payload=b'end')])])
Received end message
closed upstream
closed upstream connection
[7]:
[None, None]