kyowa-cloud-field 1.0.0

KYOWA CLOUD FIELD (KCF) data upload client library.

kyowa-cloud-field

KYOWA CLOUD FIELD (KCF) へデータをアップロードするための公式 Python クライアントライブラリです。 Python 3.10 以降のモダンな環境に最適化されており、型安全かつ直感的に時系列データを送信できます。

💡 主な特徴

• 自動分割送信: レコード数やCH数が多くデータサイズが大きい場合、内部で自動的にレコードの単位を維持したまま、適切な件数ごとにパッケージングして分割送信します。1つのレコード（測定時刻データ）そのものが途中で分割されることはありません。ユーザーはデータサイズを意識することなく、1つのメソッドを呼び出すだけで大容量データをアップロード可能です。
• 柔軟なインプット対応: 専用データクラスのほか、プレーンな「リスト」や「辞書のリスト」など、扱いやすい構造のデータをそのまま受け入れられます。
• pandas 拡張サポート: DataFrame から直接スマートに一括送信する機能を備えています。

1. インストール方法

本ライブラリは、用途に合わせて2種類の方法でインストールできます。

コア機能のみ（軽量版）

サーバーとの通信機能のみを利用する場合（pandas を使用しない環境向け）：

pip install kyowa-cloud-field

コア機能＋pandas 拡張対応版

CSVやExcel、pandasの DataFrame から直接一括アップロードを行う機能を利用する場合：

pip install kyowa-cloud-field[pandas]

2. 基本的なデータ送信方法

スクリプトから直接データを組み立てて送信する方法。

データの組み立て

パターンA：専用のデータクラスKcfRecord, KcfChannelDataを利用する場合

from kyowacloudfield import KcfRecord, KcfChannelData
from datetime import datetime

# 送信データの組み立て
upload_datas = [
    KcfRecord(
        time="2026/06/25 10:00:00.000",  # 文字列形式（ミリ秒対応）で測定時刻を記述できます
        measure=[
            KcfChannelData(ch=1, name="sensor1", data=[100, 9.8, 0]),
            KcfChannelData(ch=2, name="sensor2", data=[55, 1.0, 0])
        ]
    ),
    KcfRecord(
        time=datetime(2026, 6, 25, 11, 0, 0),  # datetimeオブジェクトを直接渡すことも可能です
        measure=[
            KcfChannelData(ch=1, name="sensor1", data=[103, 10.1, 0]),
            KcfChannelData(ch=2, name="sensor2", data=[999999, 19999.88, 1])
        ]
    )
]

パターンB：リスト形式でデータを組み立てる場合

from datetime import datetime

# 送信データの組み立て
upload_datas = [
    [
        "2026/06/25 10:00:00.000",  # 文字列形式（ミリ秒対応）で測定時刻を記述できます
        [
            [1, "sensor1", [100, 9.8, 0]],
            [2, "sensor2", [55, 1.0, 0]]
        ],
    ],
    [
        datetime(2026, 6, 25, 11, 0, 0),  # datetimeオブジェクトを直接渡すことも可能です
        [
            [1, "sensor1", [103, 10.1, 0]],
            [2, "sensor2", [999999, 19999.88, 1]]
        ]
    ]
]

パターンC：辞書のリストでデータを組み立てる場合

from datetime import datetime

# 送信データの組み立て
upload_datas = [
    {
        "time": "2026/06/25 10:00:00.000",  # 文字列形式（ミリ秒対応）で測定時刻を記述できます
        "measure": [
            {"ch":1, "name":"sensor1", "datas":[100, 9.8, 0]},
            {"ch":2, "name":"sensor2", "datas":[55, 1.0, 0]}
        ]
    },
    {
        "time":datetime(2026, 6, 25, 11, 0, 0),  # datetimeオブジェクトを直接渡すことも可能です
        "measure":[
            {"ch":1, "name":"sensor1", "datas":[103, 10.1, 0]},
            {"ch":2, "name":"sensor2", "datas":[999999, 19999.88, 1]}
        ]
    }
]

組み立てたデータの送信

データ全体の要素数が大きい場合には、内部で自動的にレコード単位で適切な件数ごとにパッケージングし、複数回に分けて送信します。

from kyowacloudfield import KcfClient

# クライアントの初期化 (APIバージョンは1を指定)
kcf = KcfClient(apikey="YOUR_API_KEY", api_version=1)

# アップロード実行
try:
    response = kcf.upload(
        serial="abc123456",
        test="experiment_01",
        data_names=["raw_data", "engineering_value", "status"],
        time_series_datas=upload_datas
    )
    print(f"送信成功")
    print(f"所要時間: {response.average_elapsed_seconds}秒")
    print(f"レスポンスボディ: {response.body}")

except Exception as e:
    print(f"エラーが発生しました: {e}")

