"""Core firmware: the ``fw`` class, ``device`` helpers and free functions.
Everything below is **verbatim** from the device firmware
(``firmware_core.py`` of Logger Pro and the findyIoT module). ``fw`` deliberately has no
``start()``/``measure()`` - those are plain functions in the
user-editable ``last.py``, attached with ``fw.start = start`` and
``fw.measure = measure``.
Note:
Methods reference the firmware module globals (``modem``,
``protocol``, ``db``, ``counter``, ``buffer``, ``wdt``, ``led``...)
exactly as on the device.
"""
import time
import utime
import gc
import machine
from findylab.modem import P_TxGSM, P_RXGSM, P_DTR
from findylab.relay import relay5V, relay12V, PperipherialPower, Pbuzzer
from findylab.ota import OTA_APPLIED
#: I2C pins of the external ADC (address 104).
Psda = 18
Pscl = 19
#: Reed-switch input (EXT0 deep-sleep wakeup).
Preed = 15
#: Direct digital inputs (counted by the ULP as RTC GPIO 10/16).
Pinput4 = 4
Pinput14 = 14
[docs]
class fw(object):
"""Main firmware class. Verbatim from ``firmware_core.py``.
``__init__()`` does the first-boot bootstrap (IMEI + database
defaults). ``start()`` and ``measure()`` are **not** defined here -
they live in the user-editable ``last.py`` and are attached with
``fw.start = start`` / ``fw.measure = measure``.
"""
[docs]
def __init__(self):
"""First-boot bootstrap.
Reads the IMEI parameter (200); when empty, powers the modem
(power-cycling it once via :meth:`findylab.modem.eg915.reset`
if it stays silent), reads the real IMEI, initializes the
database defaults, re-opens the database and sends the
``defaults`` report. Reassigns the module-level ``db`` in that
path.
"""
print("Core v. 1.2")
#if machine.wake_reason() != 0 :
# machine.reset()
global db
global modem
global protocol
global Gimei
machine.freq(80000000)
Gimei = db.read('200')
# b'0' is the params default for '200': read() now returns it (encoded)
# for a missing key instead of False, so first boot must match both.
# b'' covers a previous first boot that stored an empty IMEI.
if Gimei == False or Gimei == b'0' or Gimei == b'':
modem.turnOn()
if modem.isOn() == False:
# PWRKEY attempts didn't bring it up - power-cycle the rail
# and try once more (a fresh rail + PWRKEY is what recovers
# a modem that ignored the first cold-boot pulse).
modem.reset()
imei = modem.getImei()
protocol.imei = imei
modem.registerNetwork()
db.initDefaults()
db.store()
gc.collect()
print(gc.mem_free())
db = database(database.params)
Gimei = db.read('200').decode('utf-8')
protocol.main('defaults')
else:
Gimei = Gimei.decode('utf-8')
# NOTE: `start()` is not defined here - see last.py.
# boot.py does `from last import fw; f = fw(); f.start()`, and last.py
# attaches the method with `fw.start = start` after importing this module.
[docs]
def seconds_to_next_wakeup(self,period_seconds):
"""Computes the seconds until the next wakeup time.
Aligns wakeups to round boundaries of the RTC clock so reports
arrive at predictable times.
Args:
period_seconds (int): The period; one of 60, 300, 900,
1800, 3600 (anything else falls back to 900).
Returns:
int: Seconds until the next boundary.
"""
if period_seconds not in [60, 300, 900, 1800, 3600]:
period_seconds = 900
# Get the current time
rtc = machine.RTC()
current_time = rtc.datetime() # (year, month, day, weekday, hour, minute, second, subseconds)
print("Current time ", current_time)
current_hour, current_minute, current_second = current_time[4], current_time[5], current_time[6]
# Compute the current time in seconds since the start of the day
current_total_seconds = current_hour * 3600 + current_minute * 60 + current_second
# Compute the next valid moment in seconds
next_wakeup_total_seconds = ((current_total_seconds // period_seconds) + 1) * period_seconds
# Compute the seconds until the next wakeup
seconds_to_next = next_wakeup_total_seconds - current_total_seconds
print("Calculated seconds to next wake up", seconds_to_next)
return seconds_to_next
[docs]
def report(self,alarm=False,datalist=None):
"""Sends a report to the HiGPS server.
Powers/registers the modem, refreshes the GSM info parameters,
then sends ``samplingsAlarm`` or ``samplings``; on a build
error falls back to the minimal ``dataBat`` report. After a
confirmed connection the alarm flag, ring-buffers and wakeup
counter are reset, the database is stored and a just-committed
OTA update is reported back to the server (53 = 0, 54 = 1)
via ``_notifyUpdateResult()``.
Args:
alarm (bool): Send the alarm variant of the report.
datalist (list): Parameter IDs to send (defined by the user
in ``last.py``). None keeps the previously set list
(``protocol.datalist``).
"""
#counter.getSamplings()
global connectionSuccess
if datalist is not None:
protocol.datalist = datalist
try:
if modem.isOn() == False:
turnOncounter =0
while modem.isOn() == False and turnOncounter < 30:
modemOn = modem.turnOn()
time.sleep(0.5)
#time.sleep(2)
if modem.registerNetwork():
if machine.wake_reason() == 2 :
beep()
time.sleep(0.2)
beep()
else :
return
buffer.get('540')
buffer.get('541')
print("Reporting")
self.getGSMInfo()
if alarm:
protocol.main('samplingsAlarm')
else :
protocol.main('samplings')
except Exception as e:
setError(" Error in report "+(str(e)))
protocol.main('dataBat')
print("Connection success", connectionSuccess)
if connectionSuccess:
db.write('440',0)
buffer.remove('540')
buffer.remove('541')
buffer.create('540')
buffer.create('541')
db.write('522','0')
db.store()
self._notifyUpdateResult()
def _notifyUpdateResult(self):
"""If boot.py committed a staged OTA and we are now running the
freshly swapped firmware, tell the server the update succeeded
(53=idle, 54=ok) on the first connection that works, then clear the
marker. base.py handles the opposite case (fell back to recovery ->
update failed). The marker is removed only after the push succeeds,
so a drop here just retries on the next report."""
import os
try:
os.stat(OTA_APPLIED)
except OSError:
return
print("OTA: post-update report - success")
db.write('53', '0')
db.write('54', '1') # updated successfully
db.store()
try:
protocol.main("setting,53", False)
protocol.main("setting,54", False)
except Exception as e:
print("OTA: post-update report failed", e)
return
try:
os.remove(OTA_APPLIED)
except OSError:
pass
[docs]
def getGSMInfo(self):
"""Copies the serving-cell info into parameters 230-238."""
print("get GSM Info ToDo")
data = modem.getNetworkInfo()
if data:
print(data)
db.write('230',data['NetworkBearer']) #Network Bearer
db.write('231',data['SignalRSSI']) #Radio Signal Strength
db.write('232',data['LinkQuality']) #Link Quality
db.write('234',data['CellID']) #Cell ID
db.write('235',data['SMNC']) #SMNC
db.write('236',data['SMCC']) #SMCC
db.write('237',data['SINR']) #SignalSNR
db.write('238',data['LAC']) #LAC
[docs]
def prepareSleep(self,t,trig):
"""Releases the pins and enters deep sleep.
Turns the modem off, reconfigures all peripheral pins as inputs
and arms the reed-switch EXT0 wakeup when ``trig`` is set, then
calls :meth:`goodNight`.
Args:
t (int): Sleep time in seconds.
trig (bool): Arm the reed-switch wakeup.
Note:
eg915 has no PSM support, so this is a plain
``machine.deepsleep()`` - no network PSM negotiation.
"""
print("GOING TO SLEEP FOR ",t)
t = t + 1
#GSM pwr
#PWR Key
modem.turnOff()
#machine.Pin(PmodemPWRKey, machine.Pin.IN,machine.Pin.PULL_UP)
machine.Pin(P_RXGSM, machine.Pin.IN)
machine.Pin(P_TxGSM, machine.Pin.IN)
machine.Pin(P_DTR, machine.Pin.IN)
machine.Pin(Pscl, machine.Pin.IN)
machine.Pin(Psda, machine.Pin.IN)
machine.Pin(PperipherialPower, machine.Pin.IN)
#machine.Pin(P_Power_GSM, machine.Pin.IN)
machine.Pin(Pbuzzer, machine.Pin.IN)
#machine.Pin(Pinput14, machine.Pin.IN)
#machine.Pin(Pinput4, machine.Pin.IN)
#machine.Pin(Preed, machine.Pin.IN)
machine.Pin(relay5V, machine.Pin.IN)
machine.Pin(relay12V, machine.Pin.IN)
#machine.Pin(PmodemPWRKey, machine.Pin.IN,machine.Pin.PULL_DOWN)
#machine.Pin(PmodemPWRKey, machine.Pin.IN,machine.Pin.PULL_DOWN)
import esp32
esp32.gpio_deep_sleep_hold(True)
led.value(0)
if trig:
print("Triggering alarm while sleep")
print("sleeping with motion trigger")
mot = machine.Pin(Preed,machine.Pin.IN,machine.Pin.PULL_DOWN)
if mot.value() == 1:
mot.irq(trigger=machine.Pin.IRQ_FALLING , wake=machine.DEEPSLEEP)
esp32.wake_on_ext0(pin = mot, level = esp32.WAKEUP_ALL_LOW)
else:
mot.irq(trigger=machine.Pin.IRQ_RISING | machine.Pin.IRQ_FALLING , wake=machine.DEEPSLEEP)
esp32.wake_on_ext0(pin = mot, level = esp32.WAKEUP_ANY_HIGH)
st = self.goodNight(t*1000)
[docs]
def goodNight(self,timeS):
"""Stores the database and enters deep sleep.
Args:
timeS (int): Sleep time in milliseconds.
Note:
Prints the exception and resets when ``db.store()`` fails.
"""
try:
db.store()
print("Going to Deep Sleep")
machine.deepsleep(timeS)
except Exception as e:
# Print before resetting - a silent bare except here made
# db.store() failures impossible to diagnose from the log.
print("goodNight error:", e)
machine.reset()
# NOTE: `measure()` is not defined here - like `start()`, it's defined in
# last.py and attached with `fw.measure = measure`, so each application
# can decide what (and whether) to measure. The hardware-level helper it
# builds on, adcMeasure(), stays here.
[docs]
def adcMeasure(self,n):
"""Performs one raw conversion on the external I2C ADC.
Channel selectors: ``b'\x90'`` AIN1 (4-20 mA), ``b'\xb0'``
AIN2 (4-20 mA), ``b'\xd0'`` external power, ``b'\xf0'`` the
0-5 V input.
Args:
n (bytes): Channel selector byte.
Returns:
int: Raw 16-bit conversion result.
"""
i2c = machine.SoftI2C(freq=100000,scl=machine.Pin(Pscl),sda=machine.Pin(Psda))
i2c.start()
#print(i2c.scan())
i2c.writeto(104,b'\x18')
time.sleep(0.01)
i2c.writeto(104,n)
time.sleep(0.01)
sampling =int.from_bytes(i2c.readfrom(104,4)[0:2],"big")
return sampling
[docs]
class device(object):
"""Device-level helpers. Never instantiated - the methods take no
``self`` and are called directly::
device.getTimestamp()
Verbatim class from ``firmware_core.py``.
"""
[docs]
def restart():
"""Resets the device (``machine.reset()``)."""
machine.reset()
[docs]
def factoryReset():
"""Wipes the database to defaults and resets the device."""
db.initDefaults()
#db.store()
machine.reset()
[docs]
def freeMemory():
"""Returns the free heap in KB.
Returns:
int: Free RAM in kilobytes.
"""
import gc
return round(gc.mem_free()/1024)
[docs]
def serialNumber():
"""Returns the unique chip ID.
Returns:
str: Hex-encoded ``machine.unique_id()``.
"""
import ubinascii
return ubinascii.hexlify(machine.unique_id()).decode()
[docs]
def memoryTotal():
"""Returns the allocated heap in KB.
Returns:
int: Allocated RAM in kilobytes.
"""
import gc
return round(gc.mem_alloc()/1024)
[docs]
def getTimestamp():
#from machine import RTC
#clock = RTC()
#timeArray = clock.datetime()
"""Returns the current Unix epoch (+946684800 legacy offset).
When the RTC looks invalid (before 2022) the time is fetched
from the server via ``modem.getTimeF()`` and the RTC is set.
Returns:
int: Epoch seconds.
"""
timestamp = time.time()+946684800
if timestamp < 1664000000:
#if timeArray[0] == 2000 :
print("Wrong date")
if modem.isOn() == False:
modem.turnOn()
modem.registerNetwork()
#modem.turnOn()
higpsTimestamp = modem.getTimeF()
if len(higpsTimestamp) == 10:
ttt=time.localtime(int(higpsTimestamp)-946684800)
machine.RTC().init((ttt[0],ttt[1],ttt[2],ttt[3],ttt[4],ttt[5],0,0))
print(machine.RTC().datetime())
timestamp = time.time()+946684800
else:
print("Cannot get time")
else :
print("Date OK")
return timestamp
[docs]
def setTimestamp(timestamp = False):
"""Sets the RTC from a given epoch or from the server.
Args:
timestamp (str): 10-digit epoch string; when False the time
is fetched from the server.
"""
if timestamp:
higpsTimestamp = timestamp
else:
modem.turnOn()
higpsTimestamp = modem.getTimeF()
if len(higpsTimestamp) == 10:
ttt=time.localtime(int(higpsTimestamp)-946684800)
machine.RTC().init((ttt[0],ttt[1],ttt[2],ttt[3],ttt[4],ttt[5],0,0))
print(machine.RTC().datetime())
timestamp = time.time()+946684800
[docs]
def wdtAlert(p):
"""Watchdog callback: stores the database, powers the modem off
and resets the device.
Args:
p: Timer instance (unused).
"""
print("WDT Reset")
try:
gc.collect()
db.store()
modem.turnOff()
except:
print("Error in WDT")
#machine.Pin(PmodemPWRKey,machine.Pin.OUT).value(0)
machine.reset()
[docs]
def beep():
"""Audible signal: 5 kHz on the buzzer (GPIO27) for 0.5 s."""
pwm0 = machine.PWM(machine.Pin(27))
pwm0.duty(500)
pwm0.freq(5000)
time.sleep(0.5)
pwm0.duty(0)
[docs]
def setError(error):
"""Accumulates an error message for the next report.
Appends to the global ``errors`` (sent as the ``&er=`` request
field) and clears ``errorsSent``.
Args:
error (str): Human-readable error text.
"""
global errors
global errorsSent
errors += " "+ error
errorsSent = False
print(error)