#author("2024-04-18T16:43:00+09:00","default:pico_wiki","pico_wiki") #author("2024-04-18T16:43:46+09:00","default:pico_wiki","pico_wiki") [[sensors-graph-ex01.html]] #code(python){{ device_id="yama_pico_0000_0000_0000_0001" #init_url="http://192.168.11.10/pukiwiki/pico_wiki/" #init_url="http://192.168.1.16/pukiwiki/pico_wiki/" #init_url="http://192.168.13.30/pukiwiki/pico_wiki/" #init_url="http://www.yama-lab.org/fwb4pi/index.php" #init_url="http://192.168.8.136/pukiwiki/" init_url="http://192.168.8.237/pukiwiki/pico_wiki/" #init_page="20240131-01" #init_page="20240131-02" #init_page="20240131-03" #init_page="work20240201-01" #init_page="work20240326-01" #init_page="work20240330-01" init_page="bme280-01" #init_page="iot64" #init_page="20240221-03" #ssid = 'Buffalo-G-B5CA' #password = 'isxvatt6i4fii' #ssid = 'halappa-a' #password = '87654321' #ssid = '0024A5C04900_G' #password = '4v3u44th8xu6x' #ssid = 'Buffalo-G-7808' #password = 'wpmm6urkpt8py' #ssid = 'Buffalo-G-24C0' #password = 'nt8kcd5vfbrut' #ssid= '0024A5C04900_G' #password = '4v3u44th8xu6x' #ssid = 'Tosyo-FreeWiFi' #password = 'tosyo-385' ssid="GL-SFT1200-8f4" password="goodlife" import machine import time import network import urequests import socket try: import ustruct as struct except: import struct from time import sleep from machine import Timer import sys import gc # Authors: Paul Cunnane 2016, Peter Dahlebrg 2016 # # This module borrows from the Adafruit BME280 Python library. Original # Copyright notices are reproduced below. # # Those libraries were written for the Raspberry Pi. This modification is # intended for the MicroPython and esp8266 boards. # # Copyright (c) 2014 Adafruit Industries # Author: Tony DiCola # # Based on the BMP280 driver with BME280 changes provided by # David J Taylor, Edinburgh (www.satsignal.eu) # # Based on Adafruit_I2C.py created by Kevin Townsend. # # Permission is hereby granted, free of charge, to any person obtaining a copy # of this software and associated documentation files (the "Software"), to deal # in the Software without restriction, including without limitation the rights # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell # copies of the Software, and to permit persons to whom the Software is # furnished to do so, subject to the following conditions: # # The above copyright notice and this permission notice shall be included in # all copies or substantial portions of the Software. # # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN # THE SOFTWARE. import time from ustruct import unpack, unpack_from from array import array # BME280 default address. BME280_I2CADDR = 0x76 # Operating Modes BME280_OSAMPLE_1 = 1 BME280_OSAMPLE_2 = 2 BME280_OSAMPLE_4 = 3 BME280_OSAMPLE_8 = 4 BME280_OSAMPLE_16 = 5 BME280_REGISTER_CONTROL_HUM = 0xF2 BME280_REGISTER_CONTROL = 0xF4 class BME280: def __init__(self, mode=BME280_OSAMPLE_1, address=BME280_I2CADDR, i2c=None, **kwargs): # Check that mode is valid. if mode not in [BME280_OSAMPLE_1, BME280_OSAMPLE_2, BME280_OSAMPLE_4, BME280_OSAMPLE_8, BME280_OSAMPLE_16]: raise ValueError( 'Unexpected mode value {0}. Set mode to one of ' 'BME280_ULTRALOWPOWER, BME280_STANDARD, BME280_HIGHRES, or ' 'BME280_ULTRAHIGHRES'.format(mode)) self._mode = mode self.address = address if i2c is None: raise ValueError('An I2C object is required.') self.i2c = i2c # load calibration data dig_88_a1 = self.i2c.readfrom_mem(self.address, 0x88, 26) dig_e1_e7 = self.i2c.readfrom_mem(self.address, 0xE1, 7) self.dig_T1, self.dig_T2, self.dig_T3, self.dig_P1, \ self.dig_P2, self.dig_P3, self.dig_P4, self.dig_P5, \ self.dig_P6, self.dig_P7, self.dig_P8, self.dig_P9, \ _, self.dig_H1 = unpack("<HhhHhhhhhhhhBB", dig_88_a1) self.dig_H2, self.dig_H3 = unpack("<hB", dig_e1_e7) e4_sign = unpack_from("<b", dig_e1_e7, 3)[0] self.dig_H4 = (e4_sign << 4) | (dig_e1_e7[4] & 0xF) e6_sign = unpack_from("<b", dig_e1_e7, 5)[0] self.dig_H5 = (e6_sign << 4) | (dig_e1_e7[4] >> 4) self.dig_H6 = unpack_from("<b", dig_e1_e7, 6)[0] self.i2c.writeto_mem(self.address, BME280_REGISTER_CONTROL, bytearray([0x3F])) self.t_fine = 0 # temporary data holders which stay allocated self._l1_barray = bytearray(1) self._l8_barray = bytearray(8) self._l3_resultarray = array("i", [0, 0, 0]) def read_raw_data(self, result): """ Reads the raw (uncompensated) data from the sensor. Args: result: array of length 3 or alike where the result will be stored, in temperature, pressure, humidity order Returns: None """ self._l1_barray[0] = self._mode self.i2c.writeto_mem(self.address, BME280_REGISTER_CONTROL_HUM, self._l1_barray) self._l1_barray[0] = self._mode << 5 | self._mode << 2 | 1 self.i2c.writeto_mem(self.address, BME280_REGISTER_CONTROL, self._l1_barray) sleep_time = 1250 + 2300 * (1 << self._mode) sleep_time = sleep_time + 2300 * (1 << self._mode) + 575 sleep_time = sleep_time + 2300 * (1 << self._mode) + 575 time.sleep_us(sleep_time) # Wait the required time # burst readout from 0xF7 to 0xFE, recommended by datasheet self.i2c.readfrom_mem_into(self.address, 0xF7, self._l8_barray) readout = self._l8_barray # pressure(0xF7): ((msb << 16) | (lsb << 8) | xlsb) >> 4 raw_press = ((readout[0] << 16) | (readout[1] << 8) | readout[2]) >> 4 # temperature(0xFA): ((msb << 16) | (lsb << 8) | xlsb) >> 4 raw_temp = ((readout[3] << 16) | (readout[4] << 8) | readout[5]) >> 4 # humidity(0xFD): (msb << 8) | lsb raw_hum = (readout[6] << 8) | readout[7] result[0] = raw_temp result[1] = raw_press result[2] = raw_hum def read_compensated_data(self, result=None): """ Reads the data from the sensor and returns the compensated data. Args: result: array of length 3 or alike where the result will be stored, in temperature, pressure, humidity order. You may use this to read out the sensor without allocating heap memory Returns: array with temperature, pressure, humidity. Will be the one from the result parameter if not None """ self.read_raw_data(self._l3_resultarray) raw_temp, raw_press, raw_hum = self._l3_resultarray # temperature var1 = ((raw_temp >> 3) - (self.dig_T1 << 1)) * (self.dig_T2 >> 11) var2 = (((((raw_temp >> 4) - self.dig_T1) * ((raw_temp >> 4) - self.dig_T1)) >> 12) * self.dig_T3) >> 14 self.t_fine = var1 + var2 temp = (self.t_fine * 5 + 128) >> 8 # pressure var1 = self.t_fine - 128000 var2 = var1 * var1 * self.dig_P6 var2 = var2 + ((var1 * self.dig_P5) << 17) var2 = var2 + (self.dig_P4 << 35) var1 = (((var1 * var1 * self.dig_P3) >> 8) + ((var1 * self.dig_P2) << 12)) var1 = (((1 << 47) + var1) * self.dig_P1) >> 33 if var1 == 0: pressure = 0 else: p = 1048576 - raw_press p = (((p << 31) - var2) * 3125) // var1 var1 = (self.dig_P9 * (p >> 13) * (p >> 13)) >> 25 var2 = (self.dig_P8 * p) >> 19 pressure = ((p + var1 + var2) >> 8) + (self.dig_P7 << 4) # humidity h = self.t_fine - 76800 h = (((((raw_hum << 14) - (self.dig_H4 << 20) - (self.dig_H5 * h)) + 16384) >> 15) * (((((((h * self.dig_H6) >> 10) * (((h * self.dig_H3) >> 11) + 32768)) >> 10) + 2097152) * self.dig_H2 + 8192) >> 14)) h = h - (((((h >> 15) * (h >> 15)) >> 7) * self.dig_H1) >> 4) h = 0 if h < 0 else h h = 419430400 if h > 419430400 else h humidity = h >> 12 if result: result[0] = temp result[1] = pressure result[2] = humidity return result return array("i", (temp, pressure, humidity)) @property def values(self): """ human readable values """ t, p, h = self.read_compensated_data() p = p // 256 pi = p // 100 pd = p - pi * 100 hi = h // 1024 hd = h * 100 // 1024 - hi * 100 return ("{}C".format(t / 100), "{}.{:02d}hPa".format(pi, pd), "{}.{:02d}%".format(hi, hd)) #import copy class pico_net: def __init__(self): print('current ssid= '+ssid) def set_access_point(self, ssid, password): self.ssid = ssid self.password = password def connect(self): self.wlan = network.WLAN(network.STA_IF) self.wlan.active(True) chipid = int.from_bytes(self.wlan.config('mac'), 'big') % 65536 self.wlan.connect(ssid, password) max_wait = 10 while max_wait > 0: if self.wlan.status() < 0 or self.wlan.status() >= 3: break max_wait -= 1 print('waiting for connection...') time.sleep(1) if self.wlan.status() != 3: raise RuntimeError('network connection failed') else: print('connected') self.status = self.wlan.ifconfig() print('ip = ' + self.status[0]) class pico_http: def __init__(self): print('pico_http') def set_url(self, url): self.url = url def get(self, url): #print('pico_http.get(urL='+url+')') self.url=url #print('urL='+url) try: r = urequests.get(url) #print(r) r_text=r.text r.close() #print('pico_http.get...return') return r_text except: print('pico_http.get...exception') return "" def put(self, url, payload): #print('pico_http.put(urL='+url+'payload='+str(payload)+')') #payload is a dictionary uri=url+'?' for key in payload: #print(key+'=',end="") #print(payload[key]) uri=uri+key+'='+payload[key]+'&' uri=uri[:-1] #print('uri='+uri) #r = urequests.post(uri) #r = urequests.post(url,data=payload) r = urequests.get(uri) if r=="": print('pico_http.put...exception') return "" #print(r) r_text=r.text r.close() #print(r_json) return r_text #post, ref: https://docs.micropython.org/en/latest/library/urequests.html def post(self, url, body, headers): #print('pico_http.post(url='+url+' headers='+str(headers)+')') try: r = urequests.post(url,data=body,headers=headers) #print(r) r_text=r.text #print(r_text) #r_json=r.json() r.close() #print(r_json) print('pico_http.post...return') return r_text except: print('pico_http.post...exception') return "" class pico_time: # reference: https://wisteriahill.sakura.ne.jp/CMS/WordPress/2023/05/10/raspberry-pi-pico-w-ntp-server/ # NTPサーバー #ntp_host = "time-c.nist.gov" #ntp_host = "time.cloudflare.com" #ntp_host = "time.google.com" ntp_host = "ntp.nict.jp" def __init__(self): print('pico_time') def ptime(self): NTP_DELTA = 2208988800 NTP_QUERY = bytearray(48) NTP_QUERY[0] = 0x1b addr = socket.getaddrinfo(self.ntp_host, 123)[0][-1] s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) # s.settimeout(1) res = s.sendto(NTP_QUERY, addr) msg = s.recv(48) s.close() val = struct.unpack("!I", msg[40:44])[0] return val - NTP_DELTA wday = ['Mon','Tue','Wed','Thu','Fri','Sat','Sun'] def zfill(self, s, width): if len(s) < width: return ("0" * (width - len(s))) + s else: return s def get_now(self): t=self.ptime() t = t + 9*60*60 #print(t) datetimetuple = time.gmtime(t) year = str(datetimetuple[0]) month = self.zfill(str(datetimetuple[1]),2) mday = self.zfill(str(datetimetuple[2]),2) hour = self.zfill(str(datetimetuple[3]),2) minute = self.zfill(str(datetimetuple[4]),2) seconds = self.zfill(str(datetimetuple[5]),2) wd_index = datetimetuple[6] #日付と時刻を分けるデリミター("|")を付加 #datetime = year + "/" + month + "/" + mday + " " + self.wday[wd_index] + "|" + hour + ":" + minute + ":" + seconds datetime = year + "/" + month + "/" + mday + " " + hour + ":" + minute + ":" + seconds return datetime class html_parser: html="" tokens=[] reserved_words={'<':'<','>':'>','&':'&','"':'"','\'':''', ' ':' ','\n':'<br>','\t':' ','\r':'', '\\':'\','{':'{','}':'}','[':'[',']':']' } a2c={' ':"+",'\n':"%0D%0A",'\r':"%0D",'\t':"%09",'!':"%21",'"':"%22", '#':"%23",'$':"%24",'%':"%25",'&':"%26", '\'':"%27",'(': "%28",')':"%29", '*':"%2A",'+':"%2B",',':"%2C",'-':"%2D",'/':"%2F",':':"%3A", ';':"%3B",'<':"%3C",'=':"%3D",'>':"%3E",'?':"%3F", '@':"%40",'[':"%5B",'\\':"%5C",']':"%5D",'^':"%5E", '`':"%60",'{':"%7B",'|':"%7C",'}':"%7D",'~':"%7E"} #Spechal char to ascii code sc2a={'<':"<",'>':">",'&':"&",'"':'"',''':"\'"} def __init__(self,html): self.html=html def html_tokenizer(self): #print("html_tokenizer") self.token_ith=0 w="" tx=[""] rest=[""] inx=self.html xlen=len(self.html) while xlen>0: if self.parse_tag(inx,tx,rest): if w!="": self.tokens.append(w) w="" self.tokens.append('<'+tx[0]+'>') #print('<'+tx[0]+'>') inx=rest[0] xlen=len(inx) elif self.parse_String_Constant(inx,tx,rest): if w!="": self.tokens.append(w) w="" self.tokens.append('"'+tx[0]+'"') #print('"'+tx[0]+'"') inx=rest[0] xlen=len(inx) else: wc=inx[0] inx=inx[1:xlen] xlen=len(inx) w=w+wc if w!="": self.tokens.append(w) w="" def get_html(self): return self.html def parse_key(self,key,inx,rest): #print("parse_key-"+key) keylen=len(key) inlen=len(inx) if inx.startswith(key): rest[0]=inx[keylen:inlen] return True rest[0]=inx return False def parse_String_Constant(self,inx,strc,rest): rx=[""] xstr="" if self.parse_key('"',inx,rx): #print("parse_String_Constant") inx=rx[0] fx=inx[0] xlen=len(inx) while fx!='"': if xlen==0: return False if fx=='\\': inx=inx[1:xlen] else: xstr=xstr+fx xlen=len(inx) inx=inx[1:xlen] fx=inx[0] if self.parse_key('"',inx,rx): strc[0]=xstr rest[0]=rx[0] return True return False def parse_tag(self,inx,tag,rest): rx=[""] xstr="" wstr=[""] if self.parse_key('<',inx,rx): #print("parse_tag") inx=rx[0] fx=inx[0] xlen=len(inx) while fx!='>': if xlen<=0: return False if fx=='\\': inx=inx[1:xlen] if self.parse_String_Constant(inx,wstr,rx): inx=rx[0] xstr=xstr+'"'+wstr[0]+'"' else: xstr=xstr+fx inx=inx[1:xlen] fx=inx[0] xlen=len(inx) if self.parse_key('>',inx,rx): tag[0]=xstr rest[0]=rx[0] return True return False def get_tokens(self): return self.tokens def next_token(self): if self.token_ith<len(self.tokens): self.token_ith=self.token_ith+1 return self.tokens[self.token_ith-1] return "" def has_more_tokens(self): if self.token_ith<len(self.tokens): return True return False def index_from(self,i,str): p=self.html[i:].index(str) return p def p_url(self,p,url): #print("p_url p="+str(p)+' '+self.html[p:p+10]) rx=[""] xurl="" pw=p xlen=len(self.html) q=self.p_key('http://',p) if p!=q: xurl='http://' else: q=self.p_key('https://',p) if p!=q: xurl='https://' else: return pw p=q while not (self.html[p] in [' ','\n','\t','\r']): p=p+1 if p>=xlen: p=xlen break url[0]=xurl+self.html[q:p] return p def p_key(self,key,p): #print("p_key key="+key+" p="+str(p)+" "+self.html[p:p+10]) keylen=len(key) if p+keylen>=len(self.html): return p q=p if self.html[q:q+keylen]==key: #print("key="+key+" have found, q="+str(q+keylen)) return q+keylen #print("key="+key+" is not found, p="+str(p)) return p def p_String_Constant(self,p,strc): #print("p_String_Constant p="+str(p)+' '+self.html[p:p+10]) rx=[""] xstr="" pw=p q=self.p_key('"',p) if p!=q: #print("p_String_Constant p="+str(p)+' '+self.html[p:p+10]) p=q fx=self.html[p] xlen=len(self.html)-p while fx!='"': if xlen==0: return pw if fx in ['\n','\t','\r']: return pw if fx=='\\': p=p+1 p=p+1 fx=self.html[p] xlen=len(self.html)-p q=self.p_key('"',p) if p!=q: strc[0]=self.html[pw+1:q-1] #print("p_string_constant, strc="+str(strc[0])) return q #print("p_String_Constant, fail to find \"") return pw def p_Date(self,p,date_str): #print("p_Date p="+str(p)+' '+self.html[p:p+10]) rx=[""] xstr="" pw=p year=[0] q=self.p_number(p,year) if p==q: return pw p=q q=self.p_key('/',p) if p==q: return pw p=q month=[0] q=self.p_number(p,month) if p==q: return pw p=q q=self.p_key('/',p) if p==q: return pw p=q day=[0] q=self.p_number(p,day) if p==q: return False p=q q=self.p_b(p) if p==q: return pw p=q hour=[0] q=self.p_number(p,hour) if p==q: return pw p=q q=self.p_key(':',p) if p==q: return pw p=q minute=[0] q=self.p_number(p,minute) if p==q: return pw p=q q=self.p_key(':',p) if p==q: return pw p=q second=[0] q=self.p_number(p,second) if p==q: return pw p=q date_str[0]=str(year[0])+'/'+str(month[0])+'/'+str(day[0])+' '+str(hour[0])+':'+str(minute[0])+':'+str(second[0]) return p def p_b(self,p): #print("parse_b") #print("p="+str(p)) xlen=len(self.html) if p>=xlen: return p while self.html[p]==' ': p=p+1 if p>=xlen: p=xlen break return p def p_name(self,p,name): #print("p_name, self.html="+ self.html[p:p+10]) #print("p="+str(p)+"..."+self.html[p:p+10]) q=p last_p=len(self.html) while self.html[q] in 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_': #print("self.html["+str(q)+"]="+self.html[q]) q=q+1 if q>=last_p: break name[0]=self.html[p:q] #print("p_name p="+str(p)+" q="+str(q)+" name="+str(name[0])) #print("p="+str(p)+" q="+str(q)+" name="+name[0]) #print("name[0]= "+name[0]) return q def p_number(self,p,num): #print("p_number") #print("p="+str(p)) n="" last_p=len(self.html) q=p while self.html[p] in '0123456789': q=q+1 if q>=last_p: break num[0]=int(self.html[p:q]) #print("p_number p="+str(p)+" q="+str(q)+" num="+str(num[0])) return q def p_an_equation(self,p,eqs): #print("p_an_equation p="+str(p)+' '+self.html[p:p+10]) pw=p p=self.p_b(p) nx=[""] strc=[""] q=self.p_name(p,nx) #print("p_an_equation name="+nx[0]) if p==q: #print("p_an_equation fail to find name") return pw p=q p=self.p_b(p) q=self.p_key('=',p) if p==q: #print("p_an_equation fail to find =") return pw p=q p=self.p_b(p) q=self.p_String_Constant(p,strc) if p==q: num=[0] q=self.p_number(p,num) if p==q: return pw else: eqs[nx[0]]=num[0] #print("p_qn_equation get "+nx[0]+"="+str(num[0])) return q else: eqs[nx[0]]=strc[0] #print("p_an_equation get "+nx[0]+"="+str(strc[0])) return q def p_equations(self,p,eqs): #print("parse_equations") #print("p="+str(p)) pw=p while True: q=self.p_an_equation(p,eqs) if p==q: return p p=q def p_get_equations_of_the_tag(self,p,tag_name,eqs): #print("p_get_equations_of_the_tag <"+tag_name) #print("p="+str(p)+' '+self.html[p:p+10]) pw=p try: q=self.html.index('<'+tag_name,pw) except: #print("p_get_equation_of_the_tag fail to find <"+tag_name) return p #print("q="+str(q)+' '+self.html[q:q+10]) q=q+len('<'+tag_name) #print("q="+str(q)+' '+self.html[q:q+10]) if pw==q: #print("p_get_equation_of_the_tag fail to find <"+tag_name) return p pw=q q=self.p_equations(pw,eqs) #print("after equations q="+str(q) +' '+self.html[q:q+10]) if pw==q: return p pw=q pw=self.p_b(pw) #print("pw="+str(pw)) q=self.p_key('>',pw) #print("after > q="+str(q)+' '+self.html[q:q+10]) if pw!=q: return q q=self.p_key('/>',pw) if pw!=q: return q return p def p_get_between_tag_with_cond(self,p,tag_name,cond,rtn): #print("p_get_between_with_cond <"+tag_name+">...</"+tag_name+">") #print("p="+str(p)+' '+self.html[p:p+10]) pw=p q=pw while True: try: q=self.html.index('<'+tag_name,p) #print("p_get_between_with_cond p="+str(p)+" q="+str(q)+' '+self.html[q:q+20]) if p==q: return pw q=q+len('<'+tag_name) p=q eqs={} q=self.p_equations(p,eqs) #print("after equations q="+str(q) +' '+self.html[q:q+10]) if p!=q: if cond[0] in eqs: #print("cond[0]="+cond[0]+" eqs[cond[0]]="+eqs[cond[0]]+" cond[1]="+cond[1]) if eqs[cond[0]]!=cond[1]: continue else: break except: #print("fail to find <"+tag_name) return pw p=self.p_b(p) #print("p_get_between q="+str(q)+' '+self.html[q:q+10]) #print("pw="+str(pw)) q=self.p_key('>',p) if p!=q: start_p=q else: q=self.p_key('/>',p) if pw!=q: start_p=q else: return p try: q=self.html.index('</'+tag_name,start_p) except: return pw end_p=q-1 p=q q=q+len('</'+tag_name) p=self.p_b(q) #print("pw="+str(pw)) q=self.p_key('>',p) if q==p: return pw rtn[0]=(start_p,end_p) #rx=self.html[start_p:end_p] #print("rx="+rx) return q def p_get_between(self,p,tag_name,eqs,rtn): #print("p_get_between <"+tag_name+">...</"+tag_name+">") #print("p="+str(p)+' '+self.html[p:p+10]) pw=p q=pw try: q=self.html.index('<'+tag_name,p) except: return pw #print("p_get_between 2 q="+str(q)+' '+self.html[q:q+10]) q=q+len('<'+tag_name) #print("p_get_between 3 q="+str(q)+' '+self.html[q:q+10]) if p==q: return pw p=q q=self.p_equations(p,eqs) #print("after equations q="+str(q) +' '+self.html[q:q+10]) p=q p=self.p_b(p) #print("p="+str(p)) q=self.p_key('>',p) #print("after > q="+str(q)+' p='+str(p)+' '+self.html[p:q+10]) if p!=q: start_p=q else: q=self.p_key('/>',p) if p!=q: end_p=q rtn[0]="" return q else: #print("get between "+tag_name+",fail to find <"+tag_name+"/>") return pw try: q=self.html.index('</'+tag_name,start_p) except: #print("get between "+tag_name+",fail to find </"+tag_name) return pw end_p=q-1 p=q q=q+len('</'+tag_name) p=self.p_b(q) #print("p="+str(p)) q=self.p_key('>',p) if q==p: return pw rtn[0]=(start_p,end_p) #rx=self.html[start_p:end_p+1] #print("get between "+tag_name+" rx="+rx) return q def code_convert(self,code): if code in self.a2c: return self.a2c[code] else: return code def text_to_post_form(self,text): #print("html_parser.text_to_post_form") #print(text) rtn="" for i in range(len(text)): rtn+=self.code_convert(text[i]) return rtn def special_char_to_text(self,text): #print("special_char_to_text") #print(text) text=text.replace("<","<") text=text.replace(">",">") text=text.replace("&","&") text=text.replace(""",'"') text=text.replace("'","'") return text def post(self,url,body): #payload is a dictionary headers={'Content-Type': 'application/x-www-form-urlencoded'} #def http_post(url, value): # https://docs.openmv.io/library/urequests.html #body = "value=%s" % (value) #headers = {'Content-Type': 'application/x-www-form-urlencoded'} response = urequests.post(url, data=body, headers=headers) r_text=response.text #print(r_text) response.close() #print(r_json) return r_text class pico_wiki_driver: url="" def __init__(self,url,page): print('pico_wiki_driver') self.url=url self.page=page self.Pico_Net=pico_net() self.Pico_Time=pico_time() self.Pico_Http=pico_http() self.Pico_Net.connect() t=self.Pico_Time.get_now() print('time now='+t) def set_url(self,url): self.url=url def set_page(self,page): self.page=page def get_url(self): return self.url def get_html(self): uri = self.url+"?"+self.page all_page=self.Pico_Http.get(uri) #print("get_html..."+uri) #print(all_page) return all_page def get_wiki_page(self): uri = self.url+"?"+self.page all_page=self.Pico_Http.get(uri) print("get_wiki_page...") #print(all_page) parser=html_parser(all_page) p=0 rtn=[(0,0)] q=parser.p_get_between_tag_with_cond(p,"div",('id',"body"),rtn) if p!=q: #print("get_wiki_page id=body..."+all_page[rtn[0][0]:rtn[0][1]]) body_x=all_page[rtn[0][0]:rtn[0][1]] body=parser.special_char_to_text(body_x) return body def replace_wiki_page(self,new_body): print("pico_wiki_driver.replace_wiki_page...") payload={'cmd':'edit', 'page':self.page} #r=Pico_Http.get(url+'index.php?work202401') #print('uri='+r) r = self.Pico_Http.put(self.url,payload) #print(r) edit_html=html_parser(r) #edit_html.html_tokenizer() p=0 array_of_equations=[] equations={} q=edit_html.p_get_equations_of_the_tag(p,'input',equations) while p!=q: #print('p='+str(p)+' q='+str(q)+' <input....>') array_of_equations.append(equations) p=q equations={} q=edit_html.p_get_equations_of_the_tag(p,'input',equations) for i in range(len(array_of_equations)): #print('i='+str(i)) eqs=array_of_equations[i] for key in eqs: #print(key+'='+array_of_equations[i][key]) if key=='name' and array_of_equations[i][key]=="encode_hint": payload['encode_hint']=array_of_equations[i]['value'] elif key=='name' and array_of_equations[i][key]=="template": payload['template']=array_of_equations[i]['value'] elif key=='name' and array_of_equations[i][key]=="cmd": payload['cmd']=array_of_equations[i]['value'] elif key=='name' and array_of_equations[i][key]=="page": payload['page']=array_of_equations[i]['value'] elif key=='name' and array_of_equations[i][key]=="digest": payload['digest']=array_of_equations[i]['value'] elif key=='name' and array_of_equations[i][key]=="write": payload['write']=array_of_equations[i]['value'] #tokens=edit_html.get_tokens() #for i in range(len(tokens)):_ # print(str(i)+'-'+tokens[i]) eqs_msg={} rtn=[(0,0)] p=edit_html.p_get_between(0,'textarea',eqs_msg,rtn) if rtn[0]!=(0,0): start_p=rtn[0][0] end_p=rtn[0][1] body=r[0:start_p]+new_body+r[end_p+1:] print("newbody=......") print(new_body) payload['msg']=edit_html.text_to_post_form(new_body) print('msg=......') print(payload['msg']) gc.collect() eqs_original={} rtn=[(0,0)] p=edit_html.p_get_between(p,'textarea',eqs_original,rtn) if rtn[0]!=(0,0): start_p=rtn[0][0] end_p=rtn[0][1] original_pre=r[start_p:end_p+1] original=edit_html.special_char_to_text(original_pre) payload['original']=edit_html.text_to_post_form(original) body="" body=body+'encode_hint='+payload['encode_hint']+'&' body=body+'template_page=&' body=body+'cmd='+payload['cmd']+'&' body=body+'page='+payload['page']+'&' body=body+'digest='+payload['digest']+'&' body=body+'msg='+payload['msg']+'&' body=body+'write='+payload['write']+'&' body=body+'original='+payload['original'] #print('body='+body) try: edit_html.post(self.url, body) except: print("something wrong in edit_html.post") #print(r) def get_wiki_source(self): payload={'cmd':'edit', 'page':self.page} #r=Pico_Http.get(url+'index.php?work202401') #print('uri='+r) r = self.Pico_Http.put(self.url,payload) #print(r) edit_html=html_parser(r) #edit_html.html_tokenizer() p=0 array_of_equations=[] equations={} q=edit_html.p_get_equations_of_the_tag(p,'input',equations) while p!=q: #print('p='+str(p)+' q='+str(q)+' <input....>') array_of_equations.append(equations) p=q equations={} q=edit_html.p_get_equations_of_the_tag(p,'input',equations) for i in range(len(array_of_equations)): #print('i='+str(i)) eqs=array_of_equations[i] for key in eqs: #print(key+'='+array_of_equations[i][key]) if key=='name' and array_of_equations[i][key]=="encode_hint": payload['encode_hint']=array_of_equations[i]['value'] elif key=='name' and array_of_equations[i][key]=="template": payload['template']=array_of_equations[i]['value'] elif key=='name' and array_of_equations[i][key]=="cmd": payload['cmd']=array_of_equations[i]['value'] elif key=='name' and array_of_equations[i][key]=="page": payload['page']=array_of_equations[i]['value'] elif key=='name' and array_of_equations[i][key]=="digest": payload['digest']=array_of_equations[i]['value'] elif key=='name' and array_of_equations[i][key]=="cancel": payload['cancel']=array_of_equations[i]['value'] #tokens=edit_html.get_tokens() #for i in range(len(tokens)):_ # print(str(i)+'-'+tokens[i]) eqs_msg={} rtn=[(0,0)] p=edit_html.p_get_between(0,'textarea',eqs_msg,rtn) wiki_source="" if rtn[0]!=(0,0): start_p=rtn[0][0] end_p=rtn[0][1] wiki_source=r[start_p:end_p] #body="" #body=body+'encode_hint='+payload['encode_hint']+'&' #body=body+'cmd='+payload['cmd']+'&' #body=body+'page='+payload['page']+'&' #body=body+'cancel='+payload['cancel'] ##print('body='+body) #r=edit_html.post(self.url,body) return wiki_source class pico_wiki_bot: result_lines=[] script_lines=[] urls=[] programs={} environments={} time_millis=0 timer=None current_program_name="" current_program="" report=[] #reportLength=120 current_url="" current_page="" current_text="" next_url_index="" def __init__(self,url,page): print('pico_wiki_bot') self.Pico_Net=pico_net() self.Pico_Time=pico_time() self.Pico_Http=pico_http() self.Pico_Net.connect() t=self.Pico_Time.get_now() print('time now='+t) self.read_initials() self.current_url=url self.current_page=page self.next_url_index=0 self.Pico_Wiki_Driver=pico_wiki_driver(url,page) self.all_text="" if self.Pico_Wiki_Driver.get_url()=="": return "" self.timer=Timer() #all_text=self.Pico_Wiki_Driver.get_wiki_page() #if all_text==None: # return None #self.timer.init(mode=Timer.PERIODIC, freq=1, callback=self.add_one_second_to_time_millis) def read_initials(self): print("read_initials") try: f=open('initials.txt','r') print("initials.txt is found") lines=f.readlines() f.close() self.initials={} p=0 for i in range(len(lines)): print(str(i)+' - '+lines[i]) parser=html_parser(lines[i]) q=parser.p_an_equation(p,self.initials) if p!=q: p=q else: break print("-----initials-----") for key in self.initials: print(key+'='+self.initials[key]) except: print("initials.txt is not found") self.initials={} self.initials['url']=init_url self.initials['page']=init_page self.environments['read_interval']=20000 self.environments['exec_interval']=0 self.environments['write_interval']=0 self.environments['report_length']=120 self.url=self.initials['url'] self.page=self.initials['page'] def set_current_page(self,page): self.current_page=page def set_current_url(self,url): self.current_url=url def clear_report(self): self.report.clear() def add_report_line(self,send): #print("add_report_line, report_length="+str(self.environments['report_length'])) #print(" current report_length="+str(len(self.report))) #print(" adding..."+send) while len(self.report)>(int(self.environments['report_length']-1)): for i in range(len(self.report)-1): self.report[i]=self.report[i+1] self.report.pop() self.report.append(send) def add_one_second_to_time_millis(self,timer): #print("pico_wiki_bot.add_one_second_to_time_millis") self.time_millis=self.time_millis+1000 #print("pico_wiki_bot.add_one_second_to_time_millis return") def stop_timer(self): self.timer=None def get_script_and_result(self): #print("pico_wiki_bot.get_script_and_result") self.Pico_Wiki_Driver.set_url(self.current_url) self.Pico_Wiki_Driver.set_page(self.current_page) html=self.Pico_Wiki_Driver.get_html() if html=="": return "" parser=html_parser(html) p=0 eqs={} rtn=[(0,0)] q=parser.p_get_between(p,'pre',eqs,rtn) be=rtn[0] command_and_result=parser.html[be[0]:be[1]] self.current_text=parser.special_char_to_text(command_and_result) return self.current_text def get_result(self): #print("pico_wiki_bot.get_result") #r=self.get_script_and_result() all_text=self.current_text if all_text==None: return None if all_text=="": return "" #parser=html_parser(all_text) p=all_text.index("result:")+len("result:") q=all_text.index("current_device=",p) if q<0: q=len(all_text) else: q=q-1 print('pico_wiki_bot.get_result return') return all_text[p+1:q] def get_script(self): #print("pico_wiki_bot.get_script") r=self.get_script_and_result() all_text=self.current_text if all_text==None: return None if all_text=="": return "" #parser=html_parser(all_text) p=0 try: q=all_text.index("result:",p) q=q-1 except: q=len(all_text) return all_text[p:q] def get_current_device_line(self): #command_and_result=self.get_script_and_result() command_and_result=self.current_text parser=html_parser(command_and_result) if command_and_result==None: return None p=command_and_result.index("current_device=") p=p+len("current_device=") try: q=command_and_result.index("\n",p) except: q=len(command_and_result) return parser.special_char_to_text(command_and_result[p:q]) def replace_wiki(self,new_wiki): self.Pico_Wiki_Driver.replace_wiki_page(new_wiki) #print(r) def replace_result(self): #print("pico_wiki_bot.replace_result") payload={'cmd':'edit', 'page':self.page} #r=Pico_Http.get(url+'index.php?work202401') #print('uri='+r) r = self.Pico_Http.put(self.url,payload) #print(r) edit_html=html_parser(r) #edit_html.html_tokenizer() p=0 array_of_equations=[] equations={} q=edit_html.p_get_equations_of_the_tag(p,'input',equations) while p!=q: #print('p='+str(p)+' q='+str(q)+' <input....>') array_of_equations.append(equations) p=q equations={} q=edit_html.p_get_equations_of_the_tag(p,'input',equations) for i in range(len(array_of_equations)): #print('i='+str(i)) eqs=array_of_equations[i] for key in eqs: #print(key+'='+array_of_equations[i][key]) if key=='name' and array_of_equations[i][key]=="encode_hint": payload['encode_hint']=array_of_equations[i]['value'] elif key=='name' and array_of_equations[i][key]=="template": payload['template']=array_of_equations[i]['value'] elif key=='name' and array_of_equations[i][key]=="cmd": payload['cmd']=array_of_equations[i]['value'] elif key=='name' and array_of_equations[i][key]=="page": payload['page']=array_of_equations[i]['value'] elif key=='name' and array_of_equations[i][key]=="digest": payload['digest']=array_of_equations[i]['value'] elif key=='name' and array_of_equations[i][key]=="write": payload['write']=array_of_equations[i]['value'] eqs_msg={} rtn=[(0,0)] xbody="" start_p=0 end_p=0 p=edit_html.p_get_between(0,'textarea',eqs_msg,rtn) if rtn[0]!=(0,0): start_p=rtn[0][0] end_p=rtn[0][1] #print("start_p="+str(start_p)+" end_p="+str(end_p)) xbody=edit_html.special_char_to_text(r[start_p:end_p]) #body=r[0:start_p]+new_body+r[end_p+1:] #print("body=......") #print(body) eqs_original={} rtn=[(0,0)] p=edit_html.p_get_between(p,'textarea',eqs_original,rtn) start_px=end_p end_px=end_p if rtn[0]!=(0,0): start_px=rtn[0][0] end_px=rtn[0][1] original_pre=r[start_px:end_px] original=edit_html.special_char_to_text(original_pre) payload['original']=edit_html.text_to_post_form(original) #print("xbody="+xbody) px=0 #print("px="+str(px)+" "+xbody[px:px+10]) try: q=xbody.index("result:",px) #print("found result: q="+str(q)+" "+xbody[q:q+10]) except: q=start_p #print("could not found result: q="+str(q)+" "+xbody[q:q+10]) #print("q="+str(q)+" "+xbody[0:10]) if px!=q: q=q+len("result:") #print("q="+str(q)+" "+xbody[q:q+10]) gc.collect() head_area=xbody[0:q] #print("head_area="+head_area) #result_area=msg[px:q] new_area=head_area+"\n" for i in range(len(self.report)): new_area=new_area+' '+self.report[i]+'\n' new_area=new_area+" current_device=\""+device_id+"\". Date="+self.Pico_Time.get_now() #xmsg=head_area+"\n"+new_area #print("xmsg=......") #print(xmsg) gc.collect() payload['msg']=edit_html.text_to_post_form(new_area) #self.Pico_Wiki_Driver.replace_wiki_page(new_area) #print("payload['msg']="+payload['msg']) body="" body=body+'encode_hint='+payload['encode_hint']+'&' body=body+'template_page=&' body=body+'cmd='+payload['cmd']+'&' body=body+'page='+payload['page']+'&' body=body+'digest='+payload['digest']+'&' body=body+'msg='+payload['msg']+'&' body=body+'write='+payload['write']+'&' body=body+'original='+payload['original'] #print('body='+body) gc.collect() try: r=edit_html.post(self.url,body) except: print('something wrong in post') #print(r) def read_script(self): #print("read_script") all_text=self.get_script() if all_text==None or all_text=="": self.script_lines=[] return False else: self.script_lines=all_text.split('\n') #print("-----script_lines-----") #for i in range(len(self.script_lines)): # print(str(i)+' - '+self.script_lines[i]) return True def eval_script(self): print("eval_script") self.interpret_lines() #def print_result(self): # print("print_result") def is_object_page(self,parser): #print("is_object_page, line="+line[:10]+"...") p=0 p=parser.p_b(p) q=parser.p_key("object_page ",p) if p==q: #print("is_object_page return False") return False p=q p=parser.p_b(p) self.urls=[] xurl=[""] i=0 q=parser.p_url(p,xurl) #print(str(i)+"th url="+xurl[0]) while p!=q: i=i+1 self.urls.append(xurl[0]) p=q p=parser.p_b(p) q=parser.p_key("or ",p) #print("after key or p="+str(p)+" q="+str(q)) if p==q: break p=q p=parser.p_b(p) xurl=[""] q=parser.p_url(p,xurl) #print(str(i)+"th url="+xurl[0]) #print("p="+str(p)+" q="+str(q)) #for i in range(len(self.urls)): # print("url["+str(i)+"]="+self.urls[i]) return True def is_device(self,parser): p=0 p=parser.p_b(p) q=parser.p_key("current_device=",p) if p==q: return False p=q p=parser.p_b(p) xname=[""] q=parser.p_String_Constant(p,xname) if p==q: return False p=parser.p_b(q) self.current_device=xname[0] q=parser.p_key(",",p) if p==q: return False p=parser.p_b(q) q=parser.p_key("Date=",p) xdate=[] q=parser.p_Date(p,xdate) def is_set_command(self,parser,p): #print("is_set_command.."+parser.html[p:p+10]) #p=parser.p_b(p) q=parser.p_key("set ",p) if p==q: #print("is_set_command return False") return False p=parser.p_b(q) eq={} q=parser.p_an_equation(p,eq) if p==q: return False for key in eq: self.environments[key]=eq[key] #print("set "+str(key)+"="+str(eq[key])) return True def is_py_command(self,parser,p): #print("is_py_command.."+parser.html[p:p+10]) #p=parser.p_b(p) q=parser.p_key("py ",p) if p==q: return False p=parser.p_b(q) #print("p="+str(p)+",line[p:10]="+parser.html[p:p+10]) xname=[""] q=parser.p_name(p,xname) self.current_program_name=xname[0] self.current_program="" def is_end_command(self,parser,p): #print("is_end_command.."+parser.html[p:p+10]) #p=parser.p_b(p) q=parser.p_key("end ",p) if p==q: return False p=parser.p_b(q) xname=[""] q=parser.p_name(p,xname) self.programs[self.current_program_name]=self.current_program def is_run_command(self,parser,p): # reference : https://qiita.com/kyoshidajp/items/57ae371b3f5d8a84fb13 # reference : https://qiita.com/kammultica/items/3201f43eec53e3e56f54 #print("is_run_command.."+parser.html[p:p+10]) q=parser.p_key("run ",p) if p==q: #print("is_run_command return False") return False p=parser.p_b(q) xname=[""] q=parser.p_name(p,xname) if p==q: #print("is_run_command return False") return False r=self.get_result() if r!=None and r!="": self.clear_report() lines=r.split('\n') for i in range(len(lines)): self.add_report_line(lines[i]) exec_locals={'self':self} #print("line="+parser.html) #print("xname[0]="+xname[0]) program=self.programs[xname[0]] #print("program="+program) exec(program,globals(),exec_locals) #try: # exec(program) #except: # print("fail to exec "+program) # self.add_report_line("fail to exec:"+xname[0]) # sys.print_exception() def is_command(self,parser): p=0 p=parser.p_b(p) q=parser.p_key("command:",p) #print("line="+parser.html) #print("command="+parser.html[q:]) if p==q: return False q=parser.p_b(q) #lx=line[q:] if self.is_set_command(parser,q): return True if self.is_py_command(parser,q): return True if self.is_end_command(parser,q): return True if self.is_run_command(parser,q): return True return False def is_py(self,parser): #print("is_py line="+line[:10]) p=0 p=parser.p_b(p) q=parser.p_key("py: ",p) if p==q: return False lx=parser.html[q:] self.current_program=self.current_program+lx+'\n' def interpret_a_line(self,line): parser=html_parser(line) if self.is_object_page(parser): return if self.is_device(parser): return if self.is_command(parser): return if self.is_py(parser): return def interpret_lines(self): print("interpret_lines") for i in range(len(self.script_lines)): self.interpret_a_line(self.script_lines[i]) def next_url(self): print("next_url") if len(self.urls)==0: return if int(self.next_url_index)>=len(self.urls): self.next_url_index=0 urlx=self.urls[int(self.next_url_index)] p=urlx.index("?") self.current_page=urlx[p+1:] self.current_url=urlx[0:p-1] print("current_url="+self.current_url) print("current_page="+self.current_page) self.next_url_index=int(self.next_url_index)+1 self.Pico_Wiki_Driver.set_url(self.current_url) self.Pico_Wiki_Driver.set_page(self.current_page) def send_result(self): print("send_result") self.replace_result() def write_line(self,line): print("write_line") self.add_report_line(line) def read_eval_print_loop(self): print("read_eval_print_loop") last_read_time=time.ticks_ms() last_eval_time=last_read_time last_write_time=last_read_time while True: gc.collect() current_time=time.ticks_ms() if current_time<last_read_time: last_read_time=current_time if current_time-last_read_time>self.environments['read_interval']: last_read_time=current_time if self.read_script(): if self.environments['exec_interval']==0: self.eval_script() else: self.next_url() continue if self.environments['exec_interval']>0: if current_time-last_eval_time>self.environments['exec_interval']: last_eval_time=current_time self.eval_script() if self.environments['write_interval']>0: if current_time-last_write_time>self.environments['write_interval']: last_write_time=current_time self.send_result() time.sleep_ms(100) def main(): bot=pico_wiki_bot(init_url,init_page) #print("-----------get_script_and_result-----------------") #print(bot.get_script_and_result()) #print("-----------get_result-----------------") #print(bot.get_result()) #print("-----------get_script-----------------") #print(bot.get_script()) #print("-----------get_current_device_line-----------------") #print(bot.get_current_device_line()) #print("-----------replace_result-----------------") #gc.collect() #bot.clear_report() #bot.add_report_line("Hello!") #bot.add_report_line("This is a test2....") #bot.add_report_line("xxxx") #bot.replace_result() #bot.replace_result("Hello!\nThis is a test2....\nxxxx") #print("-----------stop_timer--------------------") #bot.stop_timer() bot.read_eval_print_loop() main() }}