pub/sub通信ライブラリZenohを試す

# １．はじめに

　Zenohは通信プロトコル／ミドルウェアの一つで、出版購読型の通信方式に加え、キーバリューストア型の分散クエリの機能も提供しています。
　2023年末に、ZettaScale CEO/CTOのAngelo Corsaro さんが来日されるタイミングで、セミナーが開かれます。

https://connpass.com/event/304464/

　C/C++やPythonなどのAPIライブラリも提供されていますので、予習を兼ねて、PythonでZenohによるpub/sub通信を試してみました。

# ２．環境

- Ubuntu Linux 22.04
- Python 3.10
- [poetry](https://python-poetry.org/) で仮想環境
    - [参考](https://zenn.dev/yamaday/scraps/51008245130dac)

# ３．考え方

こちらのサンプルコードは、仮想的に温度を測定し、出版、購読するサンプルです。
購読・出版は１キーワード(key expressions)です。

https://zenoh.io/docs/getting-started/first-app/

今回の記事では少し拡張して、２つのkey expressionsを用いて、仮想的な温度と湿度を購読したり出版したりするサンプルを作ってみました。

```mermaid
graph LR
  
  p([publish])
  p -->|1sec毎| k1[myhome/kitchen/temp]
  p -->|1sec毎| k2[myhome/kitchen/humi]
  k1 --> s1([listener_tempハンドラ])
  k1 --> s3([listenerハンドラ])
  k2 --> s3
  k2 --> s2([listener_humiハンドラ])

  
　subgraph z_publisher.py　
    p
  end
  subgraph z_subscriber.py　
    s1
    s3
    s2
  end
```

- `z_publisher.py`は２つのkey expressionsを出版します
    - `myhome/kitchen/temp`と`myhome/kitchen/humi`です
- `z_subscriber.py`は２つのkey expressionsを購読します
    - key expressionsそれぞれを購読する専用のハンドラ（listener_temp、listener_humi）
    - 両方とも購読するハンドラ（listener）

# ４．ソースコード

poetryでPython環境を作ります。

```shell-session
$ cd ~

# poetry環境を作る
$ poetry new zenohpy
$ cd zenohpy

# zenohパッケージをインストール
$ poetry add eclipse-zenoh

# 空のファイルを作成
$ touch zenohpy/z_publisher.py zenohpy/z_subscriber.py

# 実行権限を付与（poetry runするときに実行権限が必要）
$ chmod 755 zenohpy/z_publisher.py zenohpy/z_subscriber.py
```

## publisher側

```python:zenohpy/z_publisher.py
#! /usr/bin/env python3
# -*- coding: utf-8 -*-
"""publisher
参考：https://zenoh.io/docs/getting-started/first-app/
"""
import zenoh, random, time, datetime
from typing import Tuple

random.seed()


def gen_value() -> Tuple[float, float]:
    """ランダムに値を生成

    Returns:
        Tuple[float, float]: 温度、湿度
    """
    return (random.randint(150, 300) / 10, random.randint(400, 600) / 10)


if __name__ == "__main__":
    try:
        print(f"-- zenoh sample - publisher --")

        session = zenoh.open()

        # key expressionsごとにpublisherを生成
        key_temp = "myhome/kitchen/temp"
        pub_temp = session.declare_publisher(key_temp)

        key_humi = "myhome/kitchen/humi"
        pub_humi = session.declare_publisher(key_humi)

        while True:
            # ダミーの温度・湿度の値を生成
            (t, h) = gen_value()
            dt_now = datetime.datetime.now().isoformat()

            # 温度をpublish
            print(f"Putting Data ('{key_temp}': '{t}' / {dt_now})...")
            pub_temp.put(t)

            # 湿度をpublish
            print(f"Putting Data ('{key_humi}': '{h}' / {dt_now})...")
            pub_humi.put(h)

            # 一秒待機
            time.sleep(1)

    except KeyboardInterrupt:
        # [Ctrl-C]が押された
        print("SIGINT - Exit")
    except:
        # 例外発生時にメッセージ
        import traceback

        traceback.print_exc()
    finally:
        pass

```

## subscriber側

```python:zenohpy/z_subscriber.py
#! /usr/bin/env python3
# -*- coding: utf-8 -*-
"""subscriber
参考：https://zenoh.io/docs/getting-started/first-app/
"""
import zenoh


def listener_temp(sample):
    """受信・温度"""
    print(
        f"Received {sample.kind} ('{sample.key_expr}': '{sample.payload.decode('utf-8')}'℃)"
    )


def listener_humi(sample):
    """受信・湿度"""
    print(
        f"Received {sample.encoding} ('{sample.key_expr}': '{sample.payload.decode('utf-8')}'%)"
    )


def listener(sample):
    """受信・すべて"""
    str = f"Received all: {sample.encoding},{sample.key_expr},{sample.kind},{sample.payload},{sample.timestamp},{sample.value}"
    print(str)


if __name__ == "__main__":
    try:
        print(f"-- zenoh sample - subscriber --")

        session = zenoh.open()

        # 受信したkey expressionsごとに、実行する関数を紐づけ
        sub1 = session.declare_subscriber("myhome/kitchen/temp", listener_temp)
        sub2 = session.declare_subscriber("myhome/kitchen/humi", listener_humi)
        sub3 = session.declare_subscriber("myhome/kitchen/*", listener)
        # ここで受信待機状態になる
        # time.sleep(60) ←書いても書かなくても良さそう
    except KeyboardInterrupt:
        # [Ctrl-C]が押された
        print("SIGINT - Exit")
    except:
        # 例外発生時にメッセージ
        import traceback

        traceback.print_exc()
    finally:
        pass
```

# ５．実行例

ターミナルを２つ開いて、それぞれ実行します。

```shell-session:ターミナル１・発行用
$ poetry run python ./zenoh/z_publisher.py 

-- zenoh sample - publisher --
Putting Data ('myhome/kitchen/temp': '22.5' / 2023-12-03T15:36:12.010331)...
Putting Data ('myhome/kitchen/humi': '41.3' / 2023-12-03T15:36:12.010331)...
^CSIGINT - Exit
```

２つのkey expressionsをPublishします

- `myhome/kitchen/temp`: 22.5
- `myhome/kitchen/humi`: 41.3

```shell-session:ターミナル２・購読用
$ poetry run python ./zenoh/z_subscriber.py 

-- zenoh sample - subscriber --
Received all: application/float,myhome/kitchen/temp,PUT,b'22.5',None,<zenoh.value.Value object at 0x7fe2d6d17240>
Received PUT ('myhome/kitchen/temp': '22.5'℃)
Received all: application/float,myhome/kitchen/humi,PUT,b'41.3',None,<zenoh.value.Value object at 0x7fe2d6d17240>
Received PUT ('myhome/kitchen/humi': '41.3'%)
```

２つのkey expressionsを購読します

|key expressions|`myhome/kitchen/temp`:22.5|`myhome/kitchen/humi`: 41.3|
|:--|:--:|:--:|
|ハンドラ`listener_temp`|○||
|ハンドラ`listener_humi`||○|
|ハンドラ`listener`|○|○|

このように、簡単に２つのPub/Subを作る事ができました。

# ６．参考資料

- https://www.slideshare.net/Angelo.Corsaro/zenoh-tutorial
- https://zenoh-python.readthedocs.io/en/latest/
- https://zenoh.io/
  - https://github.com/eclipse-zenoh/zenoh