mqtt.go 24 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778
  1. package api
  2. import (
  3. "bytes"
  4. "crypto/hmac"
  5. "crypto/sha1"
  6. "crypto/tls"
  7. "crypto/x509"
  8. "encoding/json"
  9. "fmt"
  10. "io/ioutil"
  11. "net/http"
  12. "os"
  13. "strconv"
  14. "time"
  15. "tmr-watch/conf/setting"
  16. "tmr-watch/http/handle/restful"
  17. "tmr-watch/models"
  18. "tmr-watch/pkg/app"
  19. "tmr-watch/pkg/e"
  20. "tmr-watch/pkg/logging"
  21. "github.com/Anderson-Lu/gofasion/gofasion"
  22. "github.com/astaxie/beego/logs"
  23. MQTT "github.com/eclipse/paho.mqtt.golang"
  24. mqtt "github.com/eclipse/paho.mqtt.golang"
  25. "github.com/gin-gonic/gin"
  26. "github.com/robfig/cron"
  27. )
  28. var c mqtt.Client
  29. var pubTopic string
  30. func InitMqttClient() {
  31. if setting.YynserverSetting.FarmId != "" {
  32. c, pubTopic = MqttClient()
  33. deviceHeartbeat(c, pubTopic)
  34. // GetFeedDataFromApi()
  35. // getBarDataFromApi()
  36. // i := 50
  37. // for {
  38. // i--
  39. // if i == 0 {
  40. // break
  41. // }
  42. // now := time.Now().AddDate(0, 0, -i).Format("2006-01-02")
  43. // // now := "2024-05-23"
  44. // fmt.Println(now, time.Now())
  45. // stirPush(c, pubTopic, now)
  46. // dustingPush(c, pubTopic, now)
  47. // // equipmentAccuracyPush(c, pubTopic, now)
  48. // // finishedWeightPush(c, pubTopic, now)
  49. // // feedtempletPush(c, pubTopic)
  50. // // CompletedTrainNumberPush(c, pubTopic, now)
  51. // }
  52. // now := "2024-05-01"
  53. // stirPush(c, pubTopic, now)
  54. // dustingPush(c, pubTopic, now)
  55. // equipmentAccuracyPush(c, pubTopic, now)
  56. // finishedWeightPush(c, pubTopic, now)
  57. // feedtempletPush(c, pubTopic)
  58. // CompletedTrainNumberPush(c, pubTopic, now)
  59. // stirPush(c, pubTopic, "2024-07-01")
  60. // dustingPush(c, pubTopic, "2024-07-01")
  61. // equipmentAccuracyPush(c, pubTopic, "2024-07-01")
  62. // finishedWeightPush(c, pubTopic, "2024-07-01")
  63. // feedtempletPush(c, pubTopic)
  64. // CompletedTrainNumberPush(c, pubTopic, "2024-07-01")
  65. mqttCron := cron.New()
  66. mqttCron.AddFunc("10 07 * * *", func() {
  67. now := time.Now().AddDate(0, 0, -1).Format("2006-01-02")
  68. stirPush(c, pubTopic, now)
  69. dustingPush(c, pubTopic, now)
  70. equipmentAccuracyPush(c, pubTopic, now)
  71. finishedWeightPush(c, pubTopic, now)
  72. feedtempletPush(c, pubTopic)
  73. CompletedTrainNumberPush(c, pubTopic, now)
  74. GetFeedDataFromApi()
  75. getBarDataFromApi()
  76. })
  77. mqttCron.Start()
  78. }
  79. }
  80. func MqttClient() (MQTT.Client, string) {
  81. // set the device info, include product key, device name, and device secret
  82. // var productKey string = "a1NmXfrjL8M"
  83. // var deviceName string = "4776_p_breed"
  84. // var deviceSecret string = "c2591b89adff22e1c9f0fc03363f56a4"
  85. // ProductKey
  86. // DeviceName
  87. // DeviceSecret
  88. var productKey string = setting.YynserverSetting.ProductKey
  89. var deviceName string = setting.YynserverSetting.DeviceName
  90. var deviceSecret string = setting.YynserverSetting.DeviceSecret
  91. // product_key =k03txxLKFae
  92. // device_name =4623_p_breed
  93. // device_secret =d06ababb2b10ba25bca3041e35ac604d
  94. // host = iot-010a5xth.mqtt.iothub.aliyuncs.com
  95. // farmId=1830004623
  96. // heartBeat=18300046234623_p_breed
  97. // TopicName=/k03txxLKFae/4623_p_breed/user/heatwatch/tmrBreed/post
  98. var timeStamp string = strconv.FormatInt(time.Now().UnixNano(), 10)
  99. var clientId string = "go" + setting.YynserverSetting.FarmId
  100. var subTopic string = "/" + productKey + "/" + deviceName + "/user/heatwatch/tmrBreed/get"
  101. var pubTopic string = "/" + productKey + "/" + deviceName + "/user/heatwatch/tmrBreed/post"
  102. // set the login broker url
  103. var raw_broker bytes.Buffer
  104. raw_broker.WriteString("tcp://")
  105. raw_broker.WriteString("iot-010a5xth.mqtt.iothub.aliyuncs.com:1883")
  106. opts := MQTT.NewClientOptions().AddBroker(raw_broker.String())
  107. // calculate the login auth info, and set it into the connection options
  108. auth := calculate_sign(clientId, productKey, deviceName, deviceSecret, timeStamp)
  109. opts.SetClientID(auth.mqttClientId)
  110. opts.SetUsername(auth.username)
  111. opts.SetPassword(auth.password)
  112. opts.SetMaxReconnectInterval(1 * time.Second)
  113. opts.AutoReconnect = true
  114. // opts.SetKeepAlive(60 * 2 * time.Second)
  115. opts.OnConnect = func(c MQTT.Client) {
  116. if token := c.Subscribe(subTopic, 0, feedHeatwatch); token.Wait() && token.Error() != nil {
  117. logging.Error("mqtt Subscribe err: ", token.Error())
  118. os.Exit(1)
  119. }
  120. }
  121. c := mqtt.NewClient(opts)
  122. if token := c.Connect(); token.Wait() && token.Error() != nil {
  123. fmt.Println(token.Error())
  124. os.Exit(1)
  125. }
  126. fmt.Print("Connect aliyun IoT Cloud Sucess\n")
  127. return c, pubTopic
  128. }
  129. func NewTLSConfig() *tls.Config {
  130. // Import trusted certificates from CAfile.pem.
  131. // Alternatively, manually add CA certificates to default openssl CA bundle.
  132. certpool := x509.NewCertPool()
  133. pemCerts, err := ioutil.ReadFile("./x509/root.pem")
  134. if err != nil {
  135. fmt.Println("0. read file error, game over!!")
  136. }
  137. certpool.AppendCertsFromPEM(pemCerts)
  138. // Create tls.Config with desired tls properties
  139. return &tls.Config{
  140. // RootCAs = certs used to verify server cert.
  141. RootCAs: certpool,
  142. // ClientAuth = whether to request cert from server.
  143. // Since the server is set up for SSL, this happens
  144. // anyways.
  145. ClientAuth: tls.NoClientCert,
  146. // ClientCAs = certs used to validate client cert.
  147. ClientCAs: nil,
  148. // InsecureSkipVerify = verify that cert contents
  149. // match server. IP matches what is in cert etc.
  150. InsecureSkipVerify: false,
  151. // Certificates = list of certs client sends to server.
  152. // Certificates: []tls.Certificate{cert},
  153. }
  154. }
  155. // define a function for the default message handler
  156. var f MQTT.MessageHandler = func(client MQTT.Client, msg MQTT.Message) {
  157. fmt.Printf("TOPIC: %s\n", msg.Topic())
  158. fmt.Printf("MSG: %s\n", msg.Payload())
  159. }
  160. type AuthInfo struct {
  161. password, username, mqttClientId string
  162. }
  163. func calculate_sign(clientId, productKey, deviceName, deviceSecret, timeStamp string) AuthInfo {
  164. var raw_passwd bytes.Buffer
  165. raw_passwd.WriteString("clientId" + clientId)
  166. raw_passwd.WriteString("deviceName")
  167. raw_passwd.WriteString(deviceName)
  168. raw_passwd.WriteString("productKey")
  169. raw_passwd.WriteString(productKey)
  170. raw_passwd.WriteString("timestamp")
  171. raw_passwd.WriteString(timeStamp)
  172. fmt.Println(raw_passwd.String())
  173. // hmac, use sha1
  174. mac := hmac.New(sha1.New, []byte(deviceSecret))
  175. mac.Write([]byte(raw_passwd.String()))
  176. password := fmt.Sprintf("%02x", mac.Sum(nil))
  177. fmt.Println(password)
  178. username := deviceName + "&" + productKey
  179. var MQTTClientId bytes.Buffer
  180. MQTTClientId.WriteString(clientId)
  181. // hmac, use sha1; securemode=2 means TLS connection
  182. MQTTClientId.WriteString("|securemode=2,_v=paho-go-1.0.0,signmethod=hmacsha1,timestamp=")
  183. MQTTClientId.WriteString(timeStamp)
  184. MQTTClientId.WriteString("|")
  185. auth := AuthInfo{password: password, username: username, mqttClientId: MQTTClientId.String()}
  186. return auth
  187. }
  188. func feedtempletPush(c MQTT.Client, pubTopic string) {
  189. tx := restful.Engine.NewSession()
  190. defer tx.Close()
  191. dataList, err := tx.SQL(`SELECT
  192. f.id AS recipeId,
  193. f.tname recipeName,
  194. ft.id ingId,
  195. ifnull(fd.fname,fdy.fname) ingName,
  196. if(fd.fname is not null, ft.fweight,fty.fweight) afQty,
  197. ft.sort mixNo,
  198. ifnull(fd.allowratio,fdy.allowratio) allowableError,
  199. ifnull(fd.fclass,fdy.fclass) ingType,
  200. if(fd.fname is not null,ft.fweight * ( fd.dry / 100 ), fty.fweight * ( fdy.dry / 100 )) dmQty,
  201. '' recipeCost
  202. FROM
  203. feedtemplet f
  204. JOIN ftdetail ft ON ft.ftid = f.id
  205. left JOIN feed fd ON fd.id = ft.fid
  206. left join feedtemplet fy on fy.id = ft.preftid
  207. left join ftdetail fty on fty.ftid = fy.id
  208. left JOIN feed fdy ON fdy.id = fty.fid
  209. `).QueryString()
  210. if err != nil {
  211. logs.Error("feedtempletPush-error-1:", err)
  212. return
  213. }
  214. pushStr := `{
  215. "apiId": "getKPTData",
  216. "param": {
  217. "farmId": "%s",
  218. "method":"getfeedtempletinfo",
  219. "rowCount": "1",
  220. "resultData":%s
  221. }
  222. }`
  223. if len(dataList) > 0 {
  224. b, _ := json.Marshal(dataList)
  225. pushStr = fmt.Sprintf(pushStr, setting.YynserverSetting.FarmId, string(b))
  226. // c.Publish(pubTopic, 2, false, pushStr)
  227. token := c.Publish(pubTopic, 2, false, pushStr)
  228. fmt.Println("publish msg: ", pushStr, token.Error())
  229. token.Wait()
  230. // time.Sleep(2 * time.Second)
  231. }
  232. }
  233. func stirPush(c MQTT.Client, pubTopic, date string) {
  234. tx := restful.Engine.NewSession()
  235. defer tx.Close()
  236. dataList, err := tx.SQL(`SELECT
  237. DATE_FORMAT(d.mydate,'%Y-%m-%d') loadDate,
  238. d.sort tmrNo,
  239. d.times loadShift,
  240. d.tempid recipeId,
  241. d.templetname recipeName,
  242. f.feedcode ingId,
  243. d1.fname ingName,
  244. 12 ingType,
  245. f.dry dmPct,
  246. d1.sort mixNo,
  247. d1.feedallowratio allowable_error,
  248. d1.lweight expWeight,
  249. d1.actualweightminus actualWeight,
  250. if((select count(1) from downloadplandtl1 where pid = d.id and sort < d1.sort order by sort desc) >0 ,(select DATE_FORMAT(intime,'%Y-%m-%d %H:%i:%s') from downloadplandtl1 where pid = d.id and sort < d1.sort order by sort desc limit 1),DATE_FORMAT(d.intime,'%Y-%m-%d %H:%i:%s') ) startTime,
  251. DATE_FORMAT(d1.intime,'%Y-%m-%d %H:%i:%s') endTime , ifnull(if(d.driverId !=0 ,(select drivername from driver where id = d.driverId),(SELECT dr.driver FROM dutyrecord dr
  252. WHERE dr.pastureid = d.pastureid AND dr.eqid = d.tmrid and dr.times= d.times AND dr.operatetime <=d.mydate
  253. ORDER BY dr.operatetime DESC LIMIT 1)),'')tmrName
  254. FROM
  255. downloadedplan d
  256. JOIN downloadplandtl1 d1 ON d1.pid = d.id
  257. JOIN feed f ON f.feedcode = d1.feedcode
  258. AND f.pastureid = d.pastureid
  259. WHERE
  260. DATE_FORMAT( d.mydate ,'%Y-%m-%d' ) = ? `, date).QueryString()
  261. if err != nil {
  262. logs.Error("feedtempletPush-error-1:", err)
  263. return
  264. }
  265. pushStr := `{
  266. "apiId": "getKPTData",
  267. "param": {
  268. "farmId": "%s",
  269. "method":"uploadadddata",
  270. "rowCount": "%d",
  271. "resultData":%s
  272. }
  273. }`
  274. if len(dataList) > 0 {
  275. b, _ := json.Marshal(dataList)
  276. pushStr = fmt.Sprintf(pushStr, setting.YynserverSetting.FarmId, len(dataList), string(b))
  277. token := c.Publish(pubTopic, 2, false, pushStr)
  278. fmt.Println("publish msg: ", pushStr, token.Error())
  279. // token.Wait()
  280. // time.Sleep(2 * time.Second)
  281. }
  282. }
  283. // 撒料信息
  284. func dustingPush(c MQTT.Client, pubTopic, date string) {
  285. tx := restful.Engine.NewSession()
  286. defer tx.Close()
  287. dataList, err := tx.SQL(`SELECT
  288. ifnull(if(d.driverId !=0 ,(select drivername from driver where id = d.driverId),(SELECT dr.driver FROM dutyrecord dr
  289. WHERE dr.pastureid = d.pastureid AND dr.eqid = d.tmrid and dr.times= d.times AND dr.operatetime <=d.mydate
  290. ORDER BY dr.operatetime DESC LIMIT 1)),'')tmrName ,
  291. DATE_FORMAT(d.mydate,'%Y-%m-%d') dropDate,
  292. d.sort tmrNo,
  293. d.times dropShift,
  294. d2.fbarid penId,
  295. b.bname penName,
  296. fp.ccount cowCount,
  297. d2.sort feedingNo,
  298. d2.lweight expWeight,
  299. d2.actualweightminus actualWeight,
  300. if((select count(1) from downloadplandtl2 where pid = d.id and sort < d2.sort order by sort desc) >0 ,(select DATE_FORMAT(intime,'%Y-%m-%d %H:%i:%s') from downloadplandtl2 where pid = d.id and sort < d2.sort order by sort desc limit 1), (select DATE_FORMAT(intime,'%Y-%m-%d %H:%i:%s') from downloadplandtl1 where pid = d.id order by sort desc limit 1) ) startTime,
  301. DATE_FORMAT(d2.intime,'%Y-%m-%d %H:%i:%s') endTime
  302. FROM
  303. downloadedplan d
  304. JOIN downloadplandtl2 d2 ON d2.pid = d.id
  305. JOIN bar b ON b.id = d2.fbarid
  306. join feedp fp on fp.barid = b.id
  307. join tmr t on t.id = d.tmrid
  308. left join driver on driver.drivercode = t.eqcode
  309. WHERE
  310. DATE_FORMAT( d.mydate ,'%Y-%m-%d' ) = ? `, date).QueryString()
  311. if err != nil {
  312. logs.Error("feedtempletPush-error-1:", err)
  313. return
  314. }
  315. pushStr := `{
  316. "apiId": "getKPTData",
  317. "param": {
  318. "farmId": "%s",
  319. "method":"uploaddiliverdata",
  320. "rowCount": "%d",
  321. "resultData":%s
  322. }
  323. }`
  324. if len(dataList) > 0 {
  325. b, _ := json.Marshal(dataList)
  326. pushStr = fmt.Sprintf(pushStr, setting.YynserverSetting.FarmId, len(dataList), string(b))
  327. token := c.Publish(pubTopic, 2, false, pushStr)
  328. fmt.Println("publish msg: ", pushStr, token.Error())
  329. // token.Wait()
  330. // time.Sleep(2 * time.Second)
  331. }
  332. }
  333. //设备心跳
  334. func deviceHeartbeat(c MQTT.Client, pubTopic string) {
  335. pushStr := fmt.Sprintf(`{"data_collect_number":%s,"status":true,"model_type":"heartbeat"}`, setting.YynserverSetting.HeartBeat)
  336. token := c.Publish(pubTopic, 2, false, pushStr)
  337. fmt.Println("publish msg: ", pushStr, token.Error())
  338. // token.Wait()
  339. // go func() {
  340. duetimecst2, _ := time.ParseInLocation("15:04:05", "00:01:00", time.Local)
  341. duetimecst3, _ := time.ParseInLocation("15:04:05", "00:00:00", time.Local)
  342. spec1 := fmt.Sprintf("@every %v", duetimecst2.Sub(duetimecst3))
  343. // for {
  344. device := cron.New()
  345. device.AddFunc(spec1, func() {
  346. token := c.Publish(pubTopic, 2, false, pushStr)
  347. fmt.Println("publish msg: ", pushStr, token.Error(), time.Now())
  348. token.Wait()
  349. })
  350. // }
  351. device.Start()
  352. // }()
  353. }
  354. // 准确率
  355. func equipmentAccuracyPush(c MQTT.Client, pubTopic, date string) {
  356. tx := restful.Engine.NewSession()
  357. defer tx.Close()
  358. dataList, err := tx.SQL(`SELECT
  359. t.tname Name,
  360. 1-abs (
  361. sum( d1.actualweightminus )- sum( d1.lweight ))/ sum( d1.lweight ) Rate,
  362. d.mydate RateDate , DATE_FORMAT(d.intime,'%Y-%m-%d %H:%i:%s') startTime,
  363. (select DATE_FORMAT(intime,'%Y-%m-%d %H:%i:%s') from downloadplandtl2 where pid = d.id order by sort desc limit 1) endTime
  364. FROM
  365. downloadedplan d
  366. JOIN downloadplandtl1 d1 ON d1.pid = d.id
  367. JOIN tmr t ON t.datacaptureno = d.datacaptureno
  368. WHERE
  369. DATE_FORMAT( d.mydate, '%Y-%m-%d' ) = ? and d.lpplantype in(0,1)
  370. GROUP BY
  371. d.datacaptureno`, date).QueryString()
  372. if err != nil {
  373. logs.Error("feedtempletPush-error-1:", err)
  374. return
  375. }
  376. pushStr := `{
  377. "apiId": "getKPTData",
  378. "param": {
  379. "resultData": %s,
  380. "farmId": "%s",
  381. "method": "uploadrate",
  382. "rowCount": %d
  383. }
  384. }`
  385. if len(dataList) > 0 {
  386. b, _ := json.Marshal(dataList)
  387. pushStr = fmt.Sprintf(pushStr, string(b), setting.YynserverSetting.FarmId, len(dataList))
  388. token := c.Publish(pubTopic, 2, false, pushStr)
  389. fmt.Println("publish msg: ", pushStr, token.Error())
  390. token.Wait()
  391. // time.Sleep(2 * time.Second)
  392. }
  393. }
  394. // 完成重量
  395. func finishedWeightPush(c MQTT.Client, pubTopic, date string) {
  396. tx := restful.Engine.NewSession()
  397. defer tx.Close()
  398. dataList, err := tx.SQL(`SELECT
  399. sum( d1.actualweightminus ) completeWeight,
  400. sum( d1.lweight ) planWeight,
  401. d.mydate weightDate , DATE_FORMAT(d.intime,'%Y-%m-%d %H:%i:%s') startTime,
  402. (select DATE_FORMAT(intime,'%Y-%m-%d %H:%i:%s') from downloadedplan where mydate = d.mydate and intime is not null order by sort desc limit 1) endTime
  403. FROM
  404. downloadedplan d
  405. JOIN downloadplandtl1 d1 ON d1.pid = d.id
  406. JOIN tmr t ON t.datacaptureno = d.datacaptureno
  407. WHERE
  408. DATE_FORMAT( d.mydate, '%Y-%m-%d' ) = ? and lpplantype in(0,1)`, date).QueryString()
  409. if err != nil {
  410. logs.Error("feedtempletPush-error-1:", err)
  411. return
  412. }
  413. pushStr := `{
  414. "apiId": "getKPTData",
  415. "param": {
  416. "resultData": %s,
  417. "farmId": "%s",
  418. "method": "uploadweight",
  419. "rowCount": "1"
  420. }
  421. }`
  422. if len(dataList) > 0 {
  423. b, _ := json.Marshal(dataList)
  424. pushStr = fmt.Sprintf(pushStr, string(b), setting.YynserverSetting.FarmId)
  425. token := c.Publish(pubTopic, 2, false, pushStr)
  426. fmt.Println("publish msg: ", pushStr, token.Error())
  427. token.Wait()
  428. // time.Sleep(2 * time.Second)
  429. }
  430. }
  431. // 完成车次
  432. func CompletedTrainNumberPush(c MQTT.Client, pubTopic, date string) {
  433. tx := restful.Engine.NewSession()
  434. defer tx.Close()
  435. dataList, err := tx.SQL(` select (select count(1) from downloadedplan where DATE_FORMAT(mydate ,'%Y-%m-%d' ) = ? and lpplantype in(0,1) ) planCar,
  436. (select count(1) from downloadedplan where DATE_FORMAT(mydate ,'%Y-%m-%d' ) = ? and iscompleted = 1 and lpplantype in(0,1)) CompleteCar ,
  437. ? carDate,
  438. (select DATE_FORMAT(intime,'%Y-%m-%d %H:%i:%s') from downloadedplan where DATE_FORMAT(mydate ,'%Y-%m-%d' ) = ? and iscompleted = 1 and lpplantype in(0,1) order by intime asc limit 1 ) startTime,
  439. (select DATE_FORMAT(intime,'%Y-%m-%d %H:%i:%s') from downloadedplan where DATE_FORMAT(mydate ,'%Y-%m-%d' ) = ? and iscompleted = 1 and lpplantype in(0,1) and intime is not null order by intime desc limit 1 ) endTime
  440. `, date, date, date, date, date).QueryString()
  441. if err != nil {
  442. logs.Error("feedtempletPush-error-1:", err)
  443. return
  444. }
  445. pushStr := `{
  446. "apiId": "getKPTData",
  447. "param": {
  448. "resultData": %s,
  449. "farmId": "%s",
  450. "method": "uploadcarnumber",
  451. "rowCount": "1"
  452. }
  453. }`
  454. if len(dataList) > 0 {
  455. b, _ := json.Marshal(dataList)
  456. pushStr = fmt.Sprintf(pushStr, string(b), setting.YynserverSetting.FarmId)
  457. token := c.Publish(pubTopic, 2, false, pushStr)
  458. fmt.Println("publish msg: ", pushStr, token.Error())
  459. token.Wait()
  460. // time.Sleep(2 * time.Second)
  461. }
  462. }
  463. func feedHeatwatch(client MQTT.Client, msg MQTT.Message) {
  464. tx := restful.Engine.NewSession()
  465. defer tx.Close()
  466. data := make(map[string]interface{})
  467. json.Unmarshal(msg.Payload(), &data)
  468. if _, ok := data["feedData"]; ok {
  469. for _, item := range data["feedData"].([]map[string]interface{}) {
  470. tx.SQL(` insert into feed(pastureid,feedcode,fname)values((SELECT column_default INTO pastureidTem FROM information_schema.COLUMNS
  471. WHERE table_name = 'recweight' AND table_schema = 'tmrwatch2' AND column_name = 'pastureid'),?,?)
  472. ON DUPLICATE KEY UPDATE feedcode = ?,fname = ? `, item["feedCode"], item["feedName"], item["feedCode"], item["feedName"]).Execute()
  473. }
  474. } else if _, ok := data["barData"]; ok {
  475. for _, item := range data["barData"].([]map[string]interface{}) {
  476. tx.SQL(` insert into bar(pastureid,bcode,bname)values((SELECT column_default INTO pastureidTem FROM information_schema.COLUMNS
  477. WHERE table_name = 'recweight' AND table_schema = 'tmrwatch2' AND column_name = 'pastureid'),?,?)
  478. ON DUPLICATE KEY UPDATE bcode = ?,bname = ? `, item["barCode"], item["barName"], item["barCode"], item["barName"]).Execute()
  479. }
  480. }
  481. }
  482. func LabourStirPush(cq *gin.Context) {
  483. appG := app.Gin{C: cq}
  484. dataByte, _ := ioutil.ReadAll(cq.Request.Body)
  485. fsion := gofasion.NewFasion(string(dataByte))
  486. date := fsion.Get("date").ValueStr()
  487. stirPush(c, pubTopic, date)
  488. appG.Response(http.StatusOK, e.SUCCESS, true)
  489. }
  490. func LabourDustingPush(cq *gin.Context) {
  491. appG := app.Gin{C: cq}
  492. dataByte, _ := ioutil.ReadAll(cq.Request.Body)
  493. fsion := gofasion.NewFasion(string(dataByte))
  494. date := fsion.Get("date").ValueStr()
  495. dustingPush(c, pubTopic, date)
  496. appG.Response(http.StatusOK, e.SUCCESS, true)
  497. }
  498. func getApiToken() (string, error) {
  499. url := "https://zhmc-api.yunyangniu.com/oauth/oauth/token"
  500. bodyMap := map[string]string{
  501. "grant_type": "client_credentials",
  502. "client_id": "HouseInfo",
  503. "client_secret": "secret",
  504. }
  505. // 构建带参数的 URL
  506. urlWithParams := url + "?"
  507. for key, value := range bodyMap {
  508. urlWithParams += fmt.Sprintf("%s=%s&", key, value)
  509. }
  510. // 移除末尾的 "&"
  511. urlWithParams = urlWithParams[:len(urlWithParams)-1]
  512. // 发送 POST 请求
  513. resp, err := http.Post(urlWithParams, "application/x-www-form-urlencoded", nil)
  514. if err != nil {
  515. return "", err
  516. }
  517. defer resp.Body.Close()
  518. // 读取响应内容
  519. body, err := ioutil.ReadAll(resp.Body)
  520. if err != nil {
  521. return "", err
  522. }
  523. // 解析 JSON 响应
  524. var tokenVo map[string]interface{}
  525. err = json.Unmarshal(body, &tokenVo)
  526. if err != nil {
  527. return "", err
  528. }
  529. // 获取 access_token
  530. token, ok := tokenVo["access_token"].(string)
  531. if !ok {
  532. return "", fmt.Errorf("access_token not found in response")
  533. }
  534. return token, nil
  535. }
  536. func GetFeedDataFromApi() {
  537. // mapResult := make(map[string]interface{})
  538. // 获取 API Token
  539. tokenString, err := getApiToken()
  540. if err != nil {
  541. return
  542. }
  543. if tokenString == "" {
  544. return
  545. }
  546. // 构建 URL
  547. urlString := fmt.Sprintf("https://zhmc-api.yunyangniu.com/duijie/v1/breed/synMaterialInfo?farmNum=%s&secret=%s", setting.YynserverSetting.FarmId, setting.YynserverSetting.YynSecret)
  548. // 构建请求头
  549. headerMap := map[string]string{
  550. "username": "KPT_USER",
  551. "Authorization": "Bearer " + tokenString,
  552. }
  553. // 发送 POST 请求
  554. reDataString, err := sendPost(urlString, headerMap, nil)
  555. if err != nil {
  556. return
  557. }
  558. if reDataString != "" {
  559. var resultMap map[string]interface{}
  560. err = json.Unmarshal([]byte(reDataString), &resultMap)
  561. if err != nil {
  562. return
  563. }
  564. msgCode, ok := resultMap["msgCode"].(string)
  565. if !ok || msgCode != "10000" {
  566. return
  567. }
  568. contentMap, ok := resultMap["content"].(map[string]interface{})
  569. if !ok {
  570. return
  571. }
  572. b, _ := json.Marshal(contentMap["materialInfoList"])
  573. feedList := make([]*models.YYNMaterialInfo, 0)
  574. json.Unmarshal(b, &feedList)
  575. tx := restful.Engine.NewSession()
  576. defer tx.Close()
  577. pastureinfo := new(udPastureInfo)
  578. err := tx.SQL(`select column_default as pastureid,(select werks from pasture where pastureid = column_default) werks from information_schema.COLUMNS
  579. WHERE table_name = 'recweight' AND table_schema = ? AND column_name = 'pastureid'`, setting.DatabaseSetting.Name).GetFirst(pastureinfo).Error
  580. if err != nil {
  581. logs.Error(err)
  582. return
  583. }
  584. for _, feed := range feedList {
  585. ids, err := setting.SnowIds.NextId()
  586. if err != nil {
  587. ids = time.Now().UnixNano()
  588. logging.Info("create SnowIds err", err)
  589. }
  590. _, err = tx.Exec(` insert into feedclass(id,pastureid,fcname,fccode,bigfeedclassid,bigfeedclassname)values(?,?,?,?,?,?) ON DUPLICATE KEY UPDATE backup1 = '云养牛'`,
  591. ids, pastureinfo.Pastureid, feed.CategoryName, feed.CategoryCode, ids, feed.CategoryName)
  592. if err != nil {
  593. logs.Error(err)
  594. return
  595. }
  596. fmt.Println(feed.CategoryName)
  597. ids1, err := setting.SnowIds.NextId()
  598. if err != nil {
  599. ids1 = time.Now().UnixNano()
  600. logging.Info("create SnowIds err", err)
  601. }
  602. tx.Exec(` insert into feed(id,pastureid,fname,feedcode,fclass,fclassid,source)
  603. values(?,?,?,?,?,(select id from feedclass where pastureid = ? and fccode = ? ),'云养牛') ON DUPLICATE KEY UPDATE source = '云养牛' `,
  604. ids1, pastureinfo.Pastureid, feed.MaterialName, feed.MaterialCode, feed.CategoryName, pastureinfo.Pastureid, feed.CategoryCode)
  605. }
  606. }
  607. }
  608. func getBarDataFromApi() {
  609. // 获取 API Token
  610. tokenString, err := getApiToken()
  611. if err != nil {
  612. return
  613. }
  614. if tokenString == "" {
  615. return
  616. }
  617. // 构建 URL
  618. urlString := fmt.Sprintf("https://zhmc-api.yunyangniu.com/duijie/v1/breed/synCowHouseInfo?farmNum=%s", setting.YynserverSetting.FarmId)
  619. urlString += fmt.Sprintf("&secret=%s", setting.YynserverSetting.YynSecret)
  620. // 构建请求头
  621. headerMap := map[string]string{
  622. "username": "KPT_USER",
  623. "Authorization": "Bearer " + tokenString,
  624. }
  625. // 发送 POST 请求
  626. reDataString, err := sendPost(urlString, headerMap, nil)
  627. if err != nil {
  628. return
  629. }
  630. fmt.Println(time.Now(), "牛舍接口返回:", reDataString)
  631. if reDataString != "" {
  632. var resultMap map[string]interface{}
  633. err = json.Unmarshal([]byte(reDataString), &resultMap)
  634. if err != nil {
  635. return
  636. }
  637. msgCode, ok := resultMap["msgCode"].(string)
  638. if !ok || msgCode != "10000" {
  639. return
  640. }
  641. contentMap, ok := resultMap["content"].(map[string]interface{})
  642. if !ok {
  643. return
  644. }
  645. cowHouseList, ok := contentMap["cowHouseList"].([]interface{})
  646. if !ok {
  647. return
  648. }
  649. fmt.Println(time.Now(), "牛舍列表返回:", cowHouseList)
  650. // cowAmount:57 farmName:吉林省牧硕养殖有限公司 farmNum:1830002720 farmUuid:737a6a60094011e9802143b3b3c94e01 houseName:1-3东蹄病
  651. // houseNum:131 houseType:lactation houseTypeName:泌乳牛舍 houseUuid:05f3e0a3127f11efbf42d1859aea01b1
  652. // _, err := tx.SQL(` insert into bar(pastureid,bcode,bname,sort,class,classcode,cattle,cattlecode,sapcode)
  653. // values(?,?,?,(select max(sort)+1 from bar b where b.pastureid = ? ),
  654. // (select distName from dist where distCode = ? and distType = '牛舍类型' ),?,
  655. // (select distName from dist where distCode = ? and distType = '牛群类别'),?,?)
  656. // ON DUPLICATE KEY UPDATE bname = ? ,
  657. // class = (select distName from dist where distCode = ? and distType = '牛舍类型' ) ,classcode = ?,
  658. // cattle = (select distName from dist where distCode = ? and distType = '牛群类别' ) ,cattlecode = ?
  659. // `,
  660. // pastureId, barMap["CHSNO"], fmt.Sprintf("%v_sap", barMap["CHSTX"]), pastureId, barMap["CHSTY"], barMap["CHSTY"],
  661. // barMap["FCWTS"], barMap["FCWTS"], barMap["CHSNO"],
  662. // fmt.Sprintf("%v_sap", barMap["CHSTX"]), barMap["CHSTY"], barMap["CHSTY"], barMap["FCWTS"], barMap["FCWTS"]).Execute()
  663. // if err != nil {
  664. // logs.Error(err)
  665. // return
  666. //}
  667. }
  668. }
  669. func sendPost(url string, headers map[string]string, body []byte) (string, error) {
  670. req, err := http.NewRequest("POST", url, bytes.NewBuffer(body))
  671. if err != nil {
  672. return "", err
  673. }
  674. for key, value := range headers {
  675. req.Header.Set(key, value)
  676. }
  677. client := &http.Client{}
  678. resp, err := client.Do(req)
  679. if err != nil {
  680. return "", err
  681. }
  682. defer resp.Body.Close()
  683. respBody, err := ioutil.ReadAll(resp.Body)
  684. if err != nil {
  685. return "", err
  686. }
  687. return string(respBody), nil
  688. }