# This program is free software: you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see . """ ########################## # Weather and GPS logger # ########################## Author : Pierrick Couturier Use with: * Adafruit Feather M0 Express (CircuitPython firmware 3.0.0) * Adafruit Ultimate GPS FeatherWing * Bosch BME280 sensor (air temperature, humidity, atmospheric pressure) on I2C https://framagit.org/arofarn/Cameteo TODO for v1 : * write data on flash drive (work-in-progress) * send data through UART (work-in-progress) """ __version__ = 0.2 ########## # config # ########## print_data = True send_json_data = True backup_data = True data_to_neopixel = True gps_enable = True update_interval = const(10) # in seconds send_json_data = True datetime_format = "{:04}/{:02}/{:02}_{:02}:{:02}:{:02}" neopixel_max_value =const(70) #max value instead of brightness to spare some mem ####################### import microcontroller import gc, os # import micropython import time, rtc from busio import I2C, UART from analogio import AnalogIn from digitalio import DigitalInOut, Direction from adafruit_bme280 import Adafruit_BME280_I2C from adafruit_gps import GPS import neopixel ########### # Classes # ########### class Data: """Class for handling data""" def __init__(self): self.data = {'SYS': {'time': "2000/01/01_00:00:00", 'vbat': int(), 'cput': float()}, 'BME': {'temp': float(), 'hum': int(), 'press': float()}, 'GPS': {'time': "2000/01/01_00:00:00", 'lat': float(), 'lon': float(), 'alt': float(), 'qual': int()} } def update(self): """Read the data from various sensors and update the data dict variable""" #Data from Feather board self.data['SYS']['time'] = datetime_format.format(*clock.datetime[0:6]) self.data['SYS']['vbat'] = round(measure_vbat(), 3) self.data['SYS']['cput'] = round(microcontroller.cpu.temperature, 2) #Data from BME280 self.data['BME']['temp'] = round(bme280.temperature, 1) self.data['BME']['hum'] = int(bme280.humidity) self.data['BME']['press'] = round(bme280.pressure, 2) if gps_enable: if gps.has_fix: self.data['GPS']['time'] = datetime_format.format(gps.timestamp_utc.tm_year, gps.timestamp_utc.tm_mon, gps.timestamp_utc.tm_mday, gps.timestamp_utc.tm_hour, gps.timestamp_utc.tm_min, gps.timestamp_utc.tm_sec) self.data['GPS']['lat'] = gps.latitude self.data['GPS']['lon'] = gps.longitude self.data['GPS']['alt'] = gps.altitude_m self.data['GPS']['qual'] = gps.fix_quality else: self.data['GPS']['lat'] = None self.data['GPS']['lon'] = None self.data['GPS']['alt'] = None else: self.data['GPS'] = None def show(self): """Serialize data for visualization on serial console""" for source in self.data.keys(): print(source + ": ") if not self.data[source] == None: for d in self.data[source].items(): print("\t{0}: {1}".format(d[0], d[1])) def json(self): """Serialize data to json-formatted string""" output = '{' first_source = True for source in self.data.keys(): if first_source: first_source = False comma_src = "" else: comma_src = "," output = '{}{}"{}":'.format(output, comma_src, source) if not self.data[source] == None: output = output + '{' first_data = True for d in self.data[source].items(): if first_data: comma = "" first_data = False else: comma = "," output = '{}{}"{}":"{}"'.format(output, comma, d[0], d[1]) output = output + '}' output = output + '}' return output def atmo2rgb(self): """Convert atmospheric data from BME280 sensor into NeoPixel color * RED => temperature : max = 35degC, min =10degC (range 25°C) * BLUE => humidity : max= 100%, mini=0% * GREEN => pression : mini=960hPa, maxi = 1030hPa (range 70hPa) """ red = int((self.data['BME']['temp']-10)*neopixel_max_value/25) if red > neopixel_max_value: red = neopixel_max_value if red < 0: red = 0 blue = int(self.data['BME']['hum']*neopixel_max_value/100) green = int((self.data['BME']['press']-960)*neopixel_max_value/70) if green > neopixel_max_value: green = neopixel_max_value if green < 0: green = 0 if print_data: print("Col:{}".format((red, green, blue))) return (red, green, blue) def write_on_flash(self): """Save the current data as csv file on SPI flash""" try: with open("data/data.csv", "a") as csv_file: csv_file.write("{};{};{};{}\n".format(self.data['SYS']['time'], self.data['BME']['temp'], self.data['BME']['hum'], self.data['BME']['press'])) except OSError as e: print("Err {}: readonly".format(e)) backup_data = False #to avoid trying again till next reset led13.value = True ############# # Functions # ############# def check_data_dir(): """Check if data directories exists""" if 'data' not in os.listdir(): os.mkdir('data') os.mkdir('data/hourly') os.mkdir('data/daily') elif 'hourly' not in os.listdir('data'): os.mkdir('data/hourly') elif 'daily' not in os.listdir('data'): os.mkdir('data/daily') def set_clock_from_GPS(): if gps_enable and gps.has_fix: #Congreen GPS timestamp into struct_time gps_datetime = time.struct_time((gps.timestamp_utc.tm_year, gps.timestamp_utc.tm_mon, gps.timestamp_utc.tm_mday, gps.timestamp_utc.tm_hour, gps.timestamp_utc.tm_min, gps.timestamp_utc.tm_sec, 0, 0, 0)) #Max difference between GPS and internal RTC (in seconds): if abs(time.mktime(gps_datetime) - time.mktime(clock.datetime)) >= 5.0: print("Clock difference with GPS!") print("Previous date/time : " + datetime_format.format(*clock.datetime[0:6])) clock.datetime = gps_datetime #Trust GPS if there is a bias print("New date/time : " + datetime_format.format(*clock.datetime[0:6])) def measure_vbat(samples=10, timestep=0.01): """Measure Vbattery as the mean of n samples with timestep second between each measurement""" # Note about v_bat calculation : # 0.000100708 = 2*3.3/65536 with # 2 : voltage is divided by 2 # 3.3 : Vref = 3.3V # 65536 : 16bit ADC v = 0 for i in range(samples): v = v + vbat.value time.sleep(timestep) return v/samples*0.000100708 ######### # Setup # ######### gc.collect() #micropython.mem_info() #Enable RTC of the feather M0 board clock = rtc.RTC() #clock.datetime = time.struct_time((2018, 7, 29, 15, 31, 30, 0, 0, 0)) # BME280 sensors (I2C) i2c = I2C(microcontroller.pin.PA23, microcontroller.pin.PA22) # i2c addresses for BME280 breakout : # 0x77 = adafruit breakout board # 0x76 = tiny chinese board bme280 = Adafruit_BME280_I2C(i2c, address=0x76) # Battery voltage vbat = AnalogIn(microcontroller.pin.PA07) gps_en_pin = DigitalInOut(microcontroller.pin.PB02) gps_en_pin.direction = Direction.OUTPUT led13 = DigitalInOut(microcontroller.pin.PA17) led13.direction = Direction.OUTPUT led13.value = False # GPS on FeatherWing board gps_en_pin.value = not gps_enable if gps_enable: gps_uart = UART(microcontroller.pin.PA10, microcontroller.pin.PA11, baudrate=9600, timeout=3000) gps = GPS(gps_uart) # Turn on the basic GGA and RMC info gps.send_command('PMTK314,0,1,0,1,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0') gps.send_command('PMTK220,1000') # 1000 ms refresh rate # second UART to communicate with raspberry pi GPIO if send_json_data: rpi_uart = UART(microcontroller.pin.PB10, microcontroller.pin.PB11, baudrate=115200, timeout=2000) # Integrated Neopixel if data_to_neopixel: pixel = neopixel.NeoPixel(microcontroller.pin.PA06, 1, brightness=1) else: #if neopixel is disable : turn off the LED pixel = neopixel.NeoPixel(microcontroller.pin.PA06, 1, brightness=1) pixel[0] = (0,0,0) pixel = None check_data_dir() ############# # Main loop # ############# data = Data() last_update = time.monotonic() while True: if gps_enable: gps.update() current = time.monotonic() if current - last_update >= update_interval: last_update = current set_clock_from_GPS() data.update() if print_data: data.show() if send_json_data: rpi_uart.write(data.json()) if backup_data: data.write_on_flash() if data_to_neopixel: pixel[0] = data.atmo2rgb() gc.collect() # micropython.mem_info(1) # print('Memory free: {} allocated: {}'.format(gc.mem_free(), gc.mem_alloc()))