feat: Add debug_force_send function for memory tracking and testing
This commit is contained in:
parent
5a8d14eb4d
commit
03b26b5339
@ -3,6 +3,71 @@ _CONFIG = {"discord_webhook_url": None, "discord_alert_webhook_url": None}
|
|||||||
# Cooldown after low-memory failures (epoch seconds)
|
# Cooldown after low-memory failures (epoch seconds)
|
||||||
_NEXT_ALLOWED_SEND_TS = 0
|
_NEXT_ALLOWED_SEND_TS = 0
|
||||||
|
|
||||||
|
def debug_force_send(message):
|
||||||
|
"""
|
||||||
|
Force one send attempt and print gc.mem_free() at key points.
|
||||||
|
Bypasses cooldown and pre-checks so you can measure peak allocations.
|
||||||
|
Use from REPL after WiFi connects:
|
||||||
|
import scripts.discord_webhook as d
|
||||||
|
d.set_config(config)
|
||||||
|
d.debug_force_send("memory test")
|
||||||
|
WARNING: this can trigger ENOMEM and crash if device free RAM is too low.
|
||||||
|
"""
|
||||||
|
global _NEXT_ALLOWED_SEND_TS
|
||||||
|
resp = None
|
||||||
|
try:
|
||||||
|
import gc # type: ignore
|
||||||
|
import time # type: ignore
|
||||||
|
|
||||||
|
url = _get_webhook_url(False)
|
||||||
|
if not url:
|
||||||
|
print("DBG_FORCE: no webhook URL configured")
|
||||||
|
return False
|
||||||
|
|
||||||
|
print("DBG_FORCE: mem before gc:", getattr(gc, "mem_free", lambda: 0)() // 1024, "KB")
|
||||||
|
gc.collect(); gc.collect()
|
||||||
|
print("DBG_FORCE: mem after gc:", getattr(gc, "mem_free", lambda: 0)() // 1024, "KB")
|
||||||
|
|
||||||
|
# Try importing urequests and show mem impact
|
||||||
|
try:
|
||||||
|
print("DBG_FORCE: importing urequests...")
|
||||||
|
import urequests as requests # type: ignore
|
||||||
|
gc.collect()
|
||||||
|
print("DBG_FORCE: mem after import:", getattr(gc, "mem_free", lambda: 0)() // 1024, "KB")
|
||||||
|
except Exception as e:
|
||||||
|
print("DBG_FORCE: urequests import failed:", e)
|
||||||
|
return False
|
||||||
|
|
||||||
|
# Build tiny payload
|
||||||
|
body_bytes = ('{"content":"%s","username":"%s"}' % (str(message)[:140], "DBG")).encode("utf-8")
|
||||||
|
print("DBG_FORCE: mem before post:", getattr(gc, "mem_free", lambda: 0)() // 1024, "KB")
|
||||||
|
|
||||||
|
try:
|
||||||
|
resp = requests.post(str(url).strip().strip('\'"'), data=body_bytes, headers={"Content-Type": "application/json"})
|
||||||
|
print("DBG_FORCE: mem after post:", getattr(gc, "mem_free", lambda: 0)() // 1024, "KB", "status:", getattr(resp, "status", None))
|
||||||
|
status = getattr(resp, "status", getattr(resp, "status_code", None))
|
||||||
|
return bool(status and 200 <= status < 300)
|
||||||
|
except Exception as e:
|
||||||
|
print("DBG_FORCE: exception during post:", e)
|
||||||
|
return False
|
||||||
|
|
||||||
|
finally:
|
||||||
|
try:
|
||||||
|
if resp:
|
||||||
|
resp.close()
|
||||||
|
except:
|
||||||
|
pass
|
||||||
|
try:
|
||||||
|
if 'requests' in globals():
|
||||||
|
del requests
|
||||||
|
except:
|
||||||
|
pass
|
||||||
|
try:
|
||||||
|
gc.collect()
|
||||||
|
print("DBG_FORCE: mem final:", getattr(gc, "mem_free", lambda: 0)() // 1024, "KB")
|
||||||
|
except:
|
||||||
|
pass
|
||||||
|
|
||||||
def set_config(cfg: dict):
|
def set_config(cfg: dict):
|
||||||
"""Initialize module with minimal values from loaded config (call from main)."""
|
"""Initialize module with minimal values from loaded config (call from main)."""
|
||||||
global _CONFIG
|
global _CONFIG
|
||||||
|
|||||||
16
Scripts/test_send.py
Normal file
16
Scripts/test_send.py
Normal file
@ -0,0 +1,16 @@
|
|||||||
|
import ujson
|
||||||
|
# Reload discord module fresh and run the forced debug send once.
|
||||||
|
try:
|
||||||
|
# ensure we use latest module on device
|
||||||
|
import sys
|
||||||
|
if "scripts.discord_webhook" in sys.modules:
|
||||||
|
del sys.modules["scripts.discord_webhook"]
|
||||||
|
import scripts.discord_webhook as d
|
||||||
|
# load config.json to populate webhook URL
|
||||||
|
with open("config.json", "r") as f:
|
||||||
|
cfg = ujson.load(f)
|
||||||
|
d.set_config(cfg)
|
||||||
|
print("Running debug_force_send() — may trigger ENOMEM, run once only")
|
||||||
|
d.debug_force_send("memory test")
|
||||||
|
except Exception as e:
|
||||||
|
print("test_send error:", e)
|
||||||
Loading…
x
Reference in New Issue
Block a user