energy_storage/src/app/Station.cpp

#include "Station.h"
#include "database/DAO.h"
#include "database/SQL.h"
#include "common/fields.h"
#include "app/Device.h"
#include "common/Spdlogger.h"
#include "common/Utils.h"
#include "protocol/MqttEntity.h"
#include "common/JsonN.h"
#include "app/Config.h"
#include "common/Snowflake.h"
#include "app/DataStruct.h"

Station::Station() : stationId(0)
{
    mqttCli = std::make_shared<MqttClient>();
}

void Station::setFields(Fields& fields)
{
    this->stationId = fields.get<int>(DMStation::STATION_ID);
    this->name = fields.value(DMStation::NAME);
    this->capacity = fields.get<double>(DMStation::CAPACITY);
    this->workMode = fields.get<int>(DMStation::WORK_MODE);
    this->code = fields.value(DMStation::CODE);
    this->status = fields.get<int>(DMStation::STATUS);
    
    this->operationDate = fields.value(DMStation::OPERATION_DATE);
    this->launchDate = fields.value("operation_date");
    this->policy.setFields(fields);
}

void Station::addDevice(int deviceId, std::shared_ptr<Device> device)
{
    mapDevice[deviceId] = device;
}

void Station::addDevice(Fields& fields)
{
    int deviceId = fields.get<int>(DMDevice::DEVICE_ID);
    int stationId = fields.get<int>(DMDevice::STATION_ID);
    if (mapDevice.find(deviceId) != mapDevice.end())
    {
        mapDevice[deviceId]->setFields(fields);
    }
    else
    {
        auto device = Device::create(fields);
        mapDevice[deviceId] = device;
    }
}

std::shared_ptr<Device> Station::getDevice(int deviceId)
{
    auto iter = mapDevice.find(deviceId);
    if (iter!=mapDevice.end())
    {
        return iter->second;
    }
    return nullptr;
}

void Station::groupDevice()
{
    for (auto iter = mapDevice.begin(); iter!=mapDevice.end(); ++iter)
    {
        auto& device = iter->second;
        char key[20] = {};
        sprintf(key, "%03d_%03d_%04d", device->stationId, device->sortNo, device->deviceId);
        mapDeviceGroup[device->category][key] = device;
    }
}

std::shared_ptr<Device> Station::getDeviceByType(int deviceType, std::string code)
{
    for (auto iter = mapDevice.begin(); iter!=mapDevice.end(); ++iter)
    {
        auto device = iter->second;
        if (device->type == deviceType && device->code == code)
        {
            return device;
        }
    }
    return nullptr;
}

void Station::getDeviceByType(int deviceType, std::vector<std::shared_ptr<Device>>& res)
{
    for (auto iter = mapDevice.begin(); iter!=mapDevice.end(); ++iter)
    {
        auto device = iter->second;
        if (device->type == deviceType)
        {
            res.push_back(device);
        }
    }
}

int Station::getDeviceCount(int category)
{
    auto iter = mapDeviceGroup.find(category);
    if (iter != mapDeviceGroup.end())
    {
        return iter->second.size();
    }
    return 0;
}

void Station::getDeviceByCategory(int category, std::vector<std::shared_ptr<Device>>& res)
{
    auto iter = mapDeviceGroup.find(category);
    if (iter != mapDeviceGroup.end())
    {
        res.resize(iter->second.size());
        int i = 0;
        for (auto& item: iter->second)
        {
            res[i++] = item.second;
        }
    }
}

void Station::setWorkMode(int modeId)
{
    this->workMode = modeId;
    std::string sql = SQL(SQL::TYPE::update).table(DMStation::TABLENAME)
        .update(DMStation::WORK_MODE, std::to_string(modeId))
        .where(DMStation::STATION_ID + "=" + std::to_string(stationId)).str();
    Errcode err = DAO::exec(NULL, sql);
    if (err != Errcode::OK)
    {
        spdlog::error("set station work mode failed.");
    }
}

void Station::setPolicy(int policyId)
{
    std::string sql = SQL(SQL::TYPE::update).table(DMStation::TABLENAME)
        .update(DMStation::POLICY_ID, std::to_string(policyId))
        .where(DMStation::STATION_ID + "=" + std::to_string(stationId)).str();
    Errcode err = DAO::exec(NULL, sql);
    if (err != Errcode::OK)
    {
        spdlog::error("set station policy failed.");
    }
}

