Core Modules

These modules form the backbone of the Oizom hardware application.

Sensor (Orchestrator)

Manager (Gateway Client)

HTTP client for communication with the Oizom Gateway container.

Provides methods to fetch device configuration, post sensor data (both periodic and realtime), send initialization payloads, update device configuration, and retrieve PPB analytics from the Gateway REST API.

All communication uses JSON over HTTP with an access token for authentication. The Gateway URL and API paths are configurable via environment variables.

Environment Variables:

GATEWAY_URL: Base URL of the Gateway (default: http://gateway:8080). GATEWAY_CONFIG_API: Config endpoint path (default: /query/config). GATEWAY_DATA_API: Data posting endpoint (default: /query/data). GATEWAY_INIT_API: Init endpoint path (default: /config/init). GATEWAY_ACCESS_TOKEN: JWT token for API authentication.

Typical usage:

from Manager import Manager

mgr = Manager()
config, status = mgr.getConfig()
mgr.sendData(data_payload, realtime_payload)

class Manager.Manager.Manager

Bases: object

HTTP client for the Oizom Gateway REST API.

Handles all communication between the hardware sensor layer and the Gateway container, including configuration fetching, data posting, and device initialization. All HTTP requests use a 10-second timeout and return status code 404 on any failure.

Variables:

• server – Base URL of the Gateway service.

• config_api – Path to the configuration endpoint.

• data_api – Path to the data posting endpoint.

• access_token – JWT token for API authentication.

• config_headers – HTTP headers sent with every request.

server = 'http://gateway:8080'

config_api = '/query/config'

data_api = '/query/data'

realtime_data_api = '/query/realtimeData'

automation_alert_api = '/alerts/automation'

init_api = '/config/init'

ppb_analytics = '/ppb/device/'

access_token = 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJlbWFpbCI6ImppbUBvaXpvbS5jb20iLCJhcGlDb3VudGVyIjowLCJpYXQiOjE0ODkyMzIwODYsImV4cCI6MTUyMDc4OTAxMn0.Cc5nwYiEv_WsV1sCMH_sY7QCKZ3dlPMQJVQYugaJbrk'

__init__()

Initialize the Manager with Gateway connection parameters.

Reads Gateway URL, API paths, and access token from environment variables, then constructs the HTTP headers used for all requests.

Return type:

None

getConfig()

Fetch the device configuration from the Gateway.

Sends a GET request to the Gateway’s configuration endpoint and returns the parsed JSON response along with the HTTP status code.

Return type:

tuple[dict, int]

Returns:

Tuple of (config_dict, status_code). Returns ({}, 404) on any request failure.

getPPBanalytics(lte, gte, deviceId, deviceType)

Fetch PPB (parts-per-billion) analytics data for a device.

Retrieves raw sensor analytics from the Gateway for a specified time range, used for calibration and quality control calculations.

Parameters:

• lte (int) – Upper bound epoch timestamp (less-than-or-equal).

• gte (int) – Lower bound epoch timestamp (greater-than-or-equal).

• deviceId (str) – Unique identifier of the target device.

• deviceType (str) – Device type string (unused in current implementation).

Return type:

tuple[Any, int]

Returns:

Tuple of (analytics_data, status_code). Returns ({}, 404) on failure.

updateConfig(config, deviceId)

Update the device configuration on the Gateway.

Sends a PUT request to update the configuration for a specific device. Used for automatic temperature compensation updates and similar runtime configuration changes.

Parameters:

• config (dict) – Configuration dictionary to apply.

• deviceId (str) – Unique identifier of the target device.

Return type:

int

Returns:

HTTP status code from the Gateway, or 404 on failure.

sendData(payload, r_data)

Post aggregated sensor data to the Gateway.

Sends the periodic data payload (d) and optional realtime data payload (r) to the Gateway’s data endpoint. This is the primary method for publishing sensor readings to the cloud.

Parameters:

• payload (<module ‘json’ from ‘/opt/buildhome/.asdf/installs/python/3.12.12/lib/python3.12/json/__init__.py’>) – Aggregated sensor data dictionary (the d payload).

• r_data (<module ‘json’ from ‘/opt/buildhome/.asdf/installs/python/3.12.12/lib/python3.12/json/__init__.py’>) – Realtime sensor data dictionary (the r payload), or None/empty to omit.

Return type:

int

Returns:

HTTP status code from the Gateway, or 404 on failure.

sendRealtimeData(payload)

Post realtime sensor data to the Gateway’s WebSocket relay.

Sends the latest sensor readings to the realtime data endpoint, which forwards them to connected WebSocket clients for live dashboard display.

Parameters:

payload (<module ‘json’ from ‘/opt/buildhome/.asdf/installs/python/3.12.12/lib/python3.12/json/__init__.py’>) – Realtime sensor data dictionary.

Return type:

int

Returns:

HTTP status code from the Gateway, or 404 on failure.

sendAutomationAlertsEmailSms(payload)

Send automation alert notifications via the Gateway.

Posts an alert payload to the Gateway’s automation alert endpoint, which triggers email and/or SMS notifications based on configured alert rules.

Parameters:

payload (<module ‘json’ from ‘/opt/buildhome/.asdf/installs/python/3.12.12/lib/python3.12/json/__init__.py’>) – Alert payload dictionary containing alert type, threshold values, and notification targets.

Return type:

int

Returns:

HTTP status code from the Gateway, or 404 on failure.

sendInit(payload)

Send device initialization data to the Gateway.

Posts hardware and firmware information to the Gateway’s init endpoint during device startup. This registers the device and its capabilities with the cloud platform.

Parameters:

payload (<module ‘json’ from ‘/opt/buildhome/.asdf/installs/python/3.12.12/lib/python3.12/json/__init__.py’>) – Initialization payload containing device ID, firmware version, sensor list, and hardware capabilities.

Return type:

int

Returns:

HTTP status code from the Gateway, or 404 on failure.

Network (Connectivity Monitor)

Network connectivity monitor for Oizom edge devices.

Provides internet reachability checking by probing configurable server endpoints via TCP socket connections. Runs as a background thread that continuously updates a shared Queue with connectivity status.

Also communicates with an external Network Manager service to query detailed interface status (Ethernet, GSM, WiFi) for LED indicator control and diagnostic reporting.

Environment Variables:

NETWORKMANAGER_URL: URL of the Network Manager service

(default: http://oz_terminal/network).

STATUS_API: Status endpoint path (default: /network/status).

Typical usage:

from Network import Network
from queue import Queue

net = Network()
q = Queue(maxsize=2)
q.put(0)  # placeholder
q.put(False)  # initial status
net.run(q)  # blocking -- run in a thread

class Network.Network.Network

Bases: object

Internet connectivity monitor and Network Manager client.

Probes a list of servers via TCP sockets to determine internet reachability, and queries the Network Manager service for detailed interface status. Designed to run in a background thread, continuously updating a shared Queue with the current connectivity state.

Variables:

• server – URL of the Network Manager service.

• serverList – List of server dicts ({"ip": ..., "port": ...}) to probe for connectivity.

• timeout – TCP connection timeout in seconds.

• looptime – Delay between connectivity checks in seconds.

• HOSTPOT_TIMER – Duration (seconds) for hotspot LED blinking mode.

Parameters:

• userList (list[dict[str, Any]] | None)

• timeout (int)

• looptime (int)

success_delay = 6

HOSTPOT_TIMER = -1

server = 'http://oz_terminal/network'

status_api = '/network/status'

__init__(userList=None, timeout=5, looptime=5)

Initialize the Network monitor.

Parameters:

• userList (list[dict[str, Any]] | None) – Optional list of server dicts to prepend to the default probe list. Each dict must have "ip" and optionally "port" keys.

• timeout (int) – TCP socket connection timeout in seconds.

• looptime (int) – Delay between connectivity check iterations in seconds.

Return type:

None

add_server(ip, port=80)

Add a server to the front of the probe list.

Parameters:

• ip (str) – IP address or hostname of the server.

• port (int) – TCP port to probe (default: 80).

Return type:

None

is_internet()

Check internet reachability by probing configured servers.

Iterates through serverList and attempts a TCP connection to each. Returns True on the first successful connection.

Return type:

bool

Returns:

True if any server in the list is reachable, False otherwise.

run(network_queue)

Run the connectivity check loop (blocking).

Continuously checks internet reachability and updates network_queue.queue[1] with the current status. Designed to be run in a background thread.

Parameters:

network_queue (Queue) – Shared Queue whose second element is updated with connectivity status (True/False or 2 for hotspot mode).

Return type:

None

putStatus(value)

Fetch network status and merge it into the given dictionary.

Parameters:

value (dict[str, Any]) – Dictionary to update with network status fields from the Network Manager service.

Return type:

None

identifyConnection()

Identify the current connection state for LED indicator control.

Queries the Network Manager for interface status and returns an integer code used to drive the device’s RGB LED indicator.

Return type:

int

Returns:

1 if connected (Ethernet, GSM, or WiFi), 2 if no connection and WiFi not associated (hotspot mode), 0 if status is unavailable.

getStatus()

Fetch detailed network interface status from the Network Manager.

Return type:

dict[str, object] | None

Returns:

Dictionary with "eth", "gsm", and "wifi" sub-dicts containing interface-level status, or None on request failure.

Configuration

Static configuration loader for Oizom environmental monitoring devices.

Reads base device configuration from Configuration/config.json at startup. This JSON file contains default settings inherited by all sensor wrappers, including sampling intervals, data formats, and device metadata.

The Config singleton is instantiated once by SensorBase.GenericSensor and its config attribute is shared across all wrapper instances.

Typical usage:

from Configuration import Config

cfg = Config()
print(cfg.config["sampling_interval"])

class Configuration.config.Config

Bases: object

Loader for the base device configuration file.

Reads and parses Configuration/config.json on instantiation, making the resulting dictionary available via the config attribute.

Variables:

config – Parsed contents of config.json. Contains base settings such as sampling intervals, device identifiers, and default parameter configurations shared by all sensor wrappers.

__init__()

Return type:

None

config: dict = {}

SensorBase (Abstract Base)

Abstract base class defining the interface for all Oizom sensor wrappers.

Every sensor wrapper in OzWrapper must inherit from GenericSensor and implement its four abstract methods. This ensures a consistent lifecycle across all sensor types:

1. initialize() – receive device configuration from the Gateway

2. initializeSensor() – set up individual hardware sensors

3. getSensorReading() – perform a measurement cycle and return data

4. putSensorValue() – merge sensor readings into the outbound payload

The orchestrator Sensor.Sensor calls these methods in order during its setup() and loop() phases.

Typical usage:

from SensorBase import GenericSensor

class OzTemp(GenericSensor):
    def initialize(self, config, init_value):
        ...
    def initializeSensor(self, sensor):
        ...
    def getSensorReading(self):
        return {"temp": 25.3, "hum": 60.0}
    def putSensorValue(self, value):
        value.update(self.getSensorReading())
        return value

class SensorBase.SensorBase.GenericSensor

Bases: object

Abstract base class for all Oizom sensor wrappers.

Provides a shared baseConfig loaded from Configuration/config.json and defines the four-method lifecycle contract that every wrapper must fulfill.

Subclasses are expected to set partNumber and parameter to their respective Gateway-assigned identifiers so the orchestrator can match incoming configuration entries to the correct wrapper.

Variables:

• baseConfig – Base device configuration dictionary loaded from Configuration/config.json. Shared across all wrapper instances.

• partNumber – Gateway part-number identifier used to match this wrapper to entries in the device configuration. Subclasses must override.

• parameter – Primary measurement parameter key (e.g., "pm", "temp"). Subclasses must override.

partNumber = 'pn'

parameter = 'pm'

__init__()

Return type:

None

baseConfig: ClassVar[dict] = {}

abstractmethod initialize(config, init_value)

Receive and store the device configuration from the Gateway.

Called once during Sensor.Sensor.setup() after the Gateway responds with the device configuration. Implementations should parse the config dictionary to determine which hardware sensors are enabled and store any initial calibration values.

Parameters:

• config (dict) – Device configuration dictionary from the Gateway containing sensor enable flags, calibration offsets, and part numbers.

• init_value (dict) – Initial sensor values dictionary, typically containing the last known readings before shutdown for continuity.

Return type:

None

abstractmethod initializeSensor(sensor)

Initialize a single hardware sensor based on its configuration entry.

Called once per enabled sensor during initialize(). Implementations should instantiate the appropriate driver, configure I2C/UART/SPI communication, and verify the sensor is responding.

Parameters:

sensor (dict) – Single sensor configuration entry from the Gateway, containing at minimum "pn" (part number) and bus/address information.

Return type:

None

abstractmethod getSensorReading()

Perform a measurement cycle and return the latest sensor data.

Called once per sensing loop iteration by the orchestrator. Implementations should read from all initialized hardware sensors, apply any calibration or averaging, and return a flat dictionary of parameter-value pairs.

Return type:

dict

Returns:

Dictionary mapping parameter keys (e.g., "temp", "hum", "pm2_5") to their measured float values. Returns zero-filled values on sensor read failure rather than raising.

abstractmethod putSensorValue(value)

Merge this wrapper’s sensor readings into the outbound data payload.

Called by the orchestrator to assemble the final data packet that gets posted to the Gateway. Implementations should call getSensorReading() and merge the result into value.

Parameters:

value (dict) – Accumulated data payload dictionary being built by the orchestrator across all active wrappers.

Return type:

dict

Returns:

The value dictionary with this wrapper’s readings merged in.

putInitvalues(value)

Merge initial/default values into the data payload.

Override in subclasses that need to inject startup defaults or last-known values into the first data payload after boot.

Parameters:

value (dict) – Accumulated data payload dictionary.

Return type:

dict

Returns:

The value dictionary, unmodified by default.

Indices (Environmental Calculations)

Environmental index calculator for derived meteorological parameters.

Computes derived environmental indices from raw sensor data, including Wet Bulb Globe Temperature (WBGT), air density, dew point, vapor pressure, evapotranspiration, heat index, wind chill, and Humidex. These indices are identified by keys i1 through i10 in the data payload and are configured per-device via the Gateway.

The calculations follow standard meteorological formulas from WMO and ASHRAE references. Each index method accepts the current data payload and returns the computed value, or None if required input parameters are missing.

Typical usage:

from Indices import Indices

idx = Indices()
idx.updateIndices(["i1", "i3", "i5"], data_payload, realtime_payload)

class Indices.Indices.Indices

Bases: object

Calculator for derived environmental indices.

Computes meteorological and environmental comfort indices from raw sensor readings. Each index is identified by a string key (i1 through i10) and computed from temperature, humidity, pressure, wind speed, and solar radiation inputs.

Index mapping:

• i1: WBGT indoor (Wet Bulb Globe Temperature)

• i2: WBGT outdoor

• i3: Air density (kg/m3)

• i4: Natural wet bulb temperature

• i5: Internal dew point

• i6: External dew point

• i7: Vapor pressure

• i8: Evapotranspiration (Penman-Monteith)

• i9: Heat index / Wind chill

• i10: Humidex

__init__()

Initialize the Indices calculator.

Return type:

None

updateIndices(indices_list, d_data_payload, r_data_payload)

Compute and inject all requested indices into the data payload.

Iterates through the list of requested index keys, computes each one from the current sensor data, and updates the data payload dictionary in place with the results.

Parameters:

• indices_list (list) – List of index key strings (e.g., ["i1", "i3"]).

• d_data_payload (dict) – Data payload dictionary to read inputs from and write computed indices into.

• r_data_payload (dict) – Realtime data payload, used by some indices (e.g., evapotranspiration) that need recent history.

Return type:

None

calculate_wbgt_indoor(payload)

Calculate indoor Wet Bulb Globe Temperature (WBGT).

Uses the formula WBGT_indoor = 0.7 * Tnwb + 0.3 * Tg where Tnwb is the natural wet-bulb temperature and Tg is the globe temperature.

Parameters:

payload (dict) – Sensor data dictionary containing keys "t3" (globe temp), "temp" (dry-bulb temp), and "hum" (relative humidity).

Return type:

float | None

Returns:

Indoor WBGT in degrees Celsius rounded to 2 decimal places, or None if required inputs are missing.

calculate_wbgt_outdoor(payload)

Calculate outdoor Wet Bulb Globe Temperature (WBGT).

Uses the formula WBGT_outdoor = 0.7 * Tnwb + 0.2 * Tg + 0.1 * Tdb which adds a dry-bulb component for outdoor solar exposure.

Parameters:

payload (dict) – Sensor data dictionary containing keys "t3" (globe temp), "temp" (dry-bulb temp), and "hum" (relative humidity).

Return type:

float | None

Returns:

Outdoor WBGT in degrees Celsius rounded to 2 decimal places, or None if required inputs are missing.

calculate_tnwb(payload)

Calculate natural wet-bulb temperature (Tnwb) using the Stull (2011) formula.

Parameters:

payload (dict) – Sensor data dictionary containing "temp" (dry-bulb temperature in degrees C) and "hum" (relative humidity in %).

Return type:

float | None

Returns:

Tnwb in degrees Celsius, or None if required inputs are missing.

calculate_air_density(payload)

Calculate air density using the ideal gas law.

Converts temperature to Kelvin and pressure to Pascals, then applies rho = P / (R_specific * T) with R_specific = 287.05.

Parameters:

payload (dict) – Sensor data dictionary containing "temp" (degrees C) and "pr" (pressure in hPa).

Return type:

float | None

Returns:

Air density in kg/m^3 rounded to 2 decimal places, or None if required inputs are missing.

calculate_internal_dew_point(payload)

Calculate the internal (box) dew point using the Magnus formula.

Uses the enclosure temperature ("t1") and enclosure humidity ("t2") readings from the BME280 inside the device housing.

Parameters:

payload (dict) – Sensor data dictionary containing "t1" (internal temp in degrees C) and "t2" (internal humidity in %).

Return type:

float | None

Returns:

Dew point in degrees Celsius rounded to 2 decimal places, or None if required inputs are missing.

calculate_external_dew_point(payload)

Calculate the external (ambient) dew point using the Magnus formula.

Uses the external SHT31 temperature ("temp") and humidity ("hum") readings.

Parameters:

payload (dict) – Sensor data dictionary containing "temp" (ambient temp in degrees C) and "hum" (ambient humidity in %).

Return type:

float | None

Returns:

Dew point in degrees Celsius rounded to 2 decimal places, or None if required inputs are missing.

calculate_vapour_pressure(payload)

Calculate actual vapour pressure from temperature and humidity.

Computes saturation vapour pressure via the Buck equation, then scales by relative humidity.

Parameters:

payload (dict) – Sensor data dictionary containing "temp" (degrees C) and "hum" (relative humidity in %).

Return type:

float | None

Returns:

Vapour pressure in hPa rounded to 2 decimal places, or None if required inputs are missing.

calculate_evapotranspiration(d_data_payload, r_data_payload)

Calculate reference evapotranspiration using the FAO Penman-Monteith method.

Requires realtime temperature and humidity arrays (min/max over the interval) plus averaged pressure, solar radiation, and wind speed.

Parameters:

• d_data_payload (dict) – Averaged data payload containing "pr" (hPa), "solar" (W/m2), and "ws" (wind speed in m/s).

• r_data_payload (dict) – Realtime data payload containing "temp" and "hum" as [values_list, timestamps_list] pairs.

Return type:

float | None

Returns:

Evapotranspiration in mm/day rounded to 2 decimal places, or None if required inputs are missing.

calculate_cloud_base_height(payload, r_data)

Estimate cloud base height from temperature and dew point spread.

Uses the rule of thumb CBH = (T - Td) * 125 where T is the ambient temperature and Td is the external dew point.

Parameters:

• payload (dict) – Sensor data dictionary containing "temp" and "hum" (used to derive dew point).

• r_data (dict) – Realtime data payload (unused, kept for interface consistency).

Return type:

float | None

Returns:

Cloud base height in metres rounded to 2 decimal places, or None if required inputs are missing.

Quality Control

Utilities

Logger

Structured logging framework for Oizom IoT edge devices.

Provides a centralized, configurable logging system with context-based filtering, timezone-aware timestamps, ANSI color output, and safe rotating file handlers. All modules in the Oizom hardware stack use this logger for consistent, machine-parseable log output.

The logging pipeline is built from composable components:

• LoggerConfig – immutable dataclass holding all configuration

• ContextFilter – injects a context token (e.g., "INIT", "READ") into every log record for easy grep-based filtering

• StructuredFormatter – pipe-delimited format with optional timezone conversion and country indicators

• ColorFormatter – ANSI color wrapper respecting NO_COLOR / TTY detection

• SafeRotatingFileHandler – rotation that tolerates missing backup files

• OizomLogger – public API combining all of the above

Example output formats:

INFO     | SENSOR     | Temperature reading: 25.3C
2025-12-10 14:30:45.123456 | INFO     | SENSOR     | Temperature reading: 25.3C

Typical usage:

from utils.oizom_logger import OizomLogger

context_logger = OizomLogger(__name__)
context_logger.info_with_context("INIT", "Sensor setup complete")

basic_logger = OizomLogger(__name__).get()
basic_logger.info("Standard log message")

Note

This module is imported by virtually every file in the codebase. Keep it free of circular imports – it must not import from Sensor, Manager, or any OzWrapper module.

class utils.oizom_logger.OizomLogger

Bases: object

Main logger wrapper - backward compatible with old oizom_logger.

USAGE:

logger = OizomLogger(__name__) logger.info_with_context(“INIT”, “Starting sensor setup”)

# Or get underlying logger for standard logging: log = logger.get() log.info(“Standard log message”)

Parameters:

• logger_name (str)

• kwargs (Any)

__init__(logger_name, **kwargs)

Initialize logger with optional config overrides.

WHY: Old logger took 30+ kwargs. We still accept them for compatibility, but internally convert to LoggerConfig for cleaner handling.

Parameters:

• logger_name (str)

• kwargs (Any)

Return type:

None

install_globally(replace_existing=True)

Attach this logger’s handlers to root logger (for 3rd party libs).

Parameters:

replace_existing (bool)

Return type:

None

log_with_context(level, context, message, *args, **kwargs)

Core method for logging with context token.

Parameters:

• level (str)

• context (str)

• message (str)

• args (Any)

• kwargs (Any)

Return type:

None

debug_with_context(context, message, *args, **kwargs)

Parameters:

• context (str)

• message (str)

• args (Any)

• kwargs (Any)

Return type:

None

info_with_context(context, message, *args, **kwargs)

Parameters:

• context (str)

• message (str)

• args (Any)

• kwargs (Any)

Return type:

None

warning_with_context(context, message, *args, **kwargs)

Parameters:

• context (str)

• message (str)

• args (Any)

• kwargs (Any)

Return type:

None

error_with_context(context, message, *args, **kwargs)

Parameters:

• context (str)

• message (str)

• args (Any)

• kwargs (Any)

Return type:

None

critical_with_context(context, message, *args, **kwargs)

Parameters:

• context (str)

• message (str)

• args (Any)

• kwargs (Any)

Return type:

None

get()

Return underlying logger for standard logging calls.

Return type:

Logger

utils.oizom_logger.apply_logger_config(config_dict)

Reconfigure all loggers at runtime (useful for config file changes).

Parameters:

config_dict (dict[str, Any])

Return type:

None

Reboot Manager

Raspberry Pi reboot manager with cooldown protection.

Provides controlled device rebooting with a configurable cooldown period (default 4 hours) to prevent reboot loops. Each reboot event is logged to a persistent file so the cooldown survives process restarts.

The reboot is triggered via the Linux SysRq mechanism (/proc/sysrq-trigger), which performs an immediate hardware reboot without a clean shutdown. This is used as a last-resort recovery mechanism when the device enters an unrecoverable error state.

Typical usage:

from utils.reboot_RPi import RebootManager

manager = RebootManager()
manager.reboot_raspberry_pi({
    "type": "events",
    "t": 1700000000000,
    "msg": {"type": "watchdog", "value": "sensor_timeout"},
})

Warning

The SysRq reboot is immediate and ungraceful. Filesystem corruption is possible if writes are in progress. The 5-second delay before triggering allows pending I/O to flush.

class utils.reboot_RPi.RebootManager

Bases: object

Manages Raspberry Pi reboots with a cooldown period to prevent loops.

Tracks reboot history in a persistent log file and enforces a minimum interval between reboots. This prevents cascading reboot loops when a persistent hardware or configuration error triggers repeated reboot requests.

Variables:

• LOG_FILE – Path to the persistent reboot log file.

• COOLDOWN_SECONDS – Minimum interval between reboots in seconds (default: 4 hours).

LOG_FILE = 'reboot_log.txt'

COOLDOWN_SECONDS = 14400

read_last_reboot()

Read the timestamp of the most recent reboot from the log file.

Parses the last line of LOG_FILE looking for a t=<epoch_ms> field. Returns None if the file does not exist, is empty, or cannot be parsed.

Return type:

int | None

Returns:

Epoch timestamp in milliseconds of the last reboot, or None if no previous reboot is recorded.

write_reboot_log(payload)

Append a reboot event to the persistent log file.

Writes a pipe-delimited line containing the event timestamp, type, and message details. Uses fsync to ensure the write is durable before the impending reboot.

Parameters:

payload (dict) – Reboot event dictionary with keys "t" (epoch ms), "type" (event category), and "msg" (dict with "type" and "value" sub-keys).

Return type:

bool

Returns:

True after successfully writing and flushing the log entry.

reboot_raspberry_pi(payload)

Initiate a Raspberry Pi reboot if the cooldown period has elapsed.

Checks the reboot log to determine whether the cooldown period has passed since the last reboot. If the cooldown is still active, the reboot is blocked and a log message is emitted. Otherwise, the event is logged and an immediate SysRq reboot is triggered.

Parameters:

payload (dict) –

Reboot event dictionary with the following structure:

{
    "type": "events",
    "t": <epoch_ms>,
    "msg": {"type": "...", "value": "..."}
}

Return type:

None