Cisco Spaces Firehose API を利用するScriptをChatGPTだけで生成する

# はじめに
この投稿はシスコシステムズ合同会社の有志でお送りする Advent Calendar 2024 の一環でお届けします。
私は2021年以降3年ぶりの投稿になります。
ぜひ他の投稿もご覧ください！

https://qiita.com/advent-calendar/2024/cisco


:::note info
免責事項
本サイトおよび対応するコメントにおいて表明される意見は、投稿者本人の個人的意見であり、シスコの意見ではありません。本サイトの内容は、情報の提供のみを目的として掲載されており、シスコや他の関係者による推奨や表明を目的としたものではありません。各利用者は、本Webサイトへの掲載により、投稿、リンクその他の方法でアップロードした全ての情報の内容に対して全責任を負い、本Web サイトの利用に関するあらゆる責任からシスコを免責することに同意したものとします。
:::

# Cisco Wi-Fi7無線アクセスポイントとの関連
Ciscoで先日Wi-Fi7のアクセスポイントとして、Cisco Wireless 9176シリーズおよび9178シリーズを発表しました。これらの無線LANアクセスポイントでは、Apple社がAirTagなどでも活用されているUWBの無線を搭載しており、Wi-FiやBluetoothより高精度な位置情報ソリューションが今後搭乗する予定です。その際もプラットフォームとして今回ご紹介するCisco Spacesが活用されることと思いますので、Cisco Spacesでそんなことができるようなるかも、と思いながら是非見ていってください。

https://www.cisco.com/c/ja_jp/products/collateral/wireless/catalyst-9100ax-access-points/wireless-9176-series-acc-point-ds.html

https://www.cisco.com/c/en/us/products/collateral/wireless/catalyst-9100ax-access-points/wireless-9178-series-access-point-ds.html


