Ambient のデータを InfluxDB(バージョン2) に同期する

動機
====

梅雨時のじめじめした季節、自宅の気温・湿度はどうなるのだろう、と思い ESP8266 を使ったセンサノードを作成しました。測定値は InfluxDB に格納して Grafana で InfluxDB 上に蓄積したデータを可視化していました。

しかし、InfluxDB を動かすサーバを自宅で常時動かし続けるのは電気代的によろしくありません。そこで、センサノードから Ambient に送信して Ambient 上でデータを参照できるように構成を変更しました。

Ambient は素晴らしいサービスなのですが、データ保存期間が１年という制約もあります。
また、Grafana の表示に慣れているのでやっぱり Grafana でデータを見たいとも思いました。

そこで、Ambient にあるデータを適当な間隔で InfluxDB に取り込むことでほぼリアルタイムで自宅にある InfluxDB に転送できるようにしようと考えました。

やりたいこと
===========

1. Ambient 上にあるデータを定期的に取得して InfluxDB に登録する。
2. 最新のデータのみでなく、未登録のデータ全てについて Ambient から InfluxDB に登録したい
(自宅 InfluxDB サーバ・クローラは常時稼働しているわけではない。自宅サーバ停止中に Ambient に蓄積したデータもコピーしたい）

実現方法
=======

以下を行うプログラムを作成することにしました:

1. InfluxDB 上になく、Ambient にしかないデータを取得する。
2. 取得したデータを InfluxDB に登録する。
3. 設定した時間だけスリープする。スリープから復帰したら 1 から再実行する。

まず必要なのは InfluxDB と Ambient 双方にアクセスする手段です。
どちらも Python の API ライブラリがあるのでこれを用いることにします。

Ambient からのデータ取得
=======================

基準時刻の決定
------------

Ambient からデータを取得するにあたって問題になるのは、
「InfluxDB 上になく、Ambient にしかないデータ」
をどう決定するかです。

単純に、
ある時刻以前のデータは InfluxDB にも Ambient にも存在し、
その時刻以降のデータは Ambient にしか存在しない
と仮定するならば、InfluxDB 上の最終データの時刻を元に決定可能です。

InfluxDB から最終データを取得する方法ですが、influx クエリにおいて
|> last()
で得ることができます。
但し、range で範囲を指定してその範囲での最終データを得る形にする必要があります。
範囲の終わりは指定なしにできますが、残念ながら開始も省略して
「とにかく全部のデータをあるだけくれ」
ということはできないようです。

```
from(bucket: "${bucket}")
|> range(start: ${start})
|> filter(fn: (r) => r["_measurement"] == "${measure}")
|> last()
```

(ここで `${bucket}` はバケット名、`${measure}` はメジャー名、`${start}` は range 開始時刻に置き換えて読んでください)

`${start}` には現在時刻基準で 24 時間前、のような形で与えることができるので、
適当に１０日前(-240h)とかにしておけば事足りると思います。

当該バケット、メジャーにあるカラム全てを Ambient から持ってくる前提ならば、
上記クエリで得られる時刻が最終データ時刻となります。
Ambient から持ってくる以外のカラムが存在する場合には更にタグキーも限定する形にクエリを修正する必要があります。

データ取得
---------

基準時刻さえ決まれば、あとは簡単です。
基準時刻から 1 日分、といった形でデータ取得を行えばよいので、Ambient からはこんな感じでデータを取得できます。

```python
import ambient

utc_last_dtime = influx.get_last_datetime()
JST = datetime.timezone(datetime.timedelta(hours=+9))
jst_start_dtime = utc_last_dtime.astimezone(JST)
jst_end_dtime = jst_start_dtime + datetime.timedelta(days=1)
am = ambient.Ambient(チャネルID, None, リードキー)
dataset = am.read(start=jst_start_dtime.strftime('%Y-%m-%d %H:%M:%S'),
                  end=jst_end_dtime.strftime('%Y-%m-%d %H:%M:%S'))
```

(ちなみに、InfluxDB で得た最終カラム時刻は UTC で、start と end は JST なので UTC から JST に変換する必要がありました)

InfluxDB へのデータ登録
======================

ラインプロトコル形式での書き込み
---------------------------

InfluxDB にはラインプロトコル形式で書き込むことにします。

```
<測定値>,<タグキー>=<タグ値> <フィールドキー>=<フィールド値> <タイムスタンプ>
```

ここで、タイムスタンプはナノ秒エポック整数値です。Ambient の created 時刻からナノ秒エポックへの変換は以下のようにできます。

```python
def _get_nsec_timestamp(self, data) -> int:
    u'''Ambient の created タイムスタンプ文字列を
    InfluxDB のナノ秒エポック値として返す。
    '''
    jst_created = dateutil.parser.parse(data['created']).astimezone(JST)
    jst_timestamp = jst_created.timestamp()
    return int(jst_timestamp * 1000000000)
```
そのうえで、

```python
from influxdb_client import InfluxDBClient
from influxdb_client.client.write_api import SYNCHRONOUS

# 以下、
# INFLUX_URL は InfluxDB の URL
# INFLUX_TOKEN は InfluxDB のトークン
# INFLUX_BUCKET は InfluxDB のバケット
# INFLUX_MEASURE は InfluxDB のメジャー
# dataset は Ambient から取得した測定値
# mapping は dataset に含まれるカラム d1, d2,... を InfluxDB のどのカラムに割り当てるかを指定するマップ
client = InfluxDBClient(url=INFLUX_URL,
                        token=INFLUX_TOKEN)
write_api = client.write_api(write_options=SYNCHRONOUS)
for data in dataset:
    nsectime = _get_nsec_timestamp(data)
    for k,tagKey in mapping.items():
        if k in data:
            line = '{0},tagKey={1} fieldKey={2} {3}'.format(
                INFLUX_MEASURE, tagKey, data[k], nsectime)
            write_api.write(INFLUX_BUCKET, INFLUX_ORG, line)
```
のように書き込むことができます。

高速化
-----

データを一つづつ書き込んでみるとすこぶる遅かったので高速化を考えます。複数のラインを '\n' 区切りで連結して一度の write で済ませることにすると書き込み性能を劇的に改善できて、これをバッチ書き込み(Batch writes) と言います。

```python
lines = []
for data in dataset:
    nsectime = _get_nsec_timestamp(data)
    for k,tagKey in mapping.items():
        if k in data:
            line = '{0},tagKey={1} fieldKey={2} {3}'.format(
                INFLUX_MEASURE, tagKey, data[k], nsectime)
            lines.append(line)
write_api.write(INFLUX_BUCKET, INFLUX_ORG, '\n'.join(line))
```

定期実行
========

ここまでできればあとは簡単で、定期的に Ambient からのデータ収集と InfluxDB への書き込みを行えばよいだけです。当方の場合、InfluxDB は Docker 上で動かしているため、無限ループでデータ収集と書き込み、スリープを行うプログラムとしました。InfluxDB をサービスとして動かしている場合には cron で定期実行したほうがよいと思います。

まとめ
======

概略上記のようにして、Ambient から InfluxDB に定期的に同期するプログラムを作成できました。

```python
import ambient
import datetime
import dateutil
from influxdb_client import InfluxDBClient
from influxdb_client.client.write_api import SYNCHRONOUS
import sys
import time

class Param:
    u'''Ambient, InfluxDB への接続パラメータ
    '''
    def __init__(self):
        self.ambient_channel_id = 0
        self.ambient_read_key = ''
        self.influx_url = ''
        self.influx_org = ''
        self.influx_bucket = ''
        self.influx_measure = ''
        self.influx_token = ''
        self.influx_query_start = ''
        self.mapping = {}
class Param1F(Param):
    u'''センサノード1''
    def __init__(self):
        略
class Param3F(Param):
    u'''センサノード3''
    def __init__(self):
        略

# 5min 毎に実行
SYNC_INTERVAL = 300

# UTC-JST 変換に用いる
JST = datetime.timezone(datetime.timedelta(hours=+9))

class Ambient(object):
    u'''Ambient のデータ取得を行う
    '''
    def __init__(self, param :Param):
        self.param = param
    def query(self,
              start_dtime :datetime.datetime,
              end_dtime :datetime.datetime):
        u'''
        start_dtime - 取得開始時刻
        end_dtime - 取得終了時刻
        '''
        am = ambient.Ambient(self.param.ambient_channel_id, None,
                             self.param.ambient_read_key)
        dataset = am.read(start=start_dtime.strftime('%Y-%m-%d %H:%M:%S'),
                          end=end_dtime.strftime('%Y-%m-%d %H:%M:%S'))
        print('query[' + str(len(dataset)) + ']')
        return dataset
class Influx(object):
    u'''Influx へのデータ登録、前回登録を行った最終時刻の取得を行う。
    '''
    def __init__(self, param: Param):
        self.client = InfluxDBClient(url=param.influx_url,
                                     token=param.influx_token)
        self.param = param
    def __del__(self):
        self.close()
    def close(self):
        if self.client:
            self.client.close()
            self.client = None
    def get_last_datetime(self):
        u'''最終カラムの時刻を返す(UTC)。
        '''
        return min([row.values['_time']
                    for table in self.last_tables()
                    for row in table.records])
    def last_tables(self):
        u'''最終テーブルを検索して返す。
        '''
        query_api = self.client.query_api()
        return query_api.query(u'''from(bucket: "${bucket}")
        |> range(start: ${start})
        |> filter(fn: (r) => r["_measurement"] == "${measure}")
        |> last()
        '''.replace(
            '${bucket}', self.param.influx_bucket).replace(
            '${measure}', self.param.influx_measure).replace(
            '${start}', self.param.influx_query_start),
                               org=self.param.influx_org)
    def write_to_influx(self, dataset):
        u'''Ambient のデータを InfluxDB に書き込む。
        '''
        write_api = self.client.write_api(write_options=SYNCHRONOUS)
        lines = []
        for data in dataset:
            nsectime = self._get_nsec_timestamp(data)
            for k,tagKey in self.param.mapping.items():
                if k in data:
                    line = '{0},tagKey={1} fieldKey={2} {3}'.format(
                        self.param.influx_measure, tagKey, data[k], nsectime)
                    print(line)
                    lines.append(line)
        write_api.write(self.param.influx_bucket,
                        self.param.influx_org, '\n'.join(lines))
    def _get_nsec_timestamp(self, data) -> int:
        u'''Ambient の created タイムスタンプ文字列を
        InfluxDB のナノ秒エポック値として返す。
        '''
        jst_created = dateutil.parser.parse(data['created']).astimezone(JST)
        jst_timestamp = jst_created.timestamp()
        return int(jst_timestamp * 1000000000)
def sync(param : Param):
    influx = Influx(param)
    utc_last_dtime = influx.get_last_datetime()
    jst_start_dtime = utc_last_dtime.astimezone(JST)
    jst_end_dtime = jst_start_dtime + datetime.timedelta(days=1)
    print('{0} -> {1}'.format(jst_start_dtime, jst_end_dtime))
    ambient = Ambient(param)
    dataset = ambient.query(jst_start_dtime, jst_end_dtime)
    print('query {0} entries'.format(len(dataset)))
    if dataset:
        influx.write_to_influx(dataset)
    influx.close()
if __name__ == '__main__':
    while True:
        sync(Param1F())
        sync(Param3F())
        print('sleep...')
        time.sleep(SYNC_INTERVAL)

```