void Station::initMqtt()
{
    if (status!=0 && mqttCli)
    {
        auto& optionMqtt = Config::option.mqtt;
        mqttCli->init(optionMqtt.host, code, optionMqtt.username, optionMqtt.password);
    }
}

void Station::polling()
{
    if (status > 0 && mqttCli)
    {
        if (mqttCli->isConnected)
        {
            mqttCli->polling();
        }
        //else
        //{
        //    // 该函数检查连接状态，若已经连接，则无操作；若未连接，则进行连接操作
        //    this->initMqtt();
        //}
    }
}

int64_t Station::getPollingTS()
{
    return (mqttCli != nullptr) ? mqttCli->tsPolling : 0;
}

void Station::setGarewayWorkMode()
{
    if (!mqttCli)
    {
        return;
    }

    njson json;
    json["ts"] = Utils::time();
    json["no"] = 1; // 设备编号
    json["40001"] = this->workMode;
   
    if (policy.type == 1) 
    {
        json["40002"] = njson::array();  // 峰谷套利
        policy.getGatewayJsonPeriods(json["40002"]);
    }
    else if (policy.type == 5)
    {
        json["40021"] = njson::array();  // 自定时段
        policy.getGatewayJsonPeriods(json["40021"]);
    }

    std::string text = json.dump();
    spdlog::info(text);
    mqttCli->publish("Gateway_YT", text);
}


string Station::getGatewayMode()
{
    // 0：手动，1：峰谷套利，2：增网配容，3：应急供电，4：并网保电，5：自定时段
    if (workModeGateway == 0) { return "手动"; }
    else if (workModeGateway == 1) { return "峰谷套利"; }
    else if (workModeGateway == 2) { return "增网配容"; }
    else if (workModeGateway == 3) { return "应急供电"; }
    else if (workModeGateway == 4) { return "并网保电"; }
    else if (workModeGateway == 5) { return "自定时段"; }
    else { return "--"; };
}

void Station::checkDevice()
{
    for (auto& item: mapDevice)
    {
        auto& device = item.second;
        if (device)
        {
            if (Utils::time() - device->ts > 60*6)
            {
                device->online = 0;
            }
        }
    }
}

void Station::readAlert(std::shared_ptr<Device> device, std::string addr, int v, std::string text)
{
    int64_t ts = Utils::time();
    std::string alertId = std::to_string(device->deviceId) + "_" + addr;
    int tsCache = mapAlertCache[alertId];
    if (ts - tsCache > 60*5)
    {
        Fields fields;
        fields.set("log_id", Snowflake::instance().getIdStr());
        if (device) { fields.set("device_id", device->deviceId); }
        fields.set("type", int(EAlertType::DEVICE));
        fields.set("content", text + ":故障(" + std::to_string(v) + ")");
        fields.set("status", 1);
        auto dao = DaoEntity::create("log_alert");
        dao->insertFields(fields);
        mapAlertCache[alertId] = ts;
    }
}

void Station::readRuntimeData(int deviceNo, string addr, int val)
{
    if (deviceNo == 1)
    {
        if (addr == "0x000B") { this->voltage = val; }  // A相电压	R	uint32	1V			0x000B
        if (addr == "0x0011") { this->current = val; }  // A相电流	R	int32	1A			0x0011
        if (addr == "0x0011") { this->power = val; }    // 三相总有功	R	int32	1kW			0x0023
    }
    else if (deviceNo == 2)
    {
        statData.ts = Utils::time();
        if (addr == "0x002F") { statData.dayElectIn = val; }		    //日充电电量	R	uint32	1kWh			0x002F
        else if (addr == "0x0031") { statData.dayElectOut = val; }	    //日放电电量	R	uint32	1kWh			0x0031
        else if (addr == "0x0033") { statData.dayFeeIn = val; }	    //日充电费用	R	uint32	1RMB			0x0033
        else if (addr == "0x0035") { statData.dayFeeOut = val; }	    //日放电费用	R	uint32	1RMB			0x0035
        else if (addr == "0x0037") { statData.dayIncome = val; }	    //日收益	    R	int32	1RMB			0x0037
        else if (addr == "0x004D") { statData.totalElectIn = val; }	    //总充电电量	R	uint32	1kWh			0x004D
        else if (addr == "0x004F") { statData.totalElectOut = val; }	//总放电电量	R	uint32	1kWh			0x004F
        else if (addr == "0x0051") { statData.totalFeeIn = val; }	//总充电费用	R	uint32	1RMB			0x0051
        else if (addr == "0x0053") { statData.totalFeeOut = val; }	//总放电费用	R	uint32	1RMB			0x0053
        else if (addr == "0x0055") { statData.totalIncome = val; }	    //总收益	    R	int32	1RMB			0x0055
    }
}