# Cisco Spacesとは
[Cisco Spaces](https://www.cisco.com/c/ja_jp/solutions/enterprise-networks/dna-spaces/index.html) というサービスはご存じでしょうか？
Cisco Spaces は Ciscoのワイヤレスインフラストラクチャ等から利用可能なSaaS型のサービスで、従来Wi-Fiサービスの提供といったインフラとしての使い方しかできなかったCisco機器を活用し、ビジネスで利用可能なデータを集積、分析ができるサービスです。これらには例えば「無線LANクライアントの位置情報」、「無線LAN干渉源の位置情報」、「IoT機器センサー情報 (Bluetooth経由)」などがあります。

![Screenshot 2024-11-18 at 10.32.42.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/385067/04a9ccb1-ffbd-73dc-66a5-adaa7c057f8e.png)


昨今のCisco Catalyst アクセスポイントの、Cisco DNA Advantage ライセンスは、Cisco Spaces Extendedを含有しており、Cisco DNA Advantage でご購入いただいたお客様はライセンス的にはご利用可能な状態になっておりますので、是非ご活用ください。
（ご活用いただけるとご存じないお客様も多いようにお見かけします）

これらの収集したデータはAPIでサードパーティでもご活用することが可能で、日本でよく知られているのは、PHONE APPLI社様「[PHONE APPLI PEOPLE](https://phoneappli.net/)」や、内田洋行社様「[SmartOfficeNagigator](https://www.uchida.co.jp/it/products/smartofficenavigator/)」などで、Cisco Spaces APIを利用してクライアントの位置情報を取得し各サービスからご利用いただくことが可能です。

さて、今回はこの Cisco Spaces の機能のうち、無線LANの位置情報はご活用が進んでいるとは思いますが、それ以外の機能、とりわけ Indoor IoT と呼ばれる Cisco Catalyst Access Point をBluetoothのスキャナとして利用する機能は、あまり知られていないとおもいますので、こちらを取り上げていきたいと思います。


# Cisco Spaces Indoor IoT とは
Cisco Spacces indoor IoT とは、Cisco Catalyst スイッチからの有線接続やCisco Catalyst アクセスポイントがもっているBluetooth機能を利用して、IoT機器へのアクセス、管理、データ収集を行うサービスになります。
従来IoTデバイス、例えばBluetooth型デバイスをばらまく場合は、IoTデバイスそのものの管理にとどまらず、Bluetoothの受信機、受信機を接続するLANなどのネットワークを情報系ネットワークと別に形成することが多く、IoT活用を始めには整備するものが多いことが課題の一つでした。 Cisco Spaces ではすでにデプロイ済のCisco Catalyst アクセスポイントおよび、それが接続する既存の有線ネットワークを利活用することでIoTをはじめやすくするものになります。

![Screenshot 2024-11-18 at 10.34.17.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/385067/6a26c3d1-4bb1-9194-6b48-48e848763a40.png)


データフローはこちらを参考ください。無線の場合Catalyst アクセスポイントからCisco Spaces Connector でgRPCによるトンネルが生成されて、そこにIoTデバイスのデータが流れる形になります。
![Screenshot 2024-11-18 at 10.35.12.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/385067/7d8780a2-5800-1211-9588-7fbe82fc703a.png)
一点注意事項があるとすると、IoTデバイスはCiscoの認定がとれているものからお選びをいただく必要があります。

:::note info 
下記の構成要素が必要になります
Catalyst 9800 ワイヤレスLANコントローラ (17.3.1以上)
Cisco Spaces Connector (2.3.2以上)
Cisco Spaces のライセンスは最上位のACTライセンス
AireOS/Catalsyt アクセスポイント (wave2以降)
:::

詳細なガイドはこちらをご参照ください
- [Cisco Spaces Connector インストレーションガイド](https://www.cisco.com/c/en/us/td/docs/wireless/spaces/connector/config/b_connector_30/m_initialsetup_30.html)
- [Cisco Spacss IoT Service Configuration Guide (Wireless)](https://www.cisco.com/c/en/us/td/docs/wireless/spaces/iot-services-wireless/b_iot_services/m_overview.html)

無事に設定すると、以下のようにCisco Spacesでセンサーデータが表示されます（温度、湿度など)
![Screenshot 2024-11-18 at 10.45.26.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/385067/8838f6b4-b52d-57a4-136b-9e4205a0913f.png)

ここまでがBlogで話す前提です（長かった）

# 今回やりたいこと
## 前提
Cisco Spaces でもこれらのセンサーデータを時系列でグラフ化することはできますが、他にもっているセンサー情報と統合して分析したい、というユースケースもあるかと思います。
今回はこのセンサーデータをAPIで取得し分析したいと思います。

## API
Cisco Spaces のAPIには、REST-APIによるいわゆるGETで情報をもっていくるタイプのAPIと、Firehose APIの2つの実装があります。原稿執筆時点(2024年12月)時点で、 **IoTデバイスのセンサーデータはFirehose APIでのみ取得可能** なので、今回はFirehose APIを利用していきます。

## Firehose API の始め方
Cisco Spaces の最下部、EXTENDのセクションでアプリケーションを作成します。
![Screenshot 2024-11-18 at 10.51.34.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/385067/eb4f0242-85ec-e193-620f-80620368c1c4.png)

このカスタムアプリケーションを生成する時点で、利用するデータの対象拠点、対象のセンサー（グループなどで指定）、そしてAPKキーを生成することになります。詳しい手順は下記をご参照ください。

- [Cisco Spaces API ガイド](https://www.cisco.com/c/dam/en/us/td/docs/wireless/cisco-dna-spaces/partner-app/partner-firehose-api/Cisco_Spaces_API_Guide.pdf)
- [Cisco Spaces API ガイド（日本語版）](https://www.cisco.com/c/dam/global/ja_jp/td/docs/wireless/cisco-dna-spaces/partner-app/partner-firehose-api/Cisco_DNA_Spaces_API_Guide.pdf)

## Firehose APIのデータ
ではさっそくAPIキーをつかって情報を収集してみましょう。curlコマンドを使って、Firehose APIのFQDNとAPI-keyを指定します。

```curl
curl "https://partners.dnaspaces.io/api/partners/v1/firehose/events" -H "X-API-Key:[YOUR API KEY IS HERE]"
```

JSON形式でデータが連続で出力されます。結果サンプルは以下になります

```json
{
    "recordUid": "event-4438fcf3",
    "recordTimestamp": 1731894987730,
    "spacesTenantId": "spaces-tenant-c1b460d8",
    "spacesTenantName": "HShimomura",
    "partnerTenantId": "hshimomura",
    "eventType": "IOT_TELEMETRY",
    "iotTelemetry": {
        "deviceInfo": {
            "deviceType": "IOT_BLE_DEVICE",
            "deviceId": "c7:1f:2c:0f:bc:41",
            "deviceMacAddress": "c7:1f:2c:0f:bc:41",
            "group": [
                "sernsors"
            ],
            "deviceName": "",
            "firmwareVersion": "",
            "rawDeviceId": "",
            "manufacturer": "KNKT",
            "companyId": "",
            "serviceUuid": "6afe",
            "label": "Sensor2",
            "vendorId": "KNKT",
            "deviceModel": "S1"
        },
        "detectedPosition": {
            "xPos": 28.9,
            "yPos": 38,
            "latitude": 34.65592171201714,
            "longitude": 138.54643172377627,
            "confidenceFactor": 24,
            "mapId": "b8204196d6094babd1ad2c69b7f68c05",
            "locationId": "location-800167de",
            "lastLocatedTime": 1731894980000
        },
        "location": {
            "locationId": "location-800167de",
            "name": "1st Floor",
            "inferredLocationTypes": [
                "FLOOR"
            ],
            "parent": {
                "locationId": "location-a642e73a",
                "name": "TOKYO",
                "inferredLocationTypes": [
                    "NETWORK",
                    "BUILDING"
                ],
                "parent": {
                    "locationId": "location-6b20ac53",
                    "name": "System Campus",
                    "inferredLocationTypes": [
                        "CAMPUS"
                    ],
                    "parent": {
                        "locationId": "location-2a9f637d",
                        "name": "HShimomura",
                        "inferredLocationTypes": [
                            "ROOT"
                        ],
                        "sourceLocationId": "",
                        "apCount": 4
                    },
                    "sourceLocationId": "11b21e92-3464-4b53-bc9f-178eeb31c426",
                    "apCount": 3
                },
                "sourceLocationId": "38964ca4-eff2-41b6-bcea-b0ec1fde1365",
                "apCount": 3
            },
            "sourceLocationId": "8e1a724d-8d5a-42ab-ba9f-746251fbfd49",
            "floorNumber": 1,
            "apCount": 3
        },
        "temperature": {
            "temperatureInCelsius": 22.484375,
            "rawTemperature": 72.47187392786145
        },
        "accelerometer": {
            "x": -2,
            "y": 32,
            "z": -116,
            "lastMovementTimestamp": 0,
            "counter": 0
        },
        "deviceRtcTime": -1,
        "rawHeader": 0,
        "rawPayload": "AgEGGxZq/gMFBhD+IIwEEQD//wQWAP//AxN8FgISPg==",
        "sequenceNum": 0,
        "humidity": {
            "humidityInPercentage": 62,
            "rawHumidity": 62
        },
        "maxDetectedRssi": -47
    }
}
```

## データ収集方針
今回は以下のような方針でいきます
- 利用言語はPythonとする
- PythonでFirehose APIにアクセスし、取得するJSONオブジェクトからtemperatureInCelsius, humidityInPercentageなどのデータを取得する
- 取得したデータをInfluxDB (2.0系)に投げ込む

たぶんそんな高度な内容ではないと思うのですが、私は大学卒業後以降一切プログラムを書いたことがないので、すべてChatGPT 4oにお願いします。
ChatGPTにお願いすること数時間でできあがったのが下記スクリプトになります。プロンプトの内容は「curlでこうすると以下のようなjsonが得られる。pythonで同じことをしてほしい」「取得したjsonからtemperatureInCelsius, humidityInPercentageデータを時系列と共に、influxdbにアップロードしたい」「influxDBには1分おきにデータを貯めてからアップロードする、パラメータはコレコレ」などを続けていきました。
試していく中で今回試したセンターの一部にはデータがばらつくものがあったので、前後の値を比較して外れ値を弾いています。閾値は適当に30%を指定しました。

:::note info
JSONデータを見ていただくとX,Yの座標もあるので、これを応用してBLE機器の位置情報をトラックすることも可能ですね。
:::

## ciscospaces.py
生成されたPython3 のスクリプトです。コメントもすべてChatGPTがつくっています。優秀すぎる。

```python
import requests
import json
import os
from influxdb_client import InfluxDBClient, Point
from influxdb_client.client.write_api import SYNCHRONOUS
from datetime import datetime, timedelta

# 環境変数の定義
api_key = 'YOUR CISCO SPACES API is HERE'
influxdb_url = "http://localhost:8086"
influxdb_token = "YOUR INFLUXDB TOKEN is HERE"
influxdb_bucket = "YOUR INFLUXDB BUCKET is HERE"
influxdb_org = "YOUR INFLUXDB ORG is HERE"
debug_logging = os.getenv('DEBUG_LOGGING', 'False').lower() in ['true', '1', 't']
log_file_path = os.getenv('LOG_FILE_PATH', 'firehose_debug.log')

# InfluxDBの設定
client = InfluxDBClient(url=influxdb_url, token=influxdb_token)
write_api = client.write_api(write_options=SYNCHRONOUS)

# デバッグ用ログファイルにメッセージを記録
def log_debug_message(message):
    if debug_logging:
        with open(log_file_path, "a") as log_file:
            log_file.write(message + "\n")

# Firehoseストリームの開始
s = requests.Session()
s.headers = {'X-API-Key': api_key}
r = s.get('https://partners.dnaspaces.io/api/partners/v1/firehose/events', stream=True)

# デバイスごとのデータと最後のアップロード時刻を保存するための辞書
device_data = {}
last_upload_times = {}
last_values = {}

# 外れ値を検出する関数
def is_outlier(current_value, last_value, threshold=0.3):
    if last_value is None or last_value == 0:
        return False
    difference = abs(current_value - last_value)
    return difference / last_value > threshold

# ストリームからデータを受信し、処理する
for line in r.iter_lines():
    if line:
        try:
            event = json.loads(line.decode('utf-8'))
            # デバッグ情報をログファイルに記録
            log_debug_message(f"Received event: {json.dumps(event)}")

            # "eventType"が"IOT_TELEMETRY"のイベントを抽出
            if event.get("eventType") == "IOT_TELEMETRY":
                device_mac = event.get("iotTelemetry", {}).get("deviceInfo", {}).get("deviceMacAddress")
                record_timestamp = event.get("recordTimestamp")

                # タイムスタンプをISOフォーマットに変換
                timestamp = datetime.utcfromtimestamp(record_timestamp / 1000).isoformat() + "Z"

                # データがない場合は初期化
                if device_mac not in device_data:
                    device_data[device_mac] = {"timestamp": timestamp}
                    last_values[device_mac] = {}

                # データを更新
                valid_data = True

                if "temperature" in event.get("iotTelemetry", {}):
                    current_temp = round(event["iotTelemetry"]["temperature"]["temperatureInCelsius"], 1)
                    if is_outlier(current_temp, last_values[device_mac].get("temperatureInCelsius")):
                        log_debug_message(f"Outlier detected for temperature: {current_temp}")
                        valid_data = False
                    else:
                        device_data[device_mac]["temperatureInCelsius"] = current_temp
                        last_values[device_mac]["temperatureInCelsius"] = current_temp

                if "humidity" in event.get("iotTelemetry", {}):
                    current_humidity = event["iotTelemetry"]["humidity"]["humidityInPercentage"]
                    if is_outlier(current_humidity, last_values[device_mac].get("humidityInPercentage")):
                        log_debug_message(f"Outlier detected for humidity: {current_humidity}")
                        valid_data = False
                    else:
                        device_data[device_mac]["humidityInPercentage"] = current_humidity
                        last_values[device_mac]["humidityInPercentage"] = current_humidity

                if "carbonEmissions" in event.get("iotTelemetry", {}):
                    current_co2 = event["iotTelemetry"]["carbonEmissions"]["co2Ppm"]
                    if is_outlier(current_co2, last_values[device_mac].get("co2Ppm")):
                        log_debug_message(f"Outlier detected for CO2: {current_co2}")
                        valid_data = False
                    else:
                        device_data[device_mac]["co2Ppm"] = current_co2
                        last_values[device_mac]["co2Ppm"] = current_co2

                if "airPressure" in event.get("iotTelemetry", {}):
                    current_pressure = round(event["iotTelemetry"]["airPressure"]["pressure"], 1)
                    if is_outlier(current_pressure, last_values[device_mac].get("airPressure")):
                        log_debug_message(f"Outlier detected for air pressure: {current_pressure}")
                        valid_data = False
                    else:
                        device_data[device_mac]["airPressure"] = current_pressure
                        last_values[device_mac]["airPressure"] = current_pressure

                if "illuminance" in event.get("iotTelemetry", {}):
                    current_illuminance = event["iotTelemetry"]["illuminance"]["value"]
                    if current_illuminance != 65535:
                        device_data[device_mac]["illuminance"] = current_illuminance
                        last_values[device_mac]["illuminance"] = current_illuminance

                if "battery" in event.get("iotTelemetry", {}):
                    current_battery = event["iotTelemetry"]["battery"]["value"]
                    if is_outlier(current_battery, last_values[device_mac].get("battery")):
                        log_debug_message(f"Outlier detected for battery: {current_battery}")
                        valid_data = False
                    else:
                        device_data[device_mac]["battery"] = current_battery
                        last_values[device_mac]["battery"] = current_battery

                if "tvoc" in event.get("iotTelemetry", {}):
                    current_tvoc = event["iotTelemetry"]["tvoc"]["valueInPpb"]
                    # if is_outlier(current_tvoc, last_values[device_mac].get("tvoc")):
                    #     log_debug_message(f"Outlier detected for TVOC: {current_tvoc}")
                    #     valid_data = False
                    # else:
                    #     device_data[device_mac]["tvoc"] = current_tvoc
                    #     last_values[device_mac]["tvoc"] = current_tvoc
                    device_data[device_mac]["tvoc"] = current_tvoc
                    last_values[device_mac]["tvoc"] = current_tvoc
                device_data[device_mac]["timestamp"] = timestamp

                if valid_data:
                    # 前回のアップロードから1分経過したか確認
                    last_upload_time = last_upload_times.get(device_mac)
                    if last_upload_time is None or (record_timestamp - last_upload_time) >= 60000:
                        # 結果をJSON形式で表示
                        result = {
                            "deviceMacAddress": device_mac,
                            "timestamp": device_data[device_mac]["timestamp"]
                        }
                        if "temperatureInCelsius" in device_data[device_mac]:
                            result["temperatureInCelsius"] = device_data[device_mac]["temperatureInCelsius"]
                        if "humidityInPercentage" in device_data[device_mac]:
                            result["humidityInPercentage"] = device_data[device_mac]["humidityInPercentage"]
                        if "co2Ppm" in device_data[device_mac]:
                            result["co2Ppm"] = device_data[device_mac]["co2Ppm"]
                        if "airPressure" in device_data[device_mac]:
                            result["airPressure"] = device_data[device_mac]["airPressure"]
                        if "illuminance" in device_data[device_mac]:
                            result["illuminance"] = device_data[device_mac]["illuminance"]
                        if "battery" in device_data[device_mac]:
                            result["battery"] = device_data[device_mac]["battery"]
                        if "tvoc" in device_data[device_mac]:
                            result["tvoc"] = device_data[device_mac]["tvoc"]

                        print(json.dumps(result, indent=4))

                        # デバッグ情報をログファイルに記録
                        log_debug_message(f"Uploaded to InfluxDB: {json.dumps(result)}")

                        # InfluxDBにデータをアップロード
                        point = Point("iot_telemetry")\
                            .tag("deviceMacAddress", device_mac)\
                            .time(device_data[device_mac]["timestamp"])
                        if "temperatureInCelsius" in device_data[device_mac]:
                            point.field("temperatureInCelsius", device_data[device_mac]["temperatureInCelsius"])
                        if "humidityInPercentage" in device_data[device_mac]:
                            point.field("humidityInPercentage", device_data[device_mac]["humidityInPercentage"])
                        if "co2Ppm" in device_data[device_mac]:
                            point.field("co2Ppm", device_data[device_mac]["co2Ppm"])
                        if "airPressure" in device_data[device_mac]:
                            point.field("airPressure", device_data[device_mac]["airPressure"])
                        if "illuminance" in device_data[device_mac]:
                            point.field("illuminance", device_data[device_mac]["illuminance"])
                        if "battery" in device_data[device_mac]:
                            point.field("battery", device_data[device_mac]["battery"])
                        if "tvoc" in device_data[device_mac]:
                            point.field("tvoc", device_data[device_mac]["tvoc"])
                        write_api.write(bucket=influxdb_bucket, org=influxdb_org, record=point)

                        # 最後のアップロード時刻を更新
                        last_upload_times[device_mac] = record_timestamp

        except json.JSONDecodeError as e:
            error_message = f"Error decoding JSON: {e}"
            print(error_message)
            log_debug_message(error_message)

# クライアントを閉じる
client.close()
```


## グラフ化
influxDB にいれたデータをGrafanaでグラフ化してみましょう。

![Screenshot 2024-11-18 at 11.08.20.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/385067/b8bafc25-eb0f-00f3-859a-1d1f8d3fe537.png)

CO2センサーは私の寝室に置いているのですが、寝ている間にCO2濃度が高くなっていることが観察されます。例えば、会議室に置いておいて閾値以上になったらアラートを飛ばす、もしくは、自動的に窓を開ける、なんてことも簡単に実装できそうですね。


# まとめ
今回のまとめは以下になります

- Cisco のワイヤレスLANは、単純な無線LANインフラを構成するだけではなく、クライアント端末の位置情報やBluetoothのデータ通信など1つで2つも3つも美味しい機能を搭載しています
- Cisco Spaces では Indoor IoT 機能をつかって、Bluetooth IoT機器のセンサー情報を取得することが可能です
- Cisco Spaces Indoor IoTのセンサー情報はREST APIではなく、Firehose APIによる取得が必要である
- Firehose APIの中身はJSONであり、簡単なPythonスクリプトで情報収集することができる
- 今はChatGPTがあるので、手軽にIoT機能のPoCを行うことができる


Cisco Catalyst Wireless をご利用のお客様で、センサーデータを活用されていないお客様、是非お試しください。