[Python] Test Version of BiliMon265Py

 

 

### Create @ 2026.05.27
import pandas as pd
import sys
from dotenv import load_dotenv

FOLDERPATH_COMMONLIBRARIES = "E:/DevData/commonlibs" # for Windows
#FOLDERPATH_COMMONLIBRARIES = "/root/commonlibs" # for Ubuntu
try:
    import keyring
    folderPath_CommonLibraries = keyring.get_password("BilientService", "CommonLibraries")
    if not folderPath_CommonLibraries: folderPath_CommonLibraries = FOLDERPATH_COMMONLIBRARIES
except:
    folderPath_CommonLibraries = FOLDERPATH_COMMONLIBRARIES

if folderPath_CommonLibraries not in sys.path:
    sys.path.append(folderPath_CommonLibraries)
    print("Path appended")

envPath = folderPath_CommonLibraries + "/.env"
load_dotenv(envPath)

from BMessenger import bMessenger
from BExcel import bExcel
from BTime import bTime

def Test01(): # 2026.05.27
    from BiliDBEntranceClient import cBiliDbEntranceClient
    print("=" * 70)
    print("BiliMon265Py dbentrance API Client 통합 테스트")
    print("=" * 70)
    
    # 클래스 인스턴스 생성
    client = cBiliDbEntranceClient(
        url="https://bilimon265.bilientsvc.org/dbentrance",
        client_id="BiliMon261",
        secret_key="CODE-20260118",
        db_name="BM265_W01"
    )

    # 1. 테이블 생성 테스트
    # print("[테스트 1] CreateTable")
    # print(client.create_table("BMG00_A4833"))

    # 2-1. 단건 데이터 삽입 테스트 (InsertData)
    # print("[테스트 2-1] InsertData (insert_data)")
    # insert_meta = {"DataType": "Environment", "DataName": "RoomA_Sensor"}
    # insert_val = '{"temperature": 25.1, "humidity": 60}'
    # print(client.insert_data("BMG00_A4833", insert_val, insert_meta))

    # 2-2. 다건 벌크 데이터 삽입 테스트 (InsertData - bulk)
    # print("[테스트 2-2] InsertData (insert_datas)")
    # bulk_data = [
    #     {"DataValue1": '{"temperature": 25.4, "humidity": 62}', "DataType": "Environment", "DataName": "RoomA_Sensor"},
    #     {"DataValue1": '{"temperature": 25.8, "humidity": 65}', "DataType": "Environment", "DataName": "RoomA_Sensor"}
    # ]
    # print(client.insert_datas("BMG00_A4833", bulk_data))

    # 3. 데이터 조회 및 청킹 전체 로드 테스트
    try:
        all_records = client.fetch_all_data(
            table_name="BMG00_A4833",
            batch_size=1000,
            delay_ms=31
        )
        print(f"로드 완료! 총 {len(all_records)}개 수집")
        dataDF_All = pd.DataFrame(all_records)
        dataDF_A0 = dataDF_All[dataDF_All['DataName'].str.startswith("A0:")]
        dataDF_A1 = dataDF_All[dataDF_All['DataName'].str.startswith("A1:")]
        dataDF_A2 = dataDF_All[dataDF_All['DataName'].str.startswith("A2:")]
        dataDF_A3 = dataDF_All[dataDF_All['DataName'].str.startswith("A3:")]
        dataDF_A4 = dataDF_All[dataDF_All['DataName'].str.startswith("A4:")]
        dataDF_A5 = dataDF_All[dataDF_All['DataName'].str.startswith("A5:")]
        dataDF_A6 = dataDF_All[dataDF_All['DataName'].str.startswith("A6:")]
        dataDF_A7 = dataDF_All[dataDF_All['DataName'].str.startswith("A7:")]

        dataInfo = {"All": dataDF_All}
        if not dataDF_A0.empty: dataInfo['A0'] = dataDF_A0
        if not dataDF_A1.empty: dataInfo['A1'] = dataDF_A1
        if not dataDF_A2.empty: dataInfo['A2'] = dataDF_A2
        if not dataDF_A3.empty: dataInfo['A3'] = dataDF_A3
        if not dataDF_A4.empty: dataInfo['A4'] = dataDF_A4
        if not dataDF_A5.empty: dataInfo['A5'] = dataDF_A5
        if not dataDF_A6.empty: dataInfo['A6'] = dataDF_A6
        if not dataDF_A7.empty: dataInfo['A7'] = dataDF_A7

        outFile = "./Test04_4833.xlsx"
        bExcel.WriteSheetInfos(outFile, dataInfo)
    except Exception as e:
        print(f"오류 발생: {e}")

    try:
        all_records = client.fetch_all_data(
            table_name="BMG00_A5101",
            batch_size=1000,
            delay_ms=31
        )
        print(f"로드 완료! 총 {len(all_records)}개 수집")
        dataDF_All = pd.DataFrame(all_records)
        dataDF_A0 = dataDF_All[dataDF_All['DataName'].str.startswith("A0:")]
        dataDF_A1 = dataDF_All[dataDF_All['DataName'].str.startswith("A1:")]
        dataDF_A2 = dataDF_All[dataDF_All['DataName'].str.startswith("A2:")]
        dataDF_A3 = dataDF_All[dataDF_All['DataName'].str.startswith("A3:")]
        dataDF_A4 = dataDF_All[dataDF_All['DataName'].str.startswith("A4:")]
        dataDF_A5 = dataDF_All[dataDF_All['DataName'].str.startswith("A5:")]
        dataDF_A6 = dataDF_All[dataDF_All['DataName'].str.startswith("A6:")]
        dataDF_A7 = dataDF_All[dataDF_All['DataName'].str.startswith("A7:")]

        dataInfo = {"All": dataDF_All}
        if not dataDF_A0.empty: dataInfo['A0'] = dataDF_A0
        if not dataDF_A1.empty: dataInfo['A1'] = dataDF_A1
        if not dataDF_A2.empty: dataInfo['A2'] = dataDF_A2
        if not dataDF_A3.empty: dataInfo['A3'] = dataDF_A3
        if not dataDF_A4.empty: dataInfo['A4'] = dataDF_A4
        if not dataDF_A5.empty: dataInfo['A5'] = dataDF_A5
        if not dataDF_A6.empty: dataInfo['A6'] = dataDF_A6
        if not dataDF_A7.empty: dataInfo['A7'] = dataDF_A7

        outFile = "./Test04_5101.xlsx"
        bExcel.WriteSheetInfos(outFile, dataInfo)
    except Exception as e:
        print(f"오류 발생: {e}")        


# ==========================================
# 모듈 실행 예시
# ==========================================
if __name__ == "__main__":

    Test01()