void Station::readTHData(int deviceNo, string addr, int val)
{
    auto& unit = mapTempHumUnit[deviceNo];
    if (addr == "0x0001") { ; }         //所属通道号	R	uint16		1		0x0001
    else if (addr == "0x0002") { ; }    //所属温湿度号	R	uint16		1~10		0x0002
    else if (addr == "0x0003") //温度	R	int16	0.1℃			0x0003
    {  
        unit.temp = float(val) * 0.1; 
        if (deviceNo == 1) temperature = unit.temp;
    }     
    else if (addr == "0x0004")  //湿度	R	int16	0.1℃			0x0004
    { 
        unit.hum = float(val) * 0.1; 
        if (deviceNo == 1) humidity = unit.hum;
    }
}

void Station::readFire40Data(int deviceNo, string addr, int val)
{
    auto& unit = mapFire40Unit[deviceNo];

    if (addr == "0x0001") { ; }	 //所属通道号	R	uint16		1~10			0x0001
    else if (addr == "0x0002") { ; }   //主控数量	R	uint16		1			0x0002
    else if (addr == "0x0003") { ; }   //主控ID	R	uint16		1			0x0003
    else if (addr == "0x0004") { unit.statusMain = val; }   //主控状态	R	uint16		0：正常 1：预警 2：火警			0x0004
    else if (addr == "0x0005") { ; }   //主控硬件版本	R	uint16[2]		主控硬件版本			0x0005
    else if (addr == "0x0007") { ; }   //主控软件版本	R	uint16[2]		主控软件版本			0x0007
    else if (addr == "0x0009") { ; }   //主电状态	R	uint16		0：使用市电 1：使用备电			0x0009
    else if (addr == "0x000A") { ; }   //备电电流	R	uint32	0.1A				0x000A
    else if (addr == "0x000C") { ; }   //备电电压	R	uint32	0.1V				0x000C
    else if (addr == "0x000E") { ; }   //可用容量	R	uint32	0.01Ah				0x000E
    else if (addr == "0x0010") { ; }   //可充放容量	R	uint32	0.01Ah				0x0010
    else if (addr == "0x0012") { unit.usedAlarm = val; }   //警铃是否使用	R	uint16					0x0012
    else if (addr == "0x0013") { unit.statusAlarm = val; }   //警铃状态	R	uint16		0：无效 1：掉线 2：正常 3：启动			0x0013
    else if (addr == "0x0014") { unit.usedValve = val; }   //瓶头阀是否使用	R	uint16					0x0014
    else if (addr == "0x0015") { unit.statusValve = val; }   //瓶头阀状态	R	uint16		0：无效 1：掉线 2：正常 3：启动			0x0015
    else if (addr == "0x0016") { unit.usedMCP = val; }   //手报是否使用	R	uint16					0x0016
    else if (addr == "0x0017") { unit.statusMCP = val; }   //手报状态	R	uint16		0：无效 1：掉线 2：正常 3：启动			0x0017
    else if (addr == "0x0018") { ; }   //簇控制器数量	R	uint16					0x0018
    else if (addr == "0x0019") { ; }   //复合探测器总数量	R	uint16					0x0019
    else if (addr == "0x001A") { ; }   //烟雾探测器总数量	R	uint16					0x001A
    else if (addr == "0x001B") { ; }   //压力探测器总数量	R	uint16					0x001B
    else if (addr == "0x001C") { ; }   //吸气式探测器总数量	R	uint16					0x001C
    else if (addr == "0x001D") { ; }   //PACK探测器总数量	R	uint16					0x001D
    else if (addr == "0x001E") { ; }   //电池总数量	R	uint16					0x001E
}