3. pandas.DataFrame からの送信（拡張機能）

pip install kyowa-cloud-field[pandas] でインストールした場合、DataFrameの形状（縦長、横並び）に合わせて以下の2つの専用メソッドを利用できます。

パターンA：標準的な時系列テーブル (upload_dataframe)

時刻、チャンネル、計測データが縦に並んでいるCSV（sample.csvなど）をアップロードします。

import pandas as pd
from kyowacloudfield import KcfClient

df = pd.read_csv("sample.csv")
kcf = KcfClient(apikey="YOUR_API_KEY", api_version=1)

response = kcf.upload_dataframe(
    serial="abc123456",
    test="dataframe_test",
    df=df,
    time_col="time",                         # 時刻が含まれる列名
    ch_col="ch",                             # チャンネル番号が含まれる列名
    name_col="name",                         # チャンネル名称が含まれる列名
    data_cols=["raw_data", "engineering_value"]  # アップロードする計測値の列名
)

パターンB：時刻に対してデータが右に並ぶテーブル (upload_wide_dataframe)

1行にその時刻の全チャンネルのデータが横に展開されているマルチインデックス形式のCSV（sample2.csvなど）をアップロードします。

import pandas as pd
from kyowacloudfield import KcfClient

# 3行の階層ヘッダー(ch, name, 項目名)を読み込み、先頭のtime列をインデックスにする
df = pd.read_csv("sample2.csv", header=[0, 1, 2], index_col=0)
kcf = KcfClient(apikey="YOUR_API_KEY", api_version=1)

response = kcf.upload_wide_dataframe(
    serial="abc123456",
    test="wide_dataframe_test",
    df=df,
    data_names=["raw_data", "engineering_value"]  # 抽出して登録したい項目名
)

4. エラーハンドリングとリトライ

本ライブラリの通信中にエラーが発生した場合、内部の例外はすべて KcfError の派生クラスにラップされて送出されます。

from kyowacloudfield import KcfClient, KcfError, KcfNetworkError

kcf = KcfClient(apikey="YOUR_API_KEY")

try:
    response = kcf.upload(...)
    # upload_count を見れば、内部で何分割されて送信されたかが分かります
    if response.upload_count > 1:
        print(f"データサイズが大きいため、内部で {response.upload_count} 回に分割して送信されました。")
except KcfNetworkError as e:
    # ネットワークタイムアウトや、APIサーバーから4xx/5xxエラーが返された場合
    print(f"通信エラーが発生しました: {e}")
    # e.unsent_data から、未送信のデータリストを取得して、そのままリトライに回すことも可能です
except KcfError as e:
    # ライブラリ共通の基底エラー（データの型間違い、バリデーション違反など）
    print(f"ライブラリ内でエラーが発生しました: {e}")

5. レスポンス・例外の詳細プロパティ仕様

正常終了時（KcfResponse）および通信エラー時（KcfNetworkError）には、以下のプロパティが提供されます。一括送信・分割送信を問わず共通のインターフェースで統計情報やサーバーの応答を取得できます。

正常レスポンス (KcfResponse)

kcf.upload() などのメソッドが正常に完了した際に返されるオブジェクトです。
分割送信されていなければ、max_elapsed_seconds、min_elapsed_seconds、average_elapsed_seconds は同一の値となります。

プロパティ名	型	説明
success	bool	送信がすべて成功したか（常に True）
total_records	int	アップロードに成功した総レコード（時刻）数
upload_count	int	実際にサーバーへ送信（リクエスト）した回数（一括なら 1）
max_elapsed_seconds	float	最大アップロード時間（秒）
min_elapsed_seconds	float	最小アップロード時間（秒）
average_elapsed_seconds	float	平均アップロード時間（秒）
body	dict	最後にサーバーから返ってきたレスポンスボディ（JSON形式）

通信例外 (KcfNetworkError)

ネットワークタイムアウトや、サーバーから200番台以外のステータスコードが返された場合にスローされる例外です。

プロパティ名	型	説明
sent_records	int	エラーが発生する直前までに送信成功が確定したレコード数
upload_count	int	エラーが発生したバッチも含めて、試みた累積リクエスト回数
unsent_data	list	サーバーに送信できなかった残りの未送信データリスト（そのまま再送に利用可能）
max_elapsed_seconds	float	エラー発生前までに計測できたリクエストの最大所要時間（秒）
min_elapsed_seconds	float	エラー発生前までに計測できたリクエストの最小所要時間（秒）
average_elapsed_seconds	float	エラー発生前までに計測できたリクエストの平均アップロード時間（秒）
body	dict	エラーを発生させたリクエストに対するサーバーからの応答ボディ（JSON形式）

6. ライセンス

MIT License に基づいて公開されています。