mqtt.go 24 KB

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