OzWrapper.OzSigfox.OzSigfox.OzSigfox

class OzWrapper.OzSigfox.OzSigfox.OzSigfox

Sigfox LPWAN communication wrapper backed by the iWire driver.

Encodes sensor values into two fixed-size 12-byte binary packs and transmits them sequentially over the Sigfox network via the drivers.iWire.iWire.iWire module. Although it is conceptually related to SensorBase.SensorBase.GenericSensor, this class does not inherit from it; it operates as a standalone uplink worker.

Payload encoding:

  • pack_1 (frame id 0x10) carries environmental parameters packed as: p1 (PM1, 2B), p2 (PM2.5, 2B), temp (2B), hum (2B), uv (2B) and rain (1B). Two-byte values are scaled by 10 and packed big-endian unsigned (!H).

  • pack_2 (frame id 0x20) carries g1 (gas, 2B), g2 (gas concentration, 4B float !f), light (4B) and flood (1B).

  • Negative values are clamped to 0 before encoding to keep unsigned packing safe.

Variables:

  • iwire (drivers.iWire.iWire.iWire | None) – iWire driver instance, or None before setup() is called.

  • isRealtime (int) – Flag controlling which payloads to transmit. 1 selects only realtime payloads (d.rdata truthy), 0 selects only non-realtime payloads.

  • payloaddata (dict | None) – Latest sensor payload dictionary pulled from the queue, or None if nothing has been received.

  • pack_1 (dict[str, list[int]]) – Primary payload register map, mapping each send-code to [byte_offset, byte_length].

  • pack_2 (dict[str, list[int]]) – Secondary payload register map.

  • sigfox_pack_1 (list[bytes]) – Primary 12-byte binary transmit buffer, prefixed with frame id 0x10.

  • sigfox_pack_2 (list[bytes]) – Secondary 12-byte binary transmit buffer, prefixed with frame id 0x20.

Example

>>> from OzWrapper.OzSigfox.OzSigfox import OzSigfox
>>> sigfox = OzSigfox()
>>> sigfox.setup({"gpio": {"port": "/dev/ttyAMA2",
...                        "baud": 115200},
...               "rcz": "6"})

Initialize the wrapper with default pack maps and frame buffers.

Sets isRealtime to 0, clears payloaddata, and prepares pack_1/pack_2 register maps plus the sigfox_pack_1/sigfox_pack_2 12-byte buffers with their frame-id prefix bytes (0x10 / 0x20).

Parameters:

None.

Returns:

None.

Raises:

None – Construction performs only in-memory initialization.

Example

>>> from OzWrapper.OzSigfox.OzSigfox import OzSigfox
>>> sigfox = OzSigfox()
>>> len(sigfox.sigfox_pack_1)
12

Note

super().__init__() is invoked to remain forward-compatible with cooperative multiple inheritance even though OzSigfox currently has no explicit base class.

convert(val: float, points: int = 2) → list[int]

Pack a sensor value into a list of bytes for the Sigfox payload.

The packing format is chosen by points:

  • 1 -> struct.pack("!B", val) (unsigned 8-bit).

  • 2 -> val is multiplied by 10 and cast to int, then packed as struct.pack("!H", val) (unsigned 16-bit BE).

  • 4 -> struct.pack("!f", val) for floats (rounded to two decimal places) or struct.pack("!I", val) for integers (unsigned 32-bit BE).

Parameters:

  • val (float) – Sensor value to encode. May be int or float depending on points.

  • points (int) – Byte length and format selector. Must be one of 1, 2, or 4. Defaults to 2.

Returns:

Sequence of integer byte values ready to be written into sigfox_pack_1 or sigfox_pack_2.

Return type:

list[int]

Raises:

  • struct.error – If val does not fit the selected struct format (e.g. negative value with unsigned packing, or integer overflow).

  • TypeError – If val has an unsupported type for the chosen points (e.g. points=4 with a non-numeric value).

Example

>>> sigfox = OzSigfox()
>>> sigfox.convert(12.5, 2)
[0, 125]

Note

Callers are expected to clamp negative values to 0 before invoking convert(); the loop in loop() already performs this normalization.

