{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "680f118a",
   "metadata": {},
   "source": [
    "# アップストリームとダウンストリーム\n",
    "\n",
    "ここではiscp-pyを使用してアップストリームとダウンストリームを行うサンプルを示します\n",
    "\n",
    "## 環境\n",
    "- python 3.10\n",
    "\n",
    "## 事前準備\n",
    "\n",
    "本サンプルを動作させるための準備をします。\n",
    "まず、アップストリーム用のノードとダウンストリーム用のノードを作成します。\n",
    "ノードはintdashのエッジとマッピングされます。intdashAPIや管理コンソールを使用して作成してください。\n",
    "既に作成したノードがある場合はそのノードIDとシークレットを使用しても問題ありません。\n",
    "作成したノードのIDとシークレットを環境に合わせて下記に示す値を環境変数としてセットします。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "39b5756d",
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "URL_API = os.getenv(\"URL_API\", \"https://example.com\")\n",
    "DOWN_NODE_ID = os.getenv(\"DOWN_NODE_ID\", \"downstream-node-id\")\n",
    "DOWN_NODE_SECRET = os.getenv(\"DOWN_NODE_SECRET\", \"xxxxxxx\")\n",
    "UP_NODE_ID = os.getenv(\"UP_NODE_ID\", \"upstream-node-id\")\n",
    "UP_NODE_SECRET = os.getenv(\"UP_NODE_SECRET\", \"xxxxxxx\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6ddae02e",
   "metadata": {},
   "source": [
    "## アクセストークンの発行\n",
    "\n",
    "アップストリーム用のアクセストークンと、ダウンストリーム用のアクセストークンをそれぞれ発行します。\n",
    "ノードのIDとシークレットはOAuth2のクライアントクレデンシャル方式によるトークン発行ができます。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "e8f7fbc8",
   "metadata": {
    "hide_input": false
   },
   "outputs": [],
   "source": [
    "import string\n",
    "\n",
    "import requests\n",
    "from oauthlib.oauth2 import BackendApplicationClient\n",
    "from requests_oauthlib import OAuth2Session\n",
    "\n",
    "_URL_TOKEN_ENDPOINT = f\"{URL_API}/api/auth/oauth2/token\"\n",
    "\n",
    "def token_client_credentials(edge_id: str, secret: str) -> str:\n",
    "    \"\"\"\n",
    "    トークンの払い出しコード\n",
    "    \"\"\"\n",
    "    oauth = OAuth2Session(client=BackendApplicationClient(client_id=edge_id))\n",
    "    oauth.fetch_token(\n",
    "        token_url=_URL_TOKEN_ENDPOINT,\n",
    "        client_id=edge_id,\n",
    "        client_secret=secret,\n",
    "        include_client_id=True,\n",
    "    )\n",
    "    return oauth.token[\"access_token\"]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "af4abdf4",
   "metadata": {},
   "outputs": [],
   "source": [
    "down_token=token_client_credentials(DOWN_NODE_ID, DOWN_NODE_SECRET)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "8da6d1f6",
   "metadata": {},
   "outputs": [],
   "source": [
    "up_token=token_client_credentials(UP_NODE_ID, UP_NODE_SECRET)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6132e424",
   "metadata": {},
   "source": [
    "## アップストリームを行うコードの定義\n",
    "\n",
    "アップストリームを行うコードのサンプルです。\n",
    "このサンプルでは、トランスポートにWebSocketを使用して基準時刻のメタデータと、文字列型のデータポイントをiSCPサーバーへ送信しています。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "74ed69ce",
   "metadata": {},
   "outputs": [],
   "source": [
    "import iscp\n",
    "import asyncio\n",
    "import urllib\n",
    "\n",
    "_URL_API = urllib.parse.urlparse(URL_API)\n",
    "_ENABLE_TLS=_URL_API.scheme == \"https\"\n",
    "_ISCP_PORT = _URL_API.port if _URL_API.port else 443 if _URL_API.scheme == \"https\" else 80\n",
    "_ADDRESS_WEBSOCKET= f\"{_URL_API.hostname}:{_ISCP_PORT}\"\n",
    "\n",
    "started_downstream = asyncio.Event()\n",
    "\n",
    "async def start_upstream():\n",
    "    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:\n",
    "        async with await conn.open_upstream(session_id=\"session_id\") as up:\n",
    "            await started_downstream.wait()\n",
    "            start = iscp.DateTime.utcnow()\n",
    "            await conn.send_base_time(\n",
    "                iscp.BaseTime(\n",
    "                    name=\"manual\",\n",
    "                    elapsed_time=1_000_000,\n",
    "                    base_time=start,\n",
    "                    priority=1000,\n",
    "                    session_id=\"session_id\",\n",
    "                )\n",
    "            )\n",
    "            for i in range(10):\n",
    "                await asyncio.sleep(1)\n",
    "                await up.write_data_points(\n",
    "                    iscp.DataID(name=\"greeting\", type=\"string\"),\n",
    "                    iscp.DataPoint(\n",
    "                        elapsed_time=iscp.DateTime.utcnow().unix_nano() - start.unix_nano(),\n",
    "                        payload=b\"hello\",\n",
    "                    ),\n",
    "                )\n",
    "            await up.write_data_points(\n",
    "                iscp.DataID(name=\"end\", type=\"string\"),\n",
    "                iscp.DataPoint(\n",
    "                    elapsed_time=iscp.DateTime.utcnow().unix_nano() - start.unix_nano(),\n",
    "                    payload=b\"end\",\n",
    "                ),\n",
    "            )\n",
    "        print(\"closed upstream\")\n",
    "    print(\"closed upstream connection\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0fa924b8",
   "metadata": {},
   "source": [
    "## ダウンストリームを行うコードの定義\n",
    "\n",
    "前述のアップストリームで送信されたデータをダウンストリームで受信するコードのサンプルです。\n",
    "\n",
    "このサンプルでは、アップストリーム開始のメタデータ、基準時刻のメタデータ、\n",
    "文字列型のデータポイントを受信しています。\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "4b11e81c",
   "metadata": {},
   "outputs": [],
   "source": [
    "import iscp\n",
    "import json\n",
    "import dataclasses\n",
    "\n",
    "async def start_downstream():\n",
    "    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:\n",
    "        async with await conn.open_downstream(\n",
    "            filters=[\n",
    "                iscp.DownstreamFilter(\n",
    "                    source_node_id=UP_NODE_ID,\n",
    "                    data_filters=[iscp.DataFilter(name=\"#\", type=\"#\")],\n",
    "                )\n",
    "            ]\n",
    "        ) as down:\n",
    "            started_downstream.set()\n",
    "            metadata = await down.metadatas().__anext__()  # 基準時刻の受信\n",
    "            print(f\"Received Metadata: {metadata}\")\n",
    "            async for msg in down.chunks():  # データポイントの受信\n",
    "                print(f\"Received DownstreamChunk: {msg}\")\n",
    "                for data_point_group in msg.data_point_groups:\n",
    "                    if data_point_group.data_id.name == \"end\":\n",
    "                        print(\"Received end message\")\n",
    "                        return"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e8c34f22",
   "metadata": {},
   "source": [
    "## 実行"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "2a356597",
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "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'>))\n",
      "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')])])\n",
      "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')])])\n",
      "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')])])\n",
      "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')])])\n",
      "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')])])\n",
      "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')])])\n",
      "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')])])\n",
      "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')])])\n",
      "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')])])\n",
      "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')])])\n",
      "Received end message\n",
      "closed upstream\n",
      "closed upstream connection\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[None, None]"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import asyncio\n",
    "\n",
    "await asyncio.gather(start_downstream(), start_upstream())\n",
    "\n",
    "# 本サンプルはJupyter Notebookから実行しています。\n",
    "# 通常の利用シーンでは、下記のように非同期処理を実行する必要があることにご注意ください。\n",
    "# import asyncio\n",
    "\n",
    "# async def main():\n",
    "#     await asyncio.gather(start_downstream(), start_upstream())\n",
    "\n",
    "# if __name__ == \"__main__\":\n",
    "#     asyncio.run(main())"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.10.11"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}