The Python MQbtT library¶

This module supports the implementation of the optimizer side communicating with a miner (be it ESP-miner running on some ESP-based board, or cgminer running on a Linux machine controlling some USB-connected ASIC miner).

Usage of the library¶

The usage is very simple: first you need to create a concrete subclass of MQbtTBase that will be used to communicate with the miner:

MQbtTOptimizer to implement the optimizer,
MQbtTLogReceiver to receive the logs from the miner,
MQbtTDebugReceiver to receive coredumps sent by the miner after a panic,
MQbtTMiner to implement a (mock) miner.

The pair id is a string that should be unique for the pair of optimizer and miner, and should be the same for both parties.

class MQbtTBase(broker_uri: str, pair_id: str, receive_subtopics: list[tuple[str, int]], keepalive: int = 60, clean_session: bool = False)¶

Base class for all MQbtT classes.

This class is not meant to be used directly, but only as a base class for other classes. It provides the basic functionality to connect to the MQTT broker and to send and receive messages.

Parameters:

broker_uri – the URI of the MQTT broker (example: mqtt://localhost)
pair_id – the pair id of the client
receive_subtopics – a list of (subtopic, qos) to receive from (i.e.: esp, srv or log)
keepalive – maximum number of seconds allowed between communications with the broker, passed to paho.mqtt.client.Client.connect()
clean_session – if True, the broker discards this client’s subscription and any queued QoS 1/2 backlog the moment it disconnects; if False (the default here), the broker keeps a persistent session across reconnects, so a briefly-disconnected client does not lose QoS 1/2 messages queued for it in the meantime. Concrete subclasses may default this differently according to their own reliability/broker-load trade-off (see MQbtTLogReceiver).

Note

Receiving methods

This class uses internally a Queue per subtopic, starts the MQTT Client and registers a callback that pushes all the messages for the subtopic to its queue.

Subclasses use such queues to provide a synchronous way to receive messages via receive_T() methods (where T is the type of message the subclass handles), and check via has_T() if there are messages to be received.

Receiving methods have a timeout parameter that (if positive) specifies the maximum number of seconds to wait for the queue to become not empty. If for whatever reason a message does not show up in the queue within the timeout, the receiving method raises a Empty exception.

In case the timeout parameter is None (the default), the blocking parameter can be set to true so that the call blocks until the queue is not empty; if set to false, the call returns immediately (possibly a None if the queue is empty), but never raises the Empty exception.

Sending methods

Subclasses provide send_T() to send messages of type T.

Such methods have a blocking parameter that if set to true (the default) makes the method block until the message is sent to the MQTT broker.

Important notes

There is a subtle difference between receiving and sending methods: on one hand, for receiving methods, blocking or timeouts may depend (indistinguishably) both on the MQTT broker or the sender, this implies that there is no way to distinguish if the reason for the exception is a networking issue, the broker or sender malfunctioning, or simply that the sender is silent. On the other hand, for sending methods, blocking depends just on the MQTT broker (that may be unreachable due to networking issues, or malfunctioning), but never on the receiver.

It is of paramount importance for users of subclasses to receive messages in order for the queue non to grow indefinitely. Once the communication is over, the stop() method must be called to stop the MQTT client and to disconnect from the broker.

P.S. on clean_session=False (the default for MQbtTBase, MQbtTOptimizer and MQbtTMiner): this class generates a fresh, random client_id on every instantiation, so the persistence guarantee only applies within one running instance. Two distinct situations follow from this, and it’s worth not conflating them:

A network hiccup while an instance keeps running is exactly what clean_session=False is for: paho reconnects the same underlying client (same client_id, same session), and the broker delivers whatever QoS 1/2 backlog queued up while disconnected. No message loss, nothing to configure.

Stopping an instance and starting a new one (e.g. restarting the optimizer with new code) is not a reconnect - the new instance gets a new client_id and therefore a brand new session; it does not, and should not, resume the old one. The old session is abandoned on the broker as a persistent, disconnected session, and will keep accumulating whatever QoS 1/2 traffic matches its subscriptions until it is cleaned up. This library does not attempt to bound that cost - it is a broker configuration concern (see persistent_client_expiration and max_queued_messages in mosquitto.conf(5)), not something a stable/derived client_id should paper over, since the whole point of starting a new instance is that it shouldn’t inherit the old one’s session.

stop()¶: Stop the MQTT client and disconnect from the broker.

Messages format¶

The parties exchange messages of various types:

Request is sent by the optimizer to the miner,
Reply and Share are sent by the miner to the optimizer.

Messages use dataclasses, in order to reduce the risk of programming errors, they are immutable and their constructor accepts only keyword arguments; to create a message, you need to create an instance of the desired class, for instance:

request = Request(
  timestamp_min = 3, 
  timestamp_max = 10, 
  nonce_start = 5, 
  nonce_size = 7, 
  reset = False
)

and to access the fields, you can use the dot notation:

request.timestamp_min
request.timestamp_max
request.nonce_start
request.nonce_size
request.reset

Messages are defined as follows. Observe that the from_bytes() and to_bytes() methods are usually not used directly, but are needed by the send() and receive() methods of the subclasses of MQbtTBase to serialize and deserialize the messages for transmission.

class Request(*, timestamp_min: int = 0, timestamp_max: int = 0, nonce_start: int = 0, nonce_size: int = 0, reset: bool = False)¶

Request message sent by the optimizer to the miner.

Parameters:

timestamp_min – Minimum timestamp.
timestamp_max – Maximum timestamp.
nonce_start – Minimum nonce.
nonce_size – Maximum nonce.
reset – Whether to reset the state.

static from_bytes(b: bytes) → Request¶

Deserialize bytes into a Request object.

Parameters:: b – The bytes to deserialize.
Returns:: The deserialized Request object.

to_bytes() → bytes¶

Serialize the Request to bytes using protobuf.

Returns:: The serialized protobuf representation of the request.

class Reply(kind: Kind = Kind.RESULT, num_shares: int = 0, new_block: bool = False)¶

Reply message from the miner to the optimizer.

Parameters:

kind – The kind of reply (e.g., RESULT).
num_shares – Number of shares in the reply.
new_block – Whether a new block was found.

static from_bytes(b: bytes) → Reply¶

Deserialize bytes into a Reply object.

Parameters:: b – The bytes to deserialize.
Returns:: The deserialized Reply object.

to_bytes() → bytes¶

Serialize the Reply to bytes using protobuf.

Returns:: The serialized protobuf representation of the reply.

The Kind enum is defined as

class Kind(value)¶

Kind of the reply messages.

BYE = 3¶

HELLO = 0¶

RESET = 2¶

RESULT = 1¶

Shares are represented by

class Share(version: int, prev_block: bytes, merkle_root: bytes, time: int, pool_time: int, bits: int, nonce: int, pool_diff: int, job_id: str, extranonce2: str, diff: float, zeroes: int)¶

Share message from the miner to the optimizer.

Parameters:

version – The version rolled by the miner.
prev_block – The hash of the previous block.
merkle_root – The bytes of the merkle root.
time – The timestamp of the share.
pool_time – The timestamp provided by the mining pool.
bits – The difficulty bits.
nonce – The nonce rolled by the miner.
pool_diff – The pool difficulty.
job_id – The job ID.
extranonce2 – The extranonce2.
diff – The difficulty of the share.
zeroes – The (approximate) number of leading binary zeroes in the share.

static from_bytes(b: bytes) → Share¶

Deserialize bytes into a Share object.

Parameters:: b – The bytes to deserialize.
Returns:: The deserialized Share object.

to_bytes() → bytes¶

Serialize the Share to bytes using protobuf.

Returns:: The serialized protobuf representation of the reply.

Finally, a miner sends a coredump after a panic as a sequence of CoredumpChunk messages, one per chunk, sharing the same uuid and each aware of total_chunks:

class CoredumpChunk(uuid: bytes, chunk_index: int, total_chunks: int, data: bytes)¶

One chunk of a raw ESP32 coredump sent after a panic.

Parameters:

uuid – 16 raw bytes identifying which coredump this chunk belongs to.
chunk_index – 0-based index of this chunk.
total_chunks – Total number of chunks in this coredump.
data – Raw coredump bytes for this chunk.

static from_bytes(b: bytes) → CoredumpChunk¶

Deserialize bytes into a CoredumpChunk object.

Parameters:: b – The bytes to deserialize.
Returns:: The deserialized CoredumpChunk object.

to_bytes() → bytes¶

Serialize the CoredumpChunk to bytes using protobuf.

Returns:: The serialized protobuf representation of the chunk.

Unlike the other messages, a single CoredumpChunk is not a complete logical unit - MQbtTDebugReceiver reassembles a full set of them (see below) into a Coredump, which is not itself a wire message (it has no to_bytes/from_bytes):

class Coredump(uuid: bytes, data: bytes)¶

A fully reassembled ESP32 coredump.

Not a wire message itself (no to_bytes()/from_bytes()) - built by reassembling a complete set of CoredumpChunk messages sharing the same uuid, either live (mqbtt.mqbtt.MQbtTDebugReceiver.receive_core()) or from archived chunks (assemble_coredump_chunks()).

Parameters:

uuid – 16 raw bytes identifying this coredump.
data – The concatenated raw coredump bytes.

The reassembly itself - verifying every chunk_index in [0, total_chunks) is present before trusting the result, rather than silently returning a truncated coredump - is centralized in a single function, used both by MQbtTDebugReceiver and by any offline consumer reassembling archived chunks (e.g. from a log store):

assemble_coredump_chunks(uuid: bytes, chunks_by_index: dict[int, bytes], total_chunks: int) → Coredump¶

Reassemble a complete set of coredump chunks into a Coredump.

The single place this gap-check/concatenate logic lives - both mqbtt.mqbtt.MQbtTDebugReceiver (live, incremental) and any offline consumer reassembling archived chunks should call this rather than reimplementing the check.

Parameters:

uuid – The coredump’s uuid (only used to build the returned Coredump).
chunks_by_index – chunk_index -> data, for every chunk collected so far.
total_chunks – The total_chunks value reported by the chunks.

Returns:

The reassembled Coredump.

Raises:

ValueError – If any chunk_index in [0, total_chunks) is missing.

The `MQbtTOptimizer`¶

The central class of the library is the one that allows to implement the optimizer.

class MQbtTOptimizer(broker_uri: str, pair_id: str, keepalive: int = 60, clean_session: bool = False)¶

Class for the optimizer to send and receive messages to the miner.

has_reply() → bool¶

Check if there is a reply from the miner.

Returns:: True if there is a reply, False otherwise.

has_share() → bool¶

Check if there is a share from the miner.

Returns:: True if there is a share, False otherwise.

receive_reply(blocking: bool = True, timeout: float | None = None) → Reply | None¶

Receive a reply from the miner.

Parameters:

blocking – Whether to block until the queue becomes not empty.
timeout – The maximum time to wait for the queue to become not empty.

receive_share(blocking=True) → Share | None¶

Receive a share from the miner.

Parameters:: blocking – Whether to block until a share is received.

send_request(request: Request, blocking: bool = True) → None¶

Send a request to the miner.

Parameters:: blocking – Whether to block until the request is sent.

A sketch of its usage is given by the following fragment of code

from mqbtt import MQbtTOptimizer, Request, Kind

# create the connection
handle = MQbtTOptimizer('parsifal.law.di.unimi.it', 'test')

while True:

  # prepare the request
  request = Request(
    timestamp_min= ...,
    timestamp_max= ..., 
    nonce_start = ..., 
    nonce_size = ..., 
    reset=...
  )
  
  # send the request 
  handle.send_request(request)

  ... 

  # receive the reply
  reply = handle.receive_reply()
  
  # process the reply
  if (reply.kind == Kind.RESULT):
    ... reply.num_shares ...
    ... reply.new_block ...

# end the communication
handle.stop()

The `MQbtTLogReceiver`¶

Receiving logs is very easy.

class MQbtTLogReceiver(broker_uri: str, pair_id: str, keepalive: int = 60, clean_session: bool = True)¶

Class to receive log messages from the MQTT broker.

Unlike MQbtTBase, clean_session defaults to True here: log messages are informational (QoS 1), so it is preferable to bound the broker-side backlog a disconnected or abandoned receiver could accumulate rather than to guarantee delivery of every log line.

receive_log(blocking: bool = True, timeout: float | None = None) → LogEntry | None¶

Receive a log message from the MQTT broker.

Parameters:

blocking – Whether to block until the queue becomes not empty.
timeout – The maximum time to wait for the queue to become not empty.

A sketch of its usage is given by the following fragment of code

from mqbt import MQbtTLogReceiver

handle = MQbtTLogReceiver('parsifal.law.di.unimi.it', 'test')

while True:
  message = handle.receive_log()
  ...

handle.stop()

The `MQbtTDebugReceiver`¶

To capture the coredumps a miner sends after a panic, use MQbtTDebugReceiver. Unlike the other receivers, receive_core() does not correspond to a single incoming MQTT message: it accumulates CoredumpChunk messages internally (possibly for several concurrent coredumps, if a device panics again before an earlier one finishes) and only returns once one coredump’s chunks are all present, as a complete Coredump.

class MQbtTDebugReceiver(broker_uri: str, pair_id: str, keepalive: int = 60, clean_session: bool = False, on_stale=None, stale_after: float = 30.0)¶

Class to receive coredump chunks from the MQTT broker and reassemble them into complete Coredump objects.

Unlike the other receivers, one wire message (a CoredumpChunk) is not one logical unit: receive_core() accumulates chunks internally - possibly for several concurrent coredumps at once, if a device panics more than once before an earlier one finishes - and only returns once some coredump’s chunk_indexs are all present.

Unlike MQbtTLogReceiver, clean_session defaults to False here: a coredump can be sent at any time, and missing it because this receiver was briefly disconnected would lose data that will not be resent - the same persistent-session trade-off as MQbtTMiner/ MQbtTOptimizer, not the throwaway-friendly default the log mode uses.

Parameters:

on_stale – optional callback (uuid, received, total) invoked whenever an incomplete coredump is evicted for having gone quiet for longer than stale_after seconds - the only way to observe a chunk that will never arrive, since receive_core() otherwise never returns for that coredump.
stale_after – seconds of inactivity on a given coredump’s uuid before it is evicted, to bound memory from a coredump that will never complete (e.g. a dropped chunk).

receive_core(blocking: bool = True, timeout: float | None = None) → Coredump | None¶

Receive a complete coredump from the MQTT broker, accumulating individual chunks internally until one coredump is complete.

Parameters:

blocking – Whether to block until a complete coredump is assembled.
timeout – The maximum total time to wait for a complete coredump - unlike the other receive_T methods, this may span several underlying chunk arrivals, not just one.

A sketch of its usage is given by the following fragment of code

from mqbtt import MQbtTDebugReceiver

handle = MQbtTDebugReceiver('parsifal.law.di.unimi.it', 'test')

while True:
  core = handle.receive_core()
  with open(f'{core.uuid.hex()}.core', 'wb') as f:
    f.write(core.data)

handle.stop()

The `MQbtTMiner`¶

For debugging purposes, the library provides a class to implement a (mock) miner.

class MQbtTMiner(broker_uri: str, pair_id: str, keepalive: int = 60, clean_session: bool = False)¶

Class for a (mock) miner to send and receive messages to the optimizer.

receive_request(blocking: bool = True, timeout: float | None = None) → Request | None¶

Receive a request from the optimizer.

Parameters:

blocking – Whether to block until the queue becomes not empty.
timeout – The maximum time to wait for the queue to become not empty.

send_reply(reply: Reply, blocking: bool = True) → None¶

Send a reply to the optimizer.

Parameters:: blocking – Whether to block until the reply is sent.

A sketch of its usage is pretty symmetrical to the one of the optimizer

from mqbt import MQbtTMiner, Reply

handle = MQbtTMiner(broker_address, pair_id)

while True:
  
  # receive the request from the optimizer
  request = handle.receive_request()

  # process the request and prepare the reply

  ... request.timestamp_min ...
  ... request.timestamp_max ...
  ... request.nonce_start ...
  ... request.nonce_size ...
  ... request.reset ...

  reply = Reply(
    num_shares= ...,
    new_block= ...,
    kind= Kind...
  )

  # send the reply
  handle.send_reply(reply)

handle.stop()

The `MQbtTDatabase` class¶

As seen in the next section, the library has a tool that can be used to record all the traffic in the mqbtt subtopics in a SQLite with the following schema

TABLE messages (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  timestamp REAL NOT NULL,
  pair_id TEXT NOT NULL,
  subtopic TEXT NOT NULL CHECK (subtopic IN ('esp', 'log', 'shr', 'srv')),
  message BLOB NOT NULL
);

where timestamp is the time of the message in seconds since the epoch for the UTC timezone, pair_id is the pair id of the optimizer and miner, subtopic is one of the four subtopics of the mqbtt topic, and message is the serialized message (in bytes).

Once the traffic is recorded, one can conveniently use the replies class to read and deserialize the data from the database to further process it.

class MQbtTDataBase(path: str, batch_size: int = 100, batch_timeout: float = 10.0)¶

A SQLite database for MQbtT messages.

close()¶: Close the database connection.

insert(record: MQbtTRecord)¶: Insert a mqtt record into the database with batched commits.

logs(pair_id: str | None = None, min_timestamp: datetime | None = None, max_timestamp: datetime | None = None, text: str | None = None, desc: bool = False) → Generator[MQbtTRecord, None, None]¶

Query the database for log messages, auto-deserializing the message payload.

Parameters:

pair_id – The pair_id to filter by.
min_timestamp – The minimum timestamp to filter by.
max_timestamp – The maximum timestamp to filter by.
text – Optional text to filter log messages by.
desc – Whether to order the results in descending timestamp order.

records(pair_id: str | None = None, subtopic: str | None = None, min_timestamp: datetime | None = None, max_timestamp: datetime | None = None, desc: bool = False) → Generator[MQbtTRecord, None, None]¶

Query the database for messages, auto-deserializing the message payload.

Parameters:

pair_id – The pair_id to filter by.
subtopic – The subtopic to filter by.
min_timestamp – The minimum timestamp to filter by.
max_timestamp – The maximum timestamp to filter by.
desc – Whether to order the results in descending timestamp order.

The class is conveniente because can be used to obtain a list of deserialized messages represented by the following class

class MQbtTRecord(*, id: int, pair_id: str, subtopic: str, timestamp: datetime, payload: bytes)¶

A (deserialized) record from a MQbtTDataBase.

Parameters:

id – The ID of the message in the database.
pair_id – The pair_id of the message.
subtopic – The subtopic of the message.
timestamp – The timestamp of the message recording.
message – The deserialized message payload (may be None).
payload – The raw message payload (bytes, may be None).

External services APIs¶

The library provides some simplified way of calling AxeOS and OpenObserve APIs.

Querying OpenObserve¶

Most of the MQTT traffic ends up in OpenObserve; the O2Query class provides a read-only client for its search API, taking care of the details that are easy to get wrong: credentials and organization ids, the microsecond time window, the stream type, missing streams, transient failures, and the chunking and paging needed to fetch more rows than a single request can return.

class O2Query(instance: str | tuple[str, str] = 'main', org: str = 'massimo', stream_type: str = 'logs', timeout: float = 180, retries: int = 5)¶

A read-only client for the OpenObserve search API.

All the methods accepting a time range take its ends as Timestamp (see to_us()), the range is [start, end); the _timestamp field of the returned rows is in microseconds since the epoch. Failed requests are retried with exponential backoff, then reported as APIException.

Parameters:

instance – The name of a known instance (legacy or main), or a (url, password) pair.
org – The name of a known organization (massimo or adam), or an organization id.
stream_type – The type of the queried streams, logs or metrics; a mismatch makes OpenObserve report the stream as not found.
timeout – The timeout of each request, in seconds.
retries – The number of times a failed request is retried.

bounds(stream: str, start: datetime | str | int | float, end: datetime | str | int | float, where: str | None = None) → tuple[int, int] | None¶

Returns the first and last _timestamp of a stream in [start, end).

Parameters:

stream – The stream name.
start – The start of the time range.
end – The end of the time range.
where – An optional SQL condition (without the WHERE keyword).

Returns:

The (first, last) timestamps, or None if there are no rows.

count(stream: str, start: datetime | str | int | float, end: datetime | str | int | float, where: str | None = None) → int¶

Counts the rows of a stream in [start, end).

Parameters:

stream – The stream name.
start – The start of the time range.
end – The end of the time range.
where – An optional SQL condition (without the WHERE keyword).

fetch(stream: str, start: datetime | str | int | float, end: datetime | str | int | float, fields: str = '*', where: str | None = None, page_size: int = 10000) → Iterator[dict]¶

Yields all the rows of a stream in [start, end), in chronological order.

The span of the matching rows is split in as many equal chunks as needed for each to hold about page_size rows (were they uniformly distributed), and each chunk is fetched in pages of page_size rows.

Parameters:

stream – The stream name.
start – The start of the time range.
end – The end of the time range.
fields – The fields to return, as a SQL select list.
where – An optional SQL condition (without the WHERE keyword).
page_size – The maximum number of rows per request.

latest(stream: str, start: datetime | str | int | float, end: datetime | str | int | float, where: str | None = None, fields: str = '*') → dict | None¶

Returns the most recent row of a stream in [start, end).

Parameters:

stream – The stream name.
start – The start of the time range.
end – The end of the time range.
where – An optional SQL condition (without the WHERE keyword).
fields – The fields to return, as a SQL select list.

search(sql: str, start: datetime | str | int | float, end: datetime | str | int | float, size: int = 10000, offset: int = 0) → list[dict]¶

Runs a SQL query on [start, end).

Parameters:

sql – The query.
start – The start of the time range.
end – The end of the time range.
size – The maximum number of rows to return.
offset – The number of rows to skip.

Returns:

The matching rows; none if the stream does not exist.

streams() → dict[str, str]¶: Returns the name to type map of the streams of the organization.

Time ranges can be given as :class:~datetime.datetime, as local time strings, or as microseconds since the epoch (the unit of the _timestamp field); the following helpers convert among these representations:

to_us(when: datetime | str | int | float) → int¶

Converts a point in time to microseconds since the epoch.

Parameters:: when – An aware (or naive, taken as local time) datetime, a string in YYYY-MM-DD HH:MM:SS format in local time, or a number (taken as microseconds if above 1e12, as seconds otherwise).

from_us(us: int, tz=datetime.timezone(datetime.timedelta(0), 'UTC')) → datetime¶

Converts microseconds since the epoch to an aware datetime.

Parameters:

us – The microseconds since the epoch (as in the _timestamp field).
tz – The timezone of the result, local time by default.

now_us() → int¶: Returns the current time in microseconds since the epoch.

window(hours: float, end: datetime | str | int | float | None = None) → tuple[int, int]¶

Returns the (start, end) microseconds of a window of the given duration.

Parameters:

hours – The duration of the window, in hours.
end – The end of the window, now by default.

chunks(start: datetime | str | int | float, end: datetime | str | int | float, hours: float) → Iterator[tuple[int, int]]¶

Splits [start, end) in consecutive windows of the given duration.

Parameters:

start – The start of the range.
end – The end of the range.
hours – The duration of each window, in hours (the last one may be shorter).

A sketch of the usage (the rows are plain dictionaries, so they can be turned into a pandas.DataFrame directly) is

from mqbtt.o2query import O2Query, window

o2 = O2Query('main')
start, end = window(hours=24)
n = o2.count('shares_detail', start, end, "miner = 'gamma01'")
rows = list(o2.fetch('shares_detail', start, end, '_timestamp, miner, diff'))

Aggregates over long windows are best computed one chunk at a time (an aggregate cannot be paged):

for s, e in chunks('2026-08-01 00:00:00', '2026-09-01 00:00:00', hours=24):
  for row in o2.search('SELECT miner, count(*) AS n FROM gammas GROUP BY miner', s, e):
    ...

Command line tools¶

The library comprises sveral command line tools that can be used to test and the communication with the miner and the optimizer, as well as automatically upgrade to the latest version.

`mqbtt_dump`¶

As a convenient debugging tool, the mqbtt_dump command is provided, its usage is

Usage: mqbtt_dump [OPTIONS]

Options:
  --version                       Show version and exit.
  --broker TEXT                   MQTT host, default: parsifal.law.di.unimi.it
  --pid TEXT                      Pair id  [required]
  --mode [log|miner|optimizer|core]
                                   Operation mode
  --out-dir TEXT                  Directory to write '<uuid>.core' files to
                                   (mode 'core' only), default: cwd
  --help                          Show this message and exit.

Once the broker host and pair id are specified, the mqbtt_dump command, acting as the operation mode requires, will report:

log: the log messages received from the miner,
optimizer: the replies the optimizer receives from the miner,
miner: the requests the miner receives from the optimizer,
core: reassembles the coredumps a miner sends after a panic (see MQbtTDebugReceiver), writing each one to <uuid>.core under --out-dir (default: the current directory) as soon as it is complete, printing the path and size.

`mqbtt_record`¶

To keep track of the traffic, the mqbtt_record command is provided, its usage is

Usage: mqbtt_record [OPTIONS]

Options:
  --version       Show version and exit.
  --broker TEXT   MQTT host, default: parsifal.law.di.unimi.it
  --db PATH       Path to the record database.  [required]
  --exclude TEXT  Comma separated list of subtopics to exclude (in: esp, log,
                  shr, srv).
  --help          Show this message and exit.

To record the traffic, once broker host and the path to the database where to record the messages are provided, the command will record the traffic in the SQLite database until stopped.

The recorded traffic can then be conveniently read using the replies class as shown in the examples/Analysis.ipynb Jupyter notebook.

Recovering the logs¶

As an example, once some traffic is in the database, the logs can be retrieved with the examples/db2logspy script that, invoked like

python examples/db2logs.py traffic.db gamma01

will generate an output similar to the one below

2025-05-22 11:15:46+0200	I (4398308) create_jobs_task_Stefano: nTime/nonce range COMPLETED: 0 ACCEPTED shares found.
2025-05-22 11:15:46+0200	I (4398313) mqbtt_mqtt: Published to mqbtt/gamma01/esp, msg_id=23958
2025-05-22 11:15:46+0200	I (4398316) mqbtt_protocol: Sent reply: kind=1, num_shares=0, new_block=0
2025-05-22 11:15:47+0200	I (4398562) mqbtt_protocol: Received request: timestamp_min=2048, timestamp_max=3071, nonce_start=120, nonce_size=8, reset=0
2025-05-22 11:15:47+0200	I (4398566) create_jobs_task_Stefano: Server asked NEW nTime range: (2048-3071)
2025-05-22 11:15:47+0200	I (4398574) create_jobs_task_Stefano: Server asked NEW Nonce range: 16/512 values (3.12% fullscan) starting at Byte1=240
2025-05-22 11:15:47+0200	I (4398584) create_jobs_task_Stefano: Inter-job time: 4.0 ms
2025-05-22 11:15:47+0200	I (4398591) create_jobs_task_Stefano: Starting Nonce range: 16/512 values (3.12% fullscan) starting at Byte1=240 - nTime range: 2048-3071, old Block
2025-05-22 11:15:47+0200	I (4399069) bm1370Module: Job ID: 48, Core: 20/6, Ver: 00024000
2025-05-22 11:15:47+0200	I (4399072) asic_result: Ver: 20024000 Nonce C5E6FC28 diff 1636.0 of 32768.```

Converting the recorded shares to CSV¶

As an example, once some traffic is in the database, it can be converted with the examples/db2csv.py script that, invoked like

python examples/db2csv.py traffic.db traffic.csv

will generate a CSV file similar to the one below

timestamp,pair_id,nonce,version,diff,pool_diff,time,pool_time,zeroes
2025-10-13 09:09:38+0200,gamma08,3788964542,580968448,369.13252686314024,32768,1760339364,1760339364,40
2025-10-13 09:09:38+0200,gamma05,2318664658,603504640,297.57749591781317,32768,1760339364,1760339364,40
2025-10-13 09:09:38+0200,gamma03,272694464,536879104,441.90105702054825,32768,1760340295,1760339364,40
2025-10-13 09:09:38+0200,casa01,2240539576,536879104,257.5074550736473,1024,1760339966,1760339341,40
2025-10-13 09:09:38+0200,gamma07,1836254274,638377984,277.5211506369693,32768,1760339364,1760339364,40
2025-10-13 09:09:38+0200,gamma07,2669741076,540524544,484.317389429616,32768,1760339364,1760339364,40
2025-10-13 09:09:38+0200,casa01,601825988,536903680,2190.8321439487095,1024,1760339971,1760339341,43
2025-10-13 09:09:38+0200,casa01,3742944944,536952832,476.465715848136,1024,1760339971,1760339341,40

`mqbtt_echo`¶

This commands provides a sort of fake miner that can be used to test the communication of the optimizer; for every request it returns the following reply

Reply(
  num_shares=(
    request.timestamp_min * 1000000
    + request.timestamp_max * 10000
    + request.nonce_start * 100
    + request.nonce_size
  ),
  new_block=request.timestamp_min % 2 == 0,
  kind=Kind(request.timestamp_min % 4),
)

Its usage is

Usage: mqbtt_echo [OPTIONS]

Options:
  --version      Show version and exit.
  --broker TEXT  MQTT host, default: parsifal.law.di.unimi.it
  --pid TEXT     Pair identifier
  --help         Show this message and exit.

`mqbtt_mock_optimizer`¶

Symmetrically to mqbtt_echo’s fake miner, this command provides a fake optimizer that negotiates nTime/nonce ranges with a real miner running Method B mining strategy (CONFIG_MQBTT_STEFANO_STRATEGY turned on in esp-miner-qbt-claude) - useful to exercise that firmware path without a real optimizer available.

For every reply received, it grows the nTime range by a fixed NTIME_DELTA plus however many shares the miner just found, resetting the range back to NTIME_MIN_INCREMENT whenever it exceeds NTIME_MAX_INCREMENT or the miner reports a new block; the nonce range alternates between two fixed profiles on every request. These are internal constants, not command-line options - edit mock_optimizer.py directly if you need different values.

Its usage is

Usage: mqbtt_mock_optimizer [OPTIONS]

Options:
  --version      Show version and exit.
  --broker TEXT  MQTT host, default: parsifal.law.di.unimi.it
  --pid TEXT     Pair identifier, default: test
  --help         Show this message and exit.

`mqbtt_o2query`¶

The command line counterpart of O2Query, writing the rows of a stream (or the result of an arbitrary query) as CSV or JSONL; its usage is

Usage: mqbtt_o2query [OPTIONS]

  Queries an OpenObserve stream, writing the rows as CSV or JSONL.

Options:
  --version                 Show version and exit.
  --instance [legacy|main]  OpenObserve instance, default: main
  --org [massimo|adam]      Organization, default: massimo
  --type [logs|metrics]     Stream type, default: logs
  --start TEXT              Start of the time range, local time as YYYY-MM-DD
                            HH:MM:SS
  --end TEXT                End of the time range, local time as YYYY-MM-DD
                            HH:MM:SS, default: now
  --hours FLOAT             Duration of the time range ending at --end
                            (alternative to --start)
  --stream TEXT             Stream to fetch (all rows, chunked and paged)
  --fields TEXT             Fields to fetch, as a SQL select list, default: *
  --where TEXT              SQL condition to fetch, without WHERE
  --sql TEXT                Arbitrary SQL query (a single request of at most
                            --size rows)
  --size INTEGER            Rows per request, default: 10000
  --count                   Print the number of matching rows and exit
  --streams                 List the streams (name and type) and exit
  --datetime                Replace _timestamp with a local ISO timestamp
  --format [csv|jsonl]      Output format, default: csv
  --output FILENAME         Output file, default: stdout
  --help                    Show this message and exit.

For example, to save a day of shares of a miner in a CSV file with local timestamps:

mqbtt_o2query --start "2026-09-01 00:00:00" --end "2026-09-02 00:00:00" \
  --stream shares_detail --fields "_timestamp, miner, nonce, diff, pool_diff" \
  --where "miner = 'gamma01'" --datetime --output shares.csv

`mqbtt_upgrade`¶

This is experimental, if run it should upgrade the library to the latest version. Needs a working DNS and network connection. Its usage is

Usage: mqbtt_upgrade [OPTIONS]

Options:
  --version  Show version and exit.
  --help     Show this message and exit.

`mqbtt_restart`¶

This is another experimental command that can be used to restart an AxeOS-based miner and log the restart event to OpenObserve. Its usage is

Usage: mqbtt_restart [OPTIONS]

Options:
  --version      Show version and exit.
  --pid TEXT     Pair identifier, default: test
  --detail TEXT  The detail of the reason for which the restart is requested
  --help         Show this message and exit.

`mqbtt_wslog`¶

In case you want to receive and store the full length log messages that are routed via websocket you can you the mqbtt_wslog command, its usage is

❯ mqbtt_wslog --help
Usage: mqbtt_wslog [OPTIONS]

Options:
  --version    Show version and exit.
  --pid TEXT   Pair identifier, default: test
  --file PATH  Path to the file where to store the logs.  [required]
  --help       Show this message and exit.

Please not that logs are no more transmitted via MQTT once a websocket connection is opened, so use this command judiciously (during its usage Open Observe will not receive any log messages).

The example code¶

The `logger` example¶

An example of logger is provided in the examples/logger.py script. It connects to the broker on parsifal.law.di.unimi.it using the pair id of test and echos every log message it receives.

If the testing Bitaxe is running, running

python examples/logger.py

should produce an output similar to

I (78745803) bm1368Module: Job ID: 00, Core: 1/1, Ver: 05942000
I (78745803) asic_result: Ver: 25942000 Nonce DE740102 diff 19283.4 of 4096.
I (78745813) stratum_api: tx: {"id": 2291, "method": "mining.submit", "params": ["bc1qnp980s5fpp8l94p5cvttmtdqy8rvrq74qly2yrfmzkdsntqzlc5qkc4rkq.bitaxe", "29163cc", "13000000", "67dd7a88", "de740102", "05942000"]}
I (78745973) stratum_task: rx: {"id":2291,"error":null,"result":true}
I (78745983) stratum_task: message result accepted
I (78757453) bm1368Module: Job ID: 28, Core: 40/9, Ver: 08352000
I (78757463) asic_result: Ver: 28352000 Nonce 7E540150 diff 542.9 of 4096.
I (78759513) bm1368Module: Job ID: 20, Core: 75/12, Ver: 00A18000
I (78759523) asic_result: Ver: 20A18000 Nonce D1BE0096 diff 322.8 of 4096.
I (78761213) bm1368Module: Job ID: 68, Core: 46/6, Ver: 0416C000
I (78761223) asic_result: Ver: 2416C000 Nonce 9CD0015C diff 995.4 of 4096.

The `optimizer` example¶

An example of optimizer is provided in the examples/optimizer.py script. It connects to the broker on parsifal.law.di.unimi.it using the pair id of test; it then sends ten requests to the miner

Request(
  timestamp_min=i, timestamp_max=i + 1, 
  nonce_start = i//2, nonce_size = i // 2 + 1, 
  reset=(i % 2 == 0)
)

for i from 0 to 9, and prints for the replies.

The `echo` fake miner¶

Testing the examples/optimizer.py script with the echo miner requires running

mqbtt_echo paid_ir

and then

python examples/optimizer.py pair_id

the output of the optimizer should be

Sent request: Request(timestamp_min=0, timestamp_max=1, nonce_start=0, nonce_size=1, reset=True)
Received reply: Reply(kind=<Kind.HELLO: 0>, num_shares=10001, new_block=True)
Sent request: Request(timestamp_min=1, timestamp_max=2, nonce_start=0, nonce_size=1, reset=False)
Received reply: Reply(kind=<Kind.RESULT: 1>, num_shares=1020001, new_block=False)
Sent request: Request(timestamp_min=2, timestamp_max=3, nonce_start=1, nonce_size=2, reset=True)
Received reply: Reply(kind=<Kind.RESET: 2>, num_shares=2030102, new_block=True)
Sent request: Request(timestamp_min=3, timestamp_max=4, nonce_start=1, nonce_size=2, reset=False)
Received reply: Reply(kind=<Kind.BYE: 3>, num_shares=3040102, new_block=False)
Sent request: Request(timestamp_min=4, timestamp_max=5, nonce_start=2, nonce_size=3, reset=True)
Received reply: Reply(kind=<Kind.HELLO: 0>, num_shares=4050203, new_block=True)
Sent request: Request(timestamp_min=5, timestamp_max=6, nonce_start=2, nonce_size=3, reset=False)
Received reply: Reply(kind=<Kind.RESULT: 1>, num_shares=5060203, new_block=False)
Sent request: Request(timestamp_min=6, timestamp_max=7, nonce_start=3, nonce_size=4, reset=True)
Received reply: Reply(kind=<Kind.RESET: 2>, num_shares=6070304, new_block=True)
Sent request: Request(timestamp_min=7, timestamp_max=8, nonce_start=3, nonce_size=4, reset=False)
Received reply: Reply(kind=<Kind.BYE: 3>, num_shares=7080304, new_block=False)
Sent request: Request(timestamp_min=8, timestamp_max=9, nonce_start=4, nonce_size=5, reset=True)
Received reply: Reply(kind=<Kind.HELLO: 0>, num_shares=8090405, new_block=True)
Sent request: Request(timestamp_min=9, timestamp_max=10, nonce_start=4, nonce_size=5, reset=False)
Received reply: Reply(kind=<Kind.RESULT: 1>, num_shares=9100405, new_block=False)

while the output of the mqbtt_echo should be

Received request: Request(timestamp_min=0, timestamp_max=1, nonce_start=0, nonce_size=1, reset=True)
Sent reply: Reply(kind=<Kind.HELLO: 0>, num_shares=10001, new_block=True)
Received request: Request(timestamp_min=1, timestamp_max=2, nonce_start=0, nonce_size=1, reset=False)
Sent reply: Reply(kind=<Kind.RESULT: 1>, num_shares=1020001, new_block=False)
Received request: Request(timestamp_min=2, timestamp_max=3, nonce_start=1, nonce_size=2, reset=True)
Sent reply: Reply(kind=<Kind.RESET: 2>, num_shares=2030102, new_block=True)
Received request: Request(timestamp_min=3, timestamp_max=4, nonce_start=1, nonce_size=2, reset=False)
Sent reply: Reply(kind=<Kind.BYE: 3>, num_shares=3040102, new_block=False)
Received request: Request(timestamp_min=4, timestamp_max=5, nonce_start=2, nonce_size=3, reset=True)
Sent reply: Reply(kind=<Kind.HELLO: 0>, num_shares=4050203, new_block=True)
Received request: Request(timestamp_min=5, timestamp_max=6, nonce_start=2, nonce_size=3, reset=False)
Sent reply: Reply(kind=<Kind.RESULT: 1>, num_shares=5060203, new_block=False)
Received request: Request(timestamp_min=6, timestamp_max=7, nonce_start=3, nonce_size=4, reset=True)
Sent reply: Reply(kind=<Kind.RESET: 2>, num_shares=6070304, new_block=True)
Received request: Request(timestamp_min=7, timestamp_max=8, nonce_start=3, nonce_size=4, reset=False)
Sent reply: Reply(kind=<Kind.BYE: 3>, num_shares=7080304, new_block=False)
Received request: Request(timestamp_min=8, timestamp_max=9, nonce_start=4, nonce_size=5, reset=True)
Sent reply: Reply(kind=<Kind.HELLO: 0>, num_shares=8090405, new_block=True)
Received request: Request(timestamp_min=9, timestamp_max=10, nonce_start=4, nonce_size=5, reset=False)
Sent reply: Reply(kind=<Kind.RESULT: 1>, num_shares=9100405, new_block=False)

The Python MQbtT library¶

Usage of the library¶

Messages format¶

The MQbtTOptimizer¶

The MQbtTLogReceiver¶

The MQbtTDebugReceiver¶

The MQbtTMiner¶

The MQbtTDatabase class¶

External services APIs¶

Querying OpenObserve¶

Command line tools¶

mqbtt_dump¶

mqbtt_record¶

Recovering the logs¶

Converting the recorded shares to CSV¶

mqbtt_echo¶

mqbtt_mock_optimizer¶

mqbtt_o2query¶

mqbtt_upgrade¶

mqbtt_restart¶

mqbtt_wslog¶

The example code¶

The logger example¶

The optimizer example¶

The echo fake miner¶

The `MQbtTOptimizer`¶

The `MQbtTLogReceiver`¶

The `MQbtTDebugReceiver`¶

The `MQbtTMiner`¶

The `MQbtTDatabase` class¶

`mqbtt_dump`¶

`mqbtt_record`¶

`mqbtt_echo`¶

`mqbtt_mock_optimizer`¶

`mqbtt_o2query`¶

`mqbtt_upgrade`¶

`mqbtt_restart`¶

`mqbtt_wslog`¶

The `logger` example¶

The `optimizer` example¶

The `echo` fake miner¶