mqtt.go 24 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779
  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. GetFeedDataFromApi()
  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. 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. //f.sapCode ingId,
  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.id = d1.fid
  258. AND f.pastureid = d.pastureid
  259. WHERE
  260. DATE_FORMAT( d.mydate ,'%Y-%m-%d' ) = ? and f.source = '云养牛' `, date).QueryString()
  261. // DATE_FORMAT( d.mydate ,'%Y-%m-%d' ) = ? and f.sapCode is not null and f.sapCode != '' `, date).QueryString()
  262. if err != nil {
  263. logs.Error("feedtempletPush-error-1:", err)
  264. return
  265. }
  266. pushStr := `{
  267. "apiId": "getKPTData",
  268. "param": {
  269. "farmId": "%s",
  270. "method":"uploadadddata",
  271. "rowCount": "%d",
  272. "resultData":%s
  273. }
  274. }`
  275. if len(dataList) > 0 {
  276. b, _ := json.Marshal(dataList)
  277. pushStr = fmt.Sprintf(pushStr, setting.YynserverSetting.FarmId, len(dataList), string(b))
  278. token := c.Publish(pubTopic, 2, false, pushStr)
  279. fmt.Println("publish msg: ", pushStr, token.Error())
  280. // token.Wait()
  281. // time.Sleep(2 * time.Second)
  282. }
  283. }
  284. // 撒料信息
  285. func dustingPush(c MQTT.Client, pubTopic, date string) {
  286. tx := restful.Engine.NewSession()
  287. defer tx.Close()
  288. dataList, err := tx.SQL(`SELECT
  289. ifnull(if(d.driverId !=0 ,(select drivername from driver where id = d.driverId),(SELECT dr.driver FROM dutyrecord dr
  290. WHERE dr.pastureid = d.pastureid AND dr.eqid = d.tmrid and dr.times= d.times AND dr.operatetime <=d.mydate
  291. ORDER BY dr.operatetime DESC LIMIT 1)),'')tmrName ,
  292. DATE_FORMAT(d.mydate,'%Y-%m-%d') dropDate,
  293. d.sort tmrNo,
  294. d.times dropShift,
  295. d2.fbarid penId,
  296. b.bname penName,
  297. fp.ccount cowCount,
  298. d2.sort feedingNo,
  299. d2.lweight expWeight,
  300. d2.actualweightminus actualWeight,
  301. 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,
  302. DATE_FORMAT(d2.intime,'%Y-%m-%d %H:%i:%s') endTime
  303. FROM
  304. downloadedplan d
  305. JOIN downloadplandtl2 d2 ON d2.pid = d.id
  306. JOIN bar b ON b.id = d2.fbarid
  307. join feedp fp on fp.barid = b.id
  308. join tmr t on t.id = d.tmrid
  309. left join driver on driver.drivercode = t.eqcode
  310. WHERE
  311. DATE_FORMAT( d.mydate ,'%Y-%m-%d' ) = ? `, date).QueryString()
  312. if err != nil {
  313. logs.Error("feedtempletPush-error-1:", err)
  314. return
  315. }
  316. pushStr := `{
  317. "apiId": "getKPTData",
  318. "param": {
  319. "farmId": "%s",
  320. "method":"uploaddiliverdata",
  321. "rowCount": "%d",
  322. "resultData":%s
  323. }
  324. }`
  325. if len(dataList) > 0 {
  326. b, _ := json.Marshal(dataList)
  327. pushStr = fmt.Sprintf(pushStr, setting.YynserverSetting.FarmId, len(dataList), string(b))
  328. c.Publish(pubTopic, 2, false, pushStr)
  329. // fmt.Println("publish msg: ", pushStr, token.Error())
  330. // token.Wait()
  331. // time.Sleep(2 * time.Second)
  332. }
  333. }
  334. //设备心跳
  335. func deviceHeartbeat(c MQTT.Client, pubTopic string) {
  336. pushStr := fmt.Sprintf(`{"data_collect_number":%s,"status":true,"model_type":"heartbeat"}`, setting.YynserverSetting.HeartBeat)
  337. c.Publish(pubTopic, 2, false, pushStr)
  338. // fmt.Println("publish msg: ", pushStr, token.Error())
  339. // token.Wait()
  340. // go func() {
  341. duetimecst2, _ := time.ParseInLocation("15:04:05", "00:01:00", time.Local)
  342. duetimecst3, _ := time.ParseInLocation("15:04:05", "00:00:00", time.Local)
  343. spec1 := fmt.Sprintf("@every %v", duetimecst2.Sub(duetimecst3))
  344. // for {
  345. device := cron.New()
  346. device.AddFunc(spec1, func() {
  347. token := c.Publish(pubTopic, 2, false, pushStr)
  348. // fmt.Println("publish msg: ", pushStr, token.Error(), time.Now())
  349. token.Wait()
  350. })
  351. // }
  352. device.Start()
  353. // }()
  354. }
  355. // 准确率
  356. func equipmentAccuracyPush(c MQTT.Client, pubTopic, date string) {
  357. tx := restful.Engine.NewSession()
  358. defer tx.Close()
  359. dataList, err := tx.SQL(`SELECT
  360. t.tname Name,
  361. 1-abs (
  362. sum( d1.actualweightminus )- sum( d1.lweight ))/ sum( d1.lweight ) Rate,
  363. d.mydate RateDate , DATE_FORMAT(d.intime,'%Y-%m-%d %H:%i:%s') startTime,
  364. (select DATE_FORMAT(intime,'%Y-%m-%d %H:%i:%s') from downloadplandtl2 where pid = d.id order by sort desc limit 1) endTime
  365. FROM
  366. downloadedplan d
  367. JOIN downloadplandtl1 d1 ON d1.pid = d.id
  368. JOIN tmr t ON t.datacaptureno = d.datacaptureno
  369. WHERE
  370. DATE_FORMAT( d.mydate, '%Y-%m-%d' ) = ? and d.lpplantype in(0,1)
  371. GROUP BY
  372. d.datacaptureno`, date).QueryString()
  373. if err != nil {
  374. logs.Error("feedtempletPush-error-1:", err)
  375. return
  376. }
  377. pushStr := `{
  378. "apiId": "getKPTData",
  379. "param": {
  380. "resultData": %s,
  381. "farmId": "%s",
  382. "method": "uploadrate",
  383. "rowCount": %d
  384. }
  385. }`
  386. if len(dataList) > 0 {
  387. b, _ := json.Marshal(dataList)
  388. pushStr = fmt.Sprintf(pushStr, string(b), setting.YynserverSetting.FarmId, len(dataList))
  389. token := c.Publish(pubTopic, 2, false, pushStr)
  390. // fmt.Println("publish msg: ", pushStr, token.Error())
  391. token.Wait()
  392. // time.Sleep(2 * time.Second)
  393. }
  394. }
  395. // 完成重量
  396. func finishedWeightPush(c MQTT.Client, pubTopic, date string) {
  397. tx := restful.Engine.NewSession()
  398. defer tx.Close()
  399. dataList, err := tx.SQL(`SELECT
  400. sum( d1.actualweightminus ) completeWeight,
  401. sum( d1.lweight ) planWeight,
  402. d.mydate weightDate , DATE_FORMAT(d.intime,'%Y-%m-%d %H:%i:%s') startTime,
  403. (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
  404. FROM
  405. downloadedplan d
  406. JOIN downloadplandtl1 d1 ON d1.pid = d.id
  407. JOIN tmr t ON t.datacaptureno = d.datacaptureno
  408. WHERE
  409. DATE_FORMAT( d.mydate, '%Y-%m-%d' ) = ? and lpplantype in(0,1)`, date).QueryString()
  410. if err != nil {
  411. logs.Error("feedtempletPush-error-1:", err)
  412. return
  413. }
  414. pushStr := `{
  415. "apiId": "getKPTData",
  416. "param": {
  417. "resultData": %s,
  418. "farmId": "%s",
  419. "method": "uploadweight",
  420. "rowCount": "1"
  421. }
  422. }`
  423. if len(dataList) > 0 {
  424. b, _ := json.Marshal(dataList)
  425. pushStr = fmt.Sprintf(pushStr, string(b), setting.YynserverSetting.FarmId)
  426. token := c.Publish(pubTopic, 2, false, pushStr)
  427. // fmt.Println("publish msg: ", pushStr, token.Error())
  428. token.Wait()
  429. // time.Sleep(2 * time.Second)
  430. }
  431. }
  432. // 完成车次
  433. func CompletedTrainNumberPush(c MQTT.Client, pubTopic, date string) {
  434. tx := restful.Engine.NewSession()
  435. defer tx.Close()
  436. dataList, err := tx.SQL(` select (select count(1) from downloadedplan where DATE_FORMAT(mydate ,'%Y-%m-%d' ) = ? and lpplantype in(0,1) ) planCar,
  437. (select count(1) from downloadedplan where DATE_FORMAT(mydate ,'%Y-%m-%d' ) = ? and iscompleted = 1 and lpplantype in(0,1)) CompleteCar ,
  438. ? carDate,
  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) order by intime asc limit 1 ) startTime,
  440. (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
  441. `, date, date, date, date, date).QueryString()
  442. if err != nil {
  443. logs.Error("feedtempletPush-error-1:", err)
  444. return
  445. }
  446. pushStr := `{
  447. "apiId": "getKPTData",
  448. "param": {
  449. "resultData": %s,
  450. "farmId": "%s",
  451. "method": "uploadcarnumber",
  452. "rowCount": "1"
  453. }
  454. }`
  455. if len(dataList) > 0 {
  456. b, _ := json.Marshal(dataList)
  457. pushStr = fmt.Sprintf(pushStr, string(b), setting.YynserverSetting.FarmId)
  458. token := c.Publish(pubTopic, 2, false, pushStr)
  459. // fmt.Println("publish msg: ", pushStr, token.Error())
  460. token.Wait()
  461. // time.Sleep(2 * time.Second)
  462. }
  463. }
  464. func feedHeatwatch(client MQTT.Client, msg MQTT.Message) {
  465. tx := restful.Engine.NewSession()
  466. defer tx.Close()
  467. data := make(map[string]interface{})
  468. json.Unmarshal(msg.Payload(), &data)
  469. if _, ok := data["feedData"]; ok {
  470. for _, item := range data["feedData"].([]map[string]interface{}) {
  471. tx.SQL(` insert into feed(pastureid,feedcode,fname)values((SELECT column_default INTO pastureidTem FROM information_schema.COLUMNS
  472. WHERE table_name = 'recweight' AND table_schema = 'tmrwatch2' AND column_name = 'pastureid'),?,?)
  473. ON DUPLICATE KEY UPDATE feedcode = ?,fname = ? `, item["feedCode"], item["feedName"], item["feedCode"], item["feedName"]).Execute()
  474. }
  475. } else if _, ok := data["barData"]; ok {
  476. for _, item := range data["barData"].([]map[string]interface{}) {
  477. tx.SQL(` insert into bar(pastureid,bcode,bname)values((SELECT column_default INTO pastureidTem FROM information_schema.COLUMNS
  478. WHERE table_name = 'recweight' AND table_schema = 'tmrwatch2' AND column_name = 'pastureid'),?,?)
  479. ON DUPLICATE KEY UPDATE bcode = ?,bname = ? `, item["barCode"], item["barName"], item["barCode"], item["barName"]).Execute()
  480. }
  481. }
  482. }
  483. func LabourStirPush(cq *gin.Context) {
  484. appG := app.Gin{C: cq}
  485. dataByte, _ := ioutil.ReadAll(cq.Request.Body)
  486. fsion := gofasion.NewFasion(string(dataByte))
  487. date := fsion.Get("date").ValueStr()
  488. stirPush(c, pubTopic, date)
  489. appG.Response(http.StatusOK, e.SUCCESS, true)
  490. }
  491. func LabourDustingPush(cq *gin.Context) {
  492. appG := app.Gin{C: cq}
  493. dataByte, _ := ioutil.ReadAll(cq.Request.Body)
  494. fsion := gofasion.NewFasion(string(dataByte))
  495. date := fsion.Get("date").ValueStr()
  496. dustingPush(c, pubTopic, date)
  497. appG.Response(http.StatusOK, e.SUCCESS, true)
  498. }
  499. func getApiToken() (string, error) {
  500. url := "https://zhmc-api.yunyangniu.com/oauth/oauth/token"
  501. bodyMap := map[string]string{
  502. "grant_type": "client_credentials",
  503. "client_id": "HouseInfo",
  504. "client_secret": "secret",
  505. }
  506. // 构建带参数的 URL
  507. urlWithParams := url + "?"
  508. for key, value := range bodyMap {
  509. urlWithParams += fmt.Sprintf("%s=%s&", key, value)
  510. }
  511. // 移除末尾的 "&"
  512. urlWithParams = urlWithParams[:len(urlWithParams)-1]
  513. // 发送 POST 请求
  514. resp, err := http.Post(urlWithParams, "application/x-www-form-urlencoded", nil)
  515. if err != nil {
  516. return "", err
  517. }
  518. defer resp.Body.Close()
  519. // 读取响应内容
  520. body, err := ioutil.ReadAll(resp.Body)
  521. if err != nil {
  522. return "", err
  523. }
  524. // 解析 JSON 响应
  525. var tokenVo map[string]interface{}
  526. err = json.Unmarshal(body, &tokenVo)
  527. if err != nil {
  528. return "", err
  529. }
  530. // 获取 access_token
  531. token, ok := tokenVo["access_token"].(string)
  532. if !ok {
  533. return "", fmt.Errorf("access_token not found in response")
  534. }
  535. return token, nil
  536. }
  537. func GetFeedDataFromApi() {
  538. // mapResult := make(map[string]interface{})
  539. // 获取 API Token
  540. tokenString, err := getApiToken()
  541. if err != nil {
  542. return
  543. }
  544. if tokenString == "" {
  545. return
  546. }
  547. // 构建 URL
  548. urlString := fmt.Sprintf("https://zhmc-api.yunyangniu.com/duijie/v1/breed/synMaterialInfo?farmNum=%s&secret=%s", setting.YynserverSetting.FarmId, setting.YynserverSetting.YynSecret)
  549. // 构建请求头
  550. headerMap := map[string]string{
  551. "username": "KPT_USER",
  552. "Authorization": "Bearer " + tokenString,
  553. }
  554. // 发送 POST 请求
  555. reDataString, err := sendPost(urlString, headerMap, nil)
  556. if err != nil {
  557. return
  558. }
  559. if reDataString != "" {
  560. var resultMap map[string]interface{}
  561. err = json.Unmarshal([]byte(reDataString), &resultMap)
  562. if err != nil {
  563. return
  564. }
  565. msgCode, ok := resultMap["msgCode"].(string)
  566. if !ok || msgCode != "10000" {
  567. return
  568. }
  569. contentMap, ok := resultMap["content"].(map[string]interface{})
  570. if !ok {
  571. return
  572. }
  573. b, _ := json.Marshal(contentMap["materialInfoList"])
  574. feedList := make([]*models.YYNMaterialInfo, 0)
  575. json.Unmarshal(b, &feedList)
  576. tx := restful.Engine.NewSession()
  577. defer tx.Close()
  578. pastureinfo := new(udPastureInfo)
  579. err := tx.SQL(`select column_default as pastureid,(select werks from pasture where pastureid = column_default) werks from information_schema.COLUMNS
  580. WHERE table_name = 'recweight' AND table_schema = ? AND column_name = 'pastureid'`, setting.DatabaseSetting.Name).GetFirst(pastureinfo).Error
  581. if err != nil {
  582. logs.Error(err)
  583. return
  584. }
  585. for _, feed := range feedList {
  586. ids, err := setting.SnowIds.NextId()
  587. if err != nil {
  588. ids = time.Now().UnixNano()
  589. logging.Info("create SnowIds err", err)
  590. }
  591. _, err = tx.Exec(` insert into feedclass(id,pastureid,fcname,fccode,bigfeedclassid,bigfeedclassname)values(?,?,?,?,?,?) ON DUPLICATE KEY UPDATE backup1 = '云养牛'`,
  592. ids, pastureinfo.Pastureid, feed.CategoryName, feed.CategoryCode, ids, feed.CategoryName)
  593. if err != nil {
  594. logs.Error(err)
  595. return
  596. }
  597. ids1, err := setting.SnowIds.NextId()
  598. if err != nil {
  599. ids1 = time.Now().UnixNano()
  600. logging.Info("create SnowIds err", err)
  601. }
  602. _, err = 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. fmt.Println(err)
  606. }
  607. }
  608. }
  609. func getBarDataFromApi() {
  610. // 获取 API Token
  611. tokenString, err := getApiToken()
  612. if err != nil {
  613. return
  614. }
  615. if tokenString == "" {
  616. return
  617. }
  618. // 构建 URL
  619. urlString := fmt.Sprintf("https://zhmc-api.yunyangniu.com/duijie/v1/breed/synCowHouseInfo?farmNum=%s", setting.YynserverSetting.FarmId)
  620. urlString += fmt.Sprintf("&secret=%s", setting.YynserverSetting.YynSecret)
  621. // 构建请求头
  622. headerMap := map[string]string{
  623. "username": "KPT_USER",
  624. "Authorization": "Bearer " + tokenString,
  625. }
  626. // 发送 POST 请求
  627. reDataString, err := sendPost(urlString, headerMap, nil)
  628. if err != nil {
  629. return
  630. }
  631. // fmt.Println(time.Now(), "牛舍接口返回:", reDataString)
  632. if reDataString != "" {
  633. var resultMap map[string]interface{}
  634. err = json.Unmarshal([]byte(reDataString), &resultMap)
  635. if err != nil {
  636. return
  637. }
  638. msgCode, ok := resultMap["msgCode"].(string)
  639. if !ok || msgCode != "10000" {
  640. return
  641. }
  642. contentMap, ok := resultMap["content"].(map[string]interface{})
  643. if !ok {
  644. return
  645. }
  646. cowHouseList, ok := contentMap["cowHouseList"].([]interface{})
  647. if !ok {
  648. return
  649. }
  650. fmt.Println(time.Now(), "牛舍列表返回:", cowHouseList)
  651. // cowAmount:57 farmName:吉林省牧硕养殖有限公司 farmNum:1830002720 farmUuid:737a6a60094011e9802143b3b3c94e01 houseName:1-3东蹄病
  652. // houseNum:131 houseType:lactation houseTypeName:泌乳牛舍 houseUuid:05f3e0a3127f11efbf42d1859aea01b1
  653. // _, err := tx.SQL(` insert into bar(pastureid,bcode,bname,sort,class,classcode,cattle,cattlecode,sapcode)
  654. // values(?,?,?,(select max(sort)+1 from bar b where b.pastureid = ? ),
  655. // (select distName from dist where distCode = ? and distType = '牛舍类型' ),?,
  656. // (select distName from dist where distCode = ? and distType = '牛群类别'),?,?)
  657. // ON DUPLICATE KEY UPDATE bname = ? ,
  658. // class = (select distName from dist where distCode = ? and distType = '牛舍类型' ) ,classcode = ?,
  659. // cattle = (select distName from dist where distCode = ? and distType = '牛群类别' ) ,cattlecode = ?
  660. // `,
  661. // pastureId, barMap["CHSNO"], fmt.Sprintf("%v_sap", barMap["CHSTX"]), pastureId, barMap["CHSTY"], barMap["CHSTY"],
  662. // barMap["FCWTS"], barMap["FCWTS"], barMap["CHSNO"],
  663. // fmt.Sprintf("%v_sap", barMap["CHSTX"]), barMap["CHSTY"], barMap["CHSTY"], barMap["FCWTS"], barMap["FCWTS"]).Execute()
  664. // if err != nil {
  665. // logs.Error(err)
  666. // return
  667. //}
  668. }
  669. }
  670. func sendPost(url string, headers map[string]string, body []byte) (string, error) {
  671. req, err := http.NewRequest("POST", url, bytes.NewBuffer(body))
  672. if err != nil {
  673. return "", err
  674. }
  675. for key, value := range headers {
  676. req.Header.Set(key, value)
  677. }
  678. client := &http.Client{}
  679. resp, err := client.Do(req)
  680. if err != nil {
  681. return "", err
  682. }
  683. defer resp.Body.Close()
  684. respBody, err := ioutil.ReadAll(resp.Body)
  685. if err != nil {
  686. return "", err
  687. }
  688. return string(respBody), nil
  689. }