void Station::readCoolingData(int deviceNo, string addr, int val)
{
    auto& unit = mapCoolingUnit[deviceNo];

    if (addr == "0x1001") { ; }	 //所属通道号	R	uint16		1		0x1001
    else if (addr == "0x1002") { ; }//    所属冷机号	R	uint16		1~10		0x1002
    else if (addr == "0x1003") { coolingStatus = unit.powerOn = val; }//    开关	R	uint16		0：关机，1：开机		0x1003
    else if (addr == "0x1004") { unit.mode = val; }//    采样模式	R	uint16	0-出水温度 1-电芯温度			0x1004
    else if (addr == "0x1005") { unit.cooling = val; }//    制冷状态	R	uint16		0：关闭, 1：启动		0x1005
    else if (addr == "0x1006") { unit.heating = val; }//    制热状态	R	uint16		0：关闭, 1：启动		0x1006
    else if (addr == "0x1007") { unit.highTemp = val; }//    高温告警	R	uint16		0：正常，1：告警		0x1007
    else if (addr == "0x1008") { unit.lowTemp = val; }//    低温告警	R	uint16		0：正常，1：告警		0x1008
    else if (addr == "0x1009") { unit.highPressure = val; }//    高压告警	R	uint16		0：正常，1：告警		0x1009
    else if (addr == "0x100A") { unit.lowPressure = val; }//    低压告警	R	uint16		0：正常，1：告警		0x100A
    else if (addr == "0x100B") { ; }//    进水温度传感器	R	uint16		0：正常，1：告警		0x100B
    else if (addr == "0x100C") { ; }//    出水温度传感器	R	uint16		0：正常，1：告警		0x100C
    else if (addr == "0x100D") { ; }//    进水压力传感器	R	uint16		0：正常，1：告警		0x100D
    else if (addr == "0x100E") { ; }//    出水压力传感器	R	uint16		0：正常，1：告警		0x100E
}

void Station::readGatewayMode(int mode)
{
    this->workModeGateway = mode;
    if (mode != this->workMode)
    {
        //this->setGarewayWorkMode();
    }
}

void Station::readGatewayStatus(int cdzStatus, int emuStatus)
{
    //充电桩 1：在线，0：离线
    if (cdzStatus >= 0)
    {
        if (cdzStatus != this->cdzStatus)
        {
            std::string text = "场站[" + name + "(" + std::to_string(stationId) + ")]充电桩状态变化：" + (cdzStatus>0 ? "在线" : "离线");
            if (this->cdzStatus < 0) { text = "系统启动，" + text; }
            DAO::insertSystemLogDevice(stationId, 0, text, cdzStatus);
            this->cdzStatus = cdzStatus;
        }
    }
    //储能 1：在线，0：离线
    if (emuStatus >= 0)
    {
        if (emuStatus != this->emuStatus)
        {
            std::string text = "场站[" + name + "(" + std::to_string(stationId) + ")]储能EMU状态变化：" + (emuStatus>0 ? "在线" : "离线");
            if (this->emuStatus < 0) { text = "系统启动，" + text; }
            DAO::insertSystemLogDevice(stationId, 0, text, emuStatus);
            this->emuStatus = emuStatus;
        }
    }
}

static std::string MapValueToJson(int npos, std::map<int, float>& mapV)
{
    njson jsonarray = njson::array();
    for (int i = 0; i<=npos; i++)
    {
        jsonarray.push_back(int(mapV[i]));
    }
    return jsonarray.dump();
}
void Station::setCache(int datatype, std::vector<float>& vd)
{
    std::map<int, float>* mapptr = NULL;
    if (datatype == 1) { mapptr = &mapCacheElectIn; }
    else if (datatype == 2) { mapptr = &mapCacheElectOut; }
    else if (datatype == 3) { mapptr = &mapCacheElectCharger; }
    if (mapptr)
    {
        const int step = 600;
        const int N = 86400/step;
        int64_t tsSeconds = Utils::timeDaySeconds();
        int npos = tsSeconds / step;
        for (int i = 0; i<N; ++i)
        {
            if (i < vd.size()) { (*mapptr)[i] = vd[i]; }
            else if (i <= npos) { (*mapptr)[i] = 0; }
        }
    }
}
void Station::cache()
{
    int64_t tDaySeconds = Utils::timeDaySeconds();
    int npos = tDaySeconds / 600;
    int offset = tDaySeconds % 600;
    bool save = false;
    if (offset >= (600-180) && npos + 1 < 144)
    {
        npos += 1;
        save = true;
    }
    else if (offset <= 180 && posCache < npos)
    {
        save = true;
        posCache = npos;
    }
    if (save)
    {
        mapCacheElectIn[npos] = Utils::random(100, 800);    // dayElectIn
        mapCacheElectOut[npos] = Utils::random(100, 800);   // dayElectOut
        mapCacheElectCharger[npos] = Utils::random(100, 800); // 暂无数据源
    }
}

