milkcocoaにて[Errno 32] Broken pipeエラーが発生する。

raspberry pi3にて、
温度、湿度、気圧を計測し、
ｍilkcocoaにpushするプログラムを作成し、
crontabにて5ヶ月ほど運用してきましたが、
ここ数日、安定して計測できなくなり、
直接コマンドを入力した所、
数回に一度、[Errno 32] Broken pipeと表示され、
milkcocoaにpushできていません。
原因はなんでしょうか。

どうやら、以下のプログラムでエラーになっているようです。
def on_push(e):
print(e)

エラー場所1
push_start
on_push_end
temp_end
on_push_end
humid_end
[Errno 32] Broken pipe
on_push_end
pressure_end

エラー場所2
push_start
on_push_end
temp_end
[Errno 32] Broken pipe
on_push_end
humid_end
on_push_end
pressure_end

python
1#coding: utf-8
2
3import sys
4import smbus
5import time
6import datetime
7import milkcocoa.milkcocoa as milkcocoa
8
9milkcocoaClient = milkcocoa.Milkcocoa.connectWithApiKey("****", "****", "***", useSSL=False,blocking=True)
10
11temp = milkcocoaClient.datastore("temp")
12humid = milkcocoaClient.datastore("humid")
13pressure = milkcocoaClient.datastore("pressure")
14
15bus_number  = 1
16i2c_address = 0x76
17
18bus = smbus.SMBus(bus_number)
19
20digT = []
21digP = []
22digH = []
23
24t_fine = 0.0
25
26sensor_data = {'temp':'0.0', 'pressure':'0.0','humidity':'0.0'}
27
28def on_push(e):
29    print(e)
30
31def writeReg(reg_address, data):
32	bus.write_byte_data(i2c_address,reg_address,data)
33
34def get_calib_param():
35	calib = []
36	
37	for i in range (0x88,0x88+24):
38		calib.append(bus.read_byte_data(i2c_address,i))
39	calib.append(bus.read_byte_data(i2c_address,0xA1))
40	for i in range (0xE1,0xE1+7):
41		calib.append(bus.read_byte_data(i2c_address,i))
42
43	digT.append((calib[1] << 8) | calib[0])
44	digT.append((calib[3] << 8) | calib[2])
45	digT.append((calib[5] << 8) | calib[4])
46	digP.append((calib[7] << 8) | calib[6])
47	digP.append((calib[9] << 8) | calib[8])
48	digP.append((calib[11]<< 8) | calib[10])
49	digP.append((calib[13]<< 8) | calib[12])
50	digP.append((calib[15]<< 8) | calib[14])
51	digP.append((calib[17]<< 8) | calib[16])
52	digP.append((calib[19]<< 8) | calib[18])
53	digP.append((calib[21]<< 8) | calib[20])
54	digP.append((calib[23]<< 8) | calib[22])
55	digH.append( calib[24] )
56	digH.append((calib[26]<< 8) | calib[25])
57	digH.append( calib[27] )
58	digH.append((calib[28]<< 4) | (0x0F & calib[29]))
59	digH.append((calib[30]<< 4) | ((calib[29] >> 4) & 0x0F))
60	digH.append( calib[31] )
61	
62	for i in range(1,2):
63		if digT[i] & 0x8000:
64			digT[i] = (-digT[i] ^ 0xFFFF) + 1
65
66	for i in range(1,8):
67		if digP[i] & 0x8000:
68			digP[i] = (-digP[i] ^ 0xFFFF) + 1
69
70	for i in range(0,6):
71		if digH[i] & 0x8000:
72			digH[i] = (-digH[i] ^ 0xFFFF) + 1  
73
74def readData():
75	data = []
76	for i in range (0xF7, 0xF7+8):
77		data.append(bus.read_byte_data(i2c_address,i))
78	pres_raw = (data[0] << 12) | (data[1] << 4) | (data[2] >> 4)
79	temp_raw = (data[3] << 12) | (data[4] << 4) | (data[5] >> 4)
80	hum_raw  = (data[6] << 8)  |  data[7]
81	
82	compensate_T(temp_raw)
83	compensate_P(pres_raw)
84	compensate_H(hum_raw)
85
86def compensate_P(adc_P):
87	global  t_fine
88	pressure = 0.0
89	
90	v1 = (t_fine / 2.0) - 64000.0
91	v2 = (((v1 / 4.0) * (v1 / 4.0)) / 2048) * digP[5]
92	v2 = v2 + ((v1 * digP[4]) * 2.0)
93	v2 = (v2 / 4.0) + (digP[3] * 65536.0)
94	v1 = (((digP[2] * (((v1 / 4.0) * (v1 / 4.0)) / 8192)) / 8)  + ((digP[1] * v1) / 2.0)) / 262144
95	v1 = ((32768 + v1) * digP[0]) / 32768
96	
97	if v1 == 0:
98		return 0
99	pressure = ((1048576 - adc_P) - (v2 / 4096)) * 3125
100	if pressure < 0x80000000:
101		pressure = (pressure * 2.0) / v1
102	else:
103		pressure = (pressure / v1) * 2
104	v1 = (digP[8] * (((pressure / 8.0) * (pressure / 8.0)) / 8192.0)) / 4096
105	v2 = ((pressure / 4.0) * digP[7]) / 8192.0
106	pressure = pressure + ((v1 + v2 + digP[6]) / 16.0)  
107
108	# print "pressure : %7.2f hPa" % (pressure/100)
109        sensor_data['pressure'] = pressure/100
110
111def compensate_T(adc_T):
112	global t_fine
113	v1 = (adc_T / 16384.0 - digT[0] / 1024.0) * digT[1]
114	v2 = (adc_T / 131072.0 - digT[0] / 8192.0) * (adc_T / 131072.0 - digT[0] / 8192.0) * digT[2]
115	t_fine = v1 + v2
116	temperature = t_fine / 5120.0
117	# print "temp : %-6.2f 邃・ % (temperature)
118        sensor_data['temp'] = temperature
119
120def compensate_H(adc_H):
121	global t_fine
122	var_h = t_fine - 76800.0
123	if var_h != 0:
124		var_h = (adc_H - (digH[3] * 64.0 + digH[4]/16384.0 * var_h)) * (digH[1] / 65536.0 * (1.0 + digH[5] / 67108864.0 * var_h * (1.0 + digH[2] / 67108864.0 * var_h)))
125	else:
126		return 0
127	var_h = var_h * (1.0 - digH[0] * var_h / 524288.0)
128	if var_h > 100.0:
129		var_h = 100.0
130	elif var_h < 0.0:
131		var_h = 0.0
132	# print "hum : %6.2f ・・ % (var_h)
133        sensor_data['humidity'] = var_h
134
135def setup():
136	osrs_t = 1			#Temperature oversampling x 1
137	osrs_p = 1			#Pressure oversampling x 1
138	osrs_h = 1			#Humidity oversampling x 1
139	mode   = 3			#Normal mode
140	t_sb   = 5			#Tstandby 1000ms
141	filter = 0			#Filter off
142	spi3w_en = 0			#3-wire SPI Disable
143
144	ctrl_meas_reg = (osrs_t << 5) | (osrs_p << 2) | mode
145	config_reg    = (t_sb << 5) | (filter << 2) | spi3w_en
146	ctrl_hum_reg  = osrs_h
147
148	writeReg(0xF2,ctrl_hum_reg)
149	writeReg(0xF4,ctrl_meas_reg)
150	writeReg(0xF5,config_reg)
151
152
153setup()
154get_calib_param()
155
156if __name__ == '__main__':
157	try:
158		readData()
159                time_str = datetime.datetime.today().strftime("%Y/%m/%d %H:%M:%S")
160                temp_str = str(sensor_data['temp'])
161                humid_str = str(sensor_data['humidity'])
162                pressure_str = str(sensor_data['pressure'])
163                #milkcocoa縺ｫ繝・・繧ｿ繧恥ush縺吶ｋ縲・
164                print("push_start");
165                temp.on("push",on_push)
166                print("on_push_end");
167                temp.push({"temp":temp_str})
168                print("temp_end");
169                time.sleep(3)
170                humid.on("push",on_push)
171                print("on_push_end");
172                humid.push({"humid":humid_str})
173                print("humid_end")
174                time.sleep(3)
175                pressure.on("push",on_push)
176                print("on_push_end");
177                pressure.push({"pressure":pressure_str})
178                print("pressure_end")
179		#print(sensor_data)
180                print(time_str+","+temp_str+","+humid_str+","+pressure_str)
181	except KeyboardInterrupt:
182		pass
183