loop(sigfoxqueue: queue.Queue) → None

Continuously consume payloads, encode them, and uplink via Sigfox.

Blocks on sigfoxqueue.get() for each iteration. The payload is accepted only if its d.rdata flag matches isRealtime semantics: realtime mode keeps payloads where rdata is truthy, non-realtime mode keeps payloads where rdata is missing or falsy. Each accepted payload is split across pack_1 and pack_2 register maps; values are clamped to >= 0, encoded via convert(), and copied into the corresponding offsets of sigfox_pack_1 / sigfox_pack_2. Both frames are then transmitted through the drivers.iWire.iWire.iWire driver with a 3-second gap between them to satisfy Sigfox duty-cycle constraints.

Parameters:

sigfoxqueue (queue.Queue) – Inter-thread queue that yields sensor data payload dictionaries shaped like {"d": {"rdata": <int>, "<sensor-key>": <value>, ...}}.

Returns:

This method runs an infinite loop and never returns under normal operation.

Return type:

None

Raises:

Exception – All exceptions raised inside the loop body are caught, logged via the contextual Oizom logger, and swallowed so the worker keeps running.

Example

>>> from queue import Queue
>>> q = Queue()
>>> sigfox.loop(q)

Note

The finally clause sleeps for one second every iteration, which throttles the loop even when payloads are missing or malformed. Adjust this interval with care: Sigfox imposes strict per-day uplink quotas.

run(sigfoxConfig: dict, sigfoxqueue: queue.Queue) → None

Set up the Sigfox modem and start the transmit loop when enabled.

Acts as the top-level entry point used by the application bootstrap. When sigfoxConfig['en'] is 1 it delegates to setup() to bring the drivers.iWire.iWire.iWire modem online and then hands control off to loop(). If the en flag is missing or zero the call returns immediately, leaving the Sigfox subsystem dormant.

Parameters:

  • sigfoxConfig (dict) – Configuration dictionary including the en enable flag (1 to start, anything else to skip) plus all keys consumed by setup().

  • sigfoxqueue (queue.Queue) – Queue forwarded to loop() to supply sensor payloads.

Returns:

None.

Raises:

Exception – Propagates errors raised by setup() during modem initialization; loop() internally swallows its own exceptions.

Example

>>> from queue import Queue
>>> sigfox = OzSigfox()
>>> sigfox.run({"en": 1, "rcz": "6"}, Queue())

Note

This method is typically invoked inside a dedicated worker thread because loop() never returns under normal operation.

setup(sigfoxConfig: dict) → None

Configure and initialize the underlying iWire Sigfox modem.

Resolves the serial port, baud rate, and Sigfox RCZ (radio configuration zone) from sigfoxConfig, falling back to defaults (/dev/ttyAMA2, 115200, RCZ 6). Also reads the rdata flag to drive isRealtime, then constructs an drivers.iWire.iWire.iWire instance and calls its initialize method.

Parameters:

sigfoxConfig (dict) –

Configuration mapping. Supported keys:

  • gpio.port (str): Serial device path.

  • gpio.baud (int): Serial baud rate.

  • rcz (str): Sigfox radio configuration zone.

  • rdata (int): Realtime-data flag for isRealtime.

Returns:

None.

Raises:

Exception – Propagates any error raised by drivers.iWire.iWire.iWire construction or initialize (e.g. serial port unavailable, modem not responding).

Example

>>> sigfox.setup({"gpio": {"port": "/dev/ttyAMA2",
...                        "baud": 115200},
...               "rcz": "6",
...               "rdata": 0})

Note

The default RCZ of "6" corresponds to Sigfox’s India zone; override it for deployments in other regions.

isRealtime = 0
iwire = None
pack_1
pack_2
payloaddata = None
sigfox_pack_1 = [b'\x10', b'\x00', b'\x00', b'\x00', b'\x00', b'\x00', b'\x00', b'\x00', b'\x00', b'\x00',...
sigfox_pack_2 = [b' ', b'\x00', b'\x00', b'\x00', b'\x00', b'\x00', b'\x00', b'\x00', b'\x00', b'\x00', b'\x00', b'\x00']