void Station::writeStatistic()
{
    auto dao = DaoEntity::create("history_day");
    std::string dt = Utils::dateStr();
    int64_t tTime = Utils::time();
    int64_t tDate = Utils::date();
    int64_t tDelta = tTime - tDate;
    int npos = (tTime-tDate) / 600;

    for (auto iter = mapDevice.begin(); iter!=mapDevice.end(); ++iter)
    {
        auto device = iter->second;
        if (device->cache(npos))
        {
            Fields fields;
            fields.set("dt", dt);
            fields.set("station_id", this->stationId);
            fields.set("device_id", device->deviceId);
            fields.set("datatype", 1);
            fields.set("value", MapValueToJson(npos, device->mapCacheVoltage));
            DAO::insertRuntimeData(dao, fields);

            fields.set("datatype", 2);
            fields.set("value", MapValueToJson(npos, device->mapCacheCurrent));
            DAO::insertRuntimeData(dao, fields);

            fields.set("datatype", 3);
            fields.set("value", MapValueToJson(npos, device->mapCachePower));
            DAO::insertRuntimeData(dao, fields);
            //spdlog::info("[device] write runtime date to database, deviceId={}", device->deviceId);
        }
    }

    if (statData.ts > 0)
    {
        Fields fields;
        fields.set("dt", Utils::dateStr(statData.ts));
        fields.set("station_id", this->stationId);
        fields.set("device_id", 0);
        fields.set("elect_in", statData.dayElectIn);
        fields.set("elect_out", statData.dayElectOut);
        fields.set("fee_in", statData.dayFeeIn);
        fields.set("fee_out", statData.dayFeeOut);
        fields.set("income", statData.dayIncome);
        //fields.set("num_in", "");
        //fields.set("num_out", "");
        //fields.set("num_err", "");
        //fields.set("t_in", "");
        //fields.set("t_out", "");
        //fields.set("usage_rate", "");
        fields.set("elect_in_total", statData.totalElectIn);
        fields.set("elect_out_total", statData.totalElectOut);
        fields.set("fee_in_total", statData.totalFeeIn);
        fields.set("fee_out_total", statData.totalFeeOut);
        fields.set("income_total", statData.totalIncome);

        dao->setTableName("stat_storage");
        std::vector<std::string> vecKeys = {
            "elect_in", "elect_out", "num_in", "num_out", "num_err", "t_in", "t_out", "usage_rate", "fee_in", "fee_out",
            "elect_in_total", "elect_out_total", "fee_in_total", "fee_out_total", "income_total"
        };
        dao->duplicateUpdate(fields, vecKeys);

        { // stat_day
            Fields fields;
            fields.set("dt", Utils::dateStr(statData.ts));
            fields.set("station_id", this->stationId);
            fields.set("storage_elect_in", statData.dayElectIn);
            fields.set("storage_elect_out", statData.dayElectOut);
            fields.set("income_elect", statData.dayIncome);
            DAO::insertStatDay(dao, fields);
        }
        {
            Fields fields;
            fields.set("station_id", this->stationId);
            fields.set("elect_in", statData.dayElectIn);
            fields.set("elect_out", statData.dayElectOut);
            fields.set("income", statData.dayIncome);
            dao->setTableName("stat_total");
            dao->duplicateUpdate(fields, {"elect_in", "elect_out", "income"});
        }

    }

    {
        // 预测数据源记录
        dao->setTableName("predict_day");

        Fields fields;
        fields.set("dt", dt);
        fields.set("station_id", stationId);
        fields.set("datatype", 1); // 1：储能充电，2：储能放电，3：充电桩充电，4：发电
        fields.set("value", MapValueToJson(npos, mapCacheElectIn));
        dao->duplicateUpdate(fields, {"value"});

        fields.set("datatype", 2); // 1：储能充电，2：储能放电，3：充电桩充电，4：发电
        fields.set("value", MapValueToJson(npos, mapCacheElectOut));
        dao->duplicateUpdate(fields, {"value"});

        fields.set("datatype", 3); // 1：储能充电，2：储能放电，3：充电桩充电，4：发电
        fields.set("value", MapValueToJson(npos, mapCacheElectCharger));
        dao->duplicateUpdate(fields, {"value"});
    }

}