行動規範の内容に同意します

回答1件

ベストアンサー

一般的にsocket通信でBad Pipeが発生するのは、接続が切れているのに送受信を実行した場合（相手先の受信終了を待たずに送信終了した場合など）のようです。
How to prevent errno 32 broken pipe?

以下の公式のサンプルのように、通信待ち(loop_forever())が必要なのではないでしょうか？
blocking.py

詳細はmilkcocoaが利用しているMQTTライブラリの説明「Network loop」や以下を参照ください。
paho-mqttで回線を切るとどうなるか

投稿2017/01/23 09:08

編集2017/01/23 09:12

8524ba23

総合スコア38343

usuallyi

2017/01/25 11:08

loop_forever()にした場合、milkcocoaから返答がないため、待機してしまい。プログラムが終わらない状況になります。また、新たなアプリリストを作成した際には、問題なくデータをpushできているので、困っています。

8524ba23

2017/01/26 06:22

ざっとライブラリを見る限り、1秒でタイムアウトし制御が戻ってくるはずですが、おかしいですね。ちなみにpush毎にSleepを入れていますが、なぜですか？

usuallyi

2017/01/26 06:35

連続で3回pushした場合、うまくpushできない時があったため、 sleepを入れて送っています。

8524ba23

2017/01/26 06:42

sleepにて「うまくpushできない」の詳細を質問に追記ください。どうも通信の同期がとれていないのが原因のような気がします。

usuallyi

2017/01/28 15:14

以下のメールが先日来ましたどうやら、障害だったようです。おかげさまで順調にアプリ数が増えておりまして、クライアントの制限を超えたアクセスなどにより、負荷が上昇してしまったことで、12,1月と夜間Milkcocoaのサーバにアクセスしづらくなっておりました。現在負荷調整のための改修作業を断続的に行っております。 2017年1月26日 17:25現在負荷が安定していることを確認していますが、今後も継続的に改修を行っていきます。継続的に皆様にも状況をご連絡いたします。ご迷惑をおかけして、大変申し訳ございません。

8524ba23

2017/01/29 00:56

原因について了解しました。障害が発生した場合のエラー処理も加えておいたほうがよいのかもしれませんね。

行動規範の内容に同意します