I am a radio enthusiast and also an E&C engineer. Though fields and waves, or antenna theory, were not my favorite subjects in engineering, I love radio waves and listening to them. I think if I had more practice-oriented classes during my engineering studies, I would have loved those subjects much more than I do now that I practice. May be something for me to remember when I teach or share.

DXing, taken from DX, the telegraphic shorthand for “distance” or “distant”, is the hobby of receiving and identifying distant radio or television signals, or making two-way radio contact with distant stations in amateur radio, citizens band radio or other two-way radio communications. – Wikipedia

If you are new to Short Wave Listening (SWL) or the HAM radio world, DXing generally means listening to distant signals. Still, I am not limiting myself to only the “distant” signals in this project, nor to radio or television signals. I am going to try local signals as well, for example, Non-Directional Beacon (NDB) from Bangalore Airport is a fair game. I might also do local FM stations, just to learn. The focus is on learning and trying new things.

My plan is not to spend more than 30 minutes a day on this. I should be able to achieve it. I have already started logging them at #30DaysOfDXing, along with all the details. If it interests you, contact me. Maybe we can do it together.

The platform we (APU) use is Moodle. I have been in the MOOC industry for a decade now, and I have heard of Moodle so much, but this is the first time I have used it to run a course. To run a programming course, you will need an easy programming environment to challenge students. In my previous cases, we have used an NSJail-based environment with CourseBuilder (now called Seek), which works really well. But in this case, for Moodle, it’s CodeRunner plugin. It seems fairly easy to use. That said, Bash is not supported out of the box as user language. So I had to use Python to make it possible. This also assumes the environment (CodeRunner/JOBE) has Bash installed, though not directly accessible through API as user language.

import subprocess

script = """{{ TEST.testcode | e('py') }}""" + '\n' + """{{ STUDENT_ANSWER | e('py') }}""" + '\n' + """{{ TEST.extra | e('py') }}"""
input = """{{ TEST.stdin | e('py') }}"""

with open('__prog__.sh', 'w') as outfile:
    outfile.write(script)

result = subprocess.run(['/bin/bash', '__prog__.sh'], capture_output=True, text=True, input=input, timeout=5)
stdout = result.stdout.strip()
print(stdout)

The template Python code takes the TEST.testcode and prepends it to the user-entered STUDENT_ANSWER code. It also takes TEST.extra code and appends it. Then, it runs it as a bash script using Python subprocess by passing TEST.stdin as input. Captures STDOUT and prints it for comparison.

Example Question 1: Read an input as score. If the score is greater than or equal to 40, then print P. If the score is less than 40, then print U.

Example Question 2: Write a function called cube. If a number is passed, it returns the cube of that number. For example cube 5 # will return 25

We did about three in-class labs using this. It worked well for all our cases. Though I must say we tried only Bash basics. Maybe there are use cases where this might break, but I think for most of it, this should work.

I do have a kind of Walkie-talkie in the form of Helmet communications. There are some issues with using it as a general communications device. Some that matter to me are

• It’s not open; it’s proprietary. The only part of that ecosystem that is open is Sena’s Universal Intercom protocol, which uses Bluetooth. It comes with its own limitations.
• It’s made for helmet communication. Even if you plan to reuse it, it’s challenging to use in other situations due to its form factor.
• It uses 2.5 GHz (the same frequency range as Bluetooth, Wi-Fi, etc.), so its range is somewhat limited. In our tests, it was around 700-900m direct line of sight. Some support a proprietary mesh network to increase the range.

Requirements

With all this, I was left to look for a walkie-talkie.

• Is the license free in India? Ideally, across the world, but at least in India.
• Open protocols
• Decent range
• Multi purpose
• Easy to use, it shouldn’t take more than five minutes to learn its functionality
• Easy to manage, charge, and rugged. I plan to use it everywhere.
• Widely recognized brand and model. So I don’t get stopped at the airport and other places while I am carrying it.
• Supported in India
• True walkie-talkie, not PTT over the cell network.

License-free and open

There are two bands used for public, license-free communication without encryption. The Family Radio Service (FRS) is used in the USA and Canada. FRS operates between 462.5625 MHz and 467.7125 MHz. And PMR446 (private mobile radio used in most of Europe and India.

PMR446 is a private mobile radio that operates between 446.0 – 446.2 MHz in India. It’s a license-free band, as per the Gazette Notification G.S.R. 1047(E). Since I am dealing with India, I will focus on it. The rules limit output power to 0.5 W. It doesn’t seem like much, but you can reach a kilometer or more in urban conditions with buildings and trees. A 5KM to 10KM in direct line of sight conditions. It’s much better than my helmet communication system. The rules allow channel spacing of 6.25 kHz (Digital) and 12.5 kHz (analog). The frequency range can accommodate 16 analog or 32 digital channels. I have a dedicated page about the PMR446 band, channels, channel guarding, etc. The most important thing to know is that we need to select walkie-talkies that operate PMR446 in India.

I have shown the snippet from the Gazette and given link to the rules for your reference. Please get your own legal advice. I am not a lawyer, and this is not legal advice.

The protocols are standardized, and most follow the ETSI Harmonised European Standard. Since this is something all vendors use, if someone wants to build, the protocol is not a secret sauce. This also makes the technology and walkies vendor-neutral.

Privacy

There is no encryption on PMR446. So you shouldn’t communicate anything that needs encryption on these bands. There is no privacy. There are ways to keep a set of people independent by using a specific channel and Squelch (CTSS or DCS). Squelch is called PL (Private Line) tone, CG (Channel Guard) tone, or QC (Quiet Channel) tone, depending on the vendor, but all refer to the same functionality. Basically, you use a channel and a specific CTSS or DCS code to keep your conversations independent from others. This setup doesn’t stop others from using the same combo and listening to you.

Motorola T82 Extreme

Armed with this information and requirement, I started my search, and my final list was

• Motorola Talkabout T82 Extreme – Manual
• Kenwood TK 3501
• Wavex PT100
• Vertel Team Talkie Radio
• Sanchar G3U
• Baofeng GT-68 PMR Walkie Talkie

If I had money, I would have tried each of them and then decided which to go with. Based on the information I found online, I went with the T82 by Motorola. If you have any other model listed here, let me know, and we can test them together.

In terms of cost, it wasn’t cheap; a pair of them cost me INR 18,000. It came in a package with two walkies, a carrying case, a charger, NIMH rechargeable batteries, earpieces, belt clips, etc. It’s probably the most expensive pair. G3U or PT100 is half the price.

Things I like

1. Brand recognition. Everyone knows Motorola. It also looks colorful, playful, rugged, and harmless. The support in India is good, and the community around it is also good. I have had other Motorola phones (pre-smartphone era cell phones), and they have been very rugged and have had very positive experiences in general.
2. It’s rugged. It can be used in conditions where others can’t be used. I won’t be scared to put it on a motorcycle handlebar or hang it outside my backpack. I am also not worried about water splashing on it (IPX4), though I wouldn’t submerge it. Also, not worried about accidentally dropping them.
3. A replaceable, rechargeable, 800 mAh NIMH battery powers it. It can be recharged using a micro USB charger. That said, you can replace the NIMH battery with 3 x AA Alkaline batteries in an emergency. The promised battery life is around 18 hours. Even if it’s just half, I will be happy. One can also upgrade to a 1300 mAh NiMH battery if required.
4. It comes with a headset with a boom mic. That makes it usable inside the helmet if you want, or use it hands-free. It has VOX/iVOX (Internal Voice Operated / Voice Operated Transmission ), which lets you transmit without pressing a button. It has three levels of sensitivity for voice activation. By default, it is in PTT (Push To Talk) mode, where you need to press and hold a button to talk. In VOX mode, sound activates the transmission. So all you do is speak, and the radio will transmit for you. This is especially useful when you are riding or doing manual work.
5. It supports 8 PMR Channels. User expandable to 16 Channels in countries where it’s allowed by government authorities. And 121 Sub-Codes (38 CTCSS Codes & 83 DCS codes). So, there are plenty of options to isolate your group from others.
6. It’s very easy to use. Manuals and settings are very easy. It probably takes less than 5 minutes to teach someone to start using it effectively.
7. It has an Emergency Alert Mode with a dedicated button that can signal members of your group for help.
8. It has dual-channel monitoring. It lets you listen to two channels and engage with the primary one.
9. There are other features, such as a flashlight, roger Tone, channel monitoring and scanning, easy pairing, etc.
10. The range has been good. I was able to get at least 1 KM in urban settings with buildings. I continue to do more range tests. I will write a separate post about it.
11. As such, there is no limit to the number of folks you can have on the same channel. This is true for most Walkie-Talkies.

Things to improve

1. Micro USB. It should have been USB-C even if it was just 5 watts.
2. Price. It is expensive. I am looking to try Wavex PT100 and Vertel, which are half the price. If you have, let me know. We can compare.

Conclusion

Should you get it? Maybe. It depends on your need. But once you start using it, you will see how useful it becomes. Especially if you ride or drive a lot in teams, or have a farm or work in a place with poor or no network, trek, or run events, etc. As far as me, I am still going to use BluArmor while riding. But I will surely carry it along with me.

It’s also a good entry point into HAM radio. It’s good enough to generate interest among kids and adults about analog communications.

For now, I will add them to my site, Archive.org, and YouTube. They are all CC-BY-SA.

I am thinking of uploading just one version of each video. No server side transcoding. Hence, I want to convert the videos into the most optimized format and size. I have been doing some experiments and seem to have found decent settings. Its 1080p or 720p with 24 frames.

If its mobile screencast then I will pad them with color, so it can be a proper FHD or HD.

I am going to use WebM as the container with AV1 as video codec and Opus as audio codec. Most screencasts could be just 720p.

Introduction

I knew Dask supports this through a scheduler and worker model. But I had not tried this. So I did, and it works pretty well. But you need to take care of some additional stuff.

1. The client, scheduler, and workers should all use the same version of Python and the same version of Python packages. So it’s better to pin the package and the Python version. I have used Podman/Docker to run the scheduler and the workers. So I have tied down the image version and hence the versions of Python, Dask, Distributed, and their dependencies. I have used the identical versions for my client. See the header files in the client – distributed_dask.py
2. The workers won’t have additional packages. You can either create a new image of the worker with your required packages, or you can dynamically install them using the PipInstall plugin. This plugin takes an array of Python packages and installs them on all workers as they start. I have also pinned these packages Easy way to ship packages to workers.
3. I have created a small plugin here called EnvPlugin. This runs on all the workers. Essentially, it copies all the ENV variables from the worker.env file to the client or worker environment. That way, it’s easy to ship that as well.
4. cloudpickle is used to serialize the code to ship it from client to workers
5. Some of the other pinned packages lkelz4, msgpack etc are used by dask and sitributed.

Client Code

This is the code you probably run from your machine. That connects to scheduler, send code and jobs so it can get the work done.

# /// script
# requires-python = "==3.10.12"
# dependencies = [
#   "dask==2024.8.2",
#   "distributed==2024.8.2",
#   "toolz==0.12.0",
#   "lz4==4.3.3",
#   "tornado==6.4.1",
#   "cloudpickle==3.0.0",
#   "msgpack==1.0.8",
#   "python-dotenv==1.1.1",
#   "requests==2.32.5"
# ]
# ///
import os
import json
import logging
import requests
import csv
from dotenv import dotenv_values

from dask import bag as db
from dask.distributed import (
    Client,
    LocalCluster,
    WorkerPlugin,
    PipInstall,
)

logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")


# ---- EnvPlugin plugin -------------------------------------------------

class EnvPlugin(WorkerPlugin):
    def __init__(self, env):
        self.env = env

    def setup(self, worker):
        logging.warning("EnvPlugin loaded with keys: %s", list(self.env.keys()))
        os.environ.update(self.env)


URL = "https://httpbin.org/post"
HEADERS = {"accept": "application/json"}

# -- Acutal thing that runs on workers -----------------------------------------
def post_row(row: dict) -> dict:
    """Send one row as JSON payload"""
    API_KEY_FOR_HTTP_BIN = os.environ.get("API_KEY_FOR_HTTP_BIN")
    HEADERS["x-api-key"] = API_KEY_FOR_HTTP_BIN
    print(API_KEY_FOR_HTTP_BIN)
    r = requests.post(URL, json=row, headers=HEADERS, timeout=20)
    r.raise_for_status()
    return r.json()


def main(csv_path: str):
    with open(csv_path, newline="", encoding="utf-8") as f:
        records = list(csv.DictReader(f))

    env_vars = dotenv_values("worker.env")
    print("==========env_vars for workers=============")
    print(env_vars)
    os.environ.update(env_vars)

    scheduler_address = os.environ.get("DASK_SCHEDULER_ADDRESS")

    logging.info(f"Connecting to remote scheduler at {scheduler_address}")
    client = Client(scheduler_address)

    # Setup env on the workers
    client.register_plugin(
        EnvPlugin(env_vars), name="env"
    )

    # Install packages on the workers
    plugin = PipInstall(
        packages=[
            "python-dotenv==1.1.1",
            "requests==2.32.5",
        ],
        pip_options=["--upgrade"],
    )
    client.register_plugin(plugin)

    b = db.from_sequence(records, partition_size=10)
    # Send to workers
    results = b.map(post_row).compute()
    client.close()

    # Print gathered results
    print(results)
    logging.info(f"Completed {len(results)} requests")


if __name__ == "__main__":
    import sys
    if len(sys.argv) != 2:
        print("uv run distributed_dask.py data.csv")
        raise SystemExit(1)

    main(sys.argv[1])

Setup and Run

I am using Podman/Docker to run the scheduler and workers. It makes things easy and keeps the environment clean. I have used the Docker images provided by the project. But in your case, you might want to build them and harden them. Especially the workers, as they will run whatever the client sends. So you will have to be careful if you are running a worker.

Create a network and then run the scheduler in that network ( you could run as part of host too). It starts the scheduler and starts listening for workers to join on port 8786. And exposes a dashboard on another port 8787. That you can access by going to http://localhost:8787

podman network create dask

# start the scheduler
podman run --network dask \
 -p 8787:8787 -p 8686:8686 \
 --name scheduler \
 ghcr.io/dask/dask:2024.8.2 dask-scheduler 

Now you can create as many workers as you want and make them join the scheduler. Here I have two.

# start the worker 1
podman run -d \
  --name dask-worker-1 \
  --network dask \
  ghcr.io/dask/dask:2024.8.2 \
  dask-worker scheduler:8786 \
  --nworkers 1 \
  --nthreads 2 \
  --worker-port 9000:9009

# start the worker 2
podman run -d \
  --name dask-worker-2 \
  --network dask \
  ghcr.io/dask/dask:2024.8.2 \
  dask-worker scheduler:8786 \
  --nworkers 1 \
  --nthreads 2 \
  --worker-port 9000:9009

Clients use two ports. One 8786 port on the scheduler to notify them of joining. And another random port so the scheduler can talk to them. It usually auto-picks. But you can also specify a range, so it assigns one of the available ones in that range. So give some additional ports. If there are 8 workers then give a range of 9 or more. This also helps you recognize the open ports easily. You can check the scheduler logs to see if the workers have joined. You can also check the web dashboard to do the same.

Now just submit the jobs by calling our client script.

export DASK_SCHEDULER_ADDRESS=tcp://127.0.0.1:8786
uv run distributed_dask.py data.csv

That’s it. Now the client should be able to connect to the scheduler, send code and setup instructions to workers, then send jobs to workers, collect all the responses, and print all of them.

Usage

The example I have shown here seems trivial. But I built a CPU and network heave pipeline for processing I used this logic to parse Bihar SIR data for The Hindu (WN-35/2025) in a really short time. It was a CPU and network-intensive job. I could use the combined power of all my laptops to run this and get the job done so that they could run the story. I felt like I was running a supercomputer at home.

Also note what I have not shown in the example client code but it shouldn’t be difficult to add.

1. Error handling. I have not done any here. Any error will break it
2. Retry mechanism
3. There is no real or persistent message queue here. So, at some point, if the scheduler goes down, you will lose work.
4. Hardening the workers. Anyway, as a worker, you shouldn’t join a scheduler you don’t trust, as they can send any kind of code to execute on the worker.

The songs are mainly from the playlists “Road Trip Across the World” and “Road Trip Across Karnataka“.

Day 1

Day 2

Day 3

The script is simple and straightforward.

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.9"
# dependencies = [
#   "gpxpy",
#   "requests",
# ]
# ///

import sys
import json
import os
from datetime import datetime, timezone
from bisect import bisect_left

import gpxpy
import requests


LB_ROOT = "https://api.listenbrainz.org"
USERNAME = "thejeshgn"
MAX_GAP_SECONDS = 600   # 10 min window


def load_gpx_points(path):
    with open(path, "r", encoding="utf-8") as f:
        gpx = gpxpy.parse(f)

    pts = []
    for tr in gpx.tracks:
        for seg in tr.segments:
            for p in seg.points:
                if not p.time:
                    continue

                t = p.time
                if t.tzinfo is None:
                    t = t.replace(tzinfo=timezone.utc)
                else:
                    t = t.astimezone(timezone.utc)

                pts.append(
                    {
                        "time": t,
                        "lat": p.latitude,
                        "lon": p.longitude,
                        "ele": p.elevation,
                    }
                )

    pts.sort(key=lambda x: x["time"])
    return pts


def build_thumbnail_and_track_urls(
    release_mbid: str | None,
    cover_art_url: str | None,
    recording_mbid: str | None,
    recording_msid: str | None,
):
    """
    Build thumbnail and ListenBrainz URL.

    Priority:
    - URL based on release_mbid: https://listenbrainz.org/release/<release_mbid>/
    - Fallback URL based on recording id: https://listenbrainz.org/track/<recording_*id>/
    - Thumbnail from cover_art_url, else Cover Art Archive with release_mbid
    """

    # URL: prefer release URL
    track_url = None
    rec_id = recording_mbid or recording_msid
    if rec_id:
        track_url = f"https://listenbrainz.org/track/{rec_id}/"

    # Thumbnail / cover art
    thumb = cover_art_url
    if not thumb and release_mbid:
        thumb = f"https://coverartarchive.org/release/{release_mbid}/front-250"

    return thumb, track_url


def fetch_listens(min_ts):
    url = f"{LB_ROOT}/1/user/{USERNAME}/listens"
    params = {
        "min_ts": int(min_ts),
        "count": 10000,
    }

    headers = {}
    token = os.getenv("LISTENBRAINZ_TOKEN")
    if token:
        headers["Authorization"] = f"Token {token}"

    r = requests.get(url, params=params, headers=headers, timeout=30)
    r.raise_for_status()

    listens_raw = r.json().get("payload", {}).get("listens", [])
    listens = []

    for it in listens_raw:
        ts = it.get("listened_at")
        if not ts:
            continue

        t = datetime.fromtimestamp(ts, tz=timezone.utc)
        md = it.get("track_metadata", {}) or {}
        add = md.get("mbid_mapping", {}) or {}

        release_mbid = add.get("release_mbid")
        recording_mbid = add.get("recording_mbid")
        artist_mbids = add.get("artist_mbids")
        recording_msid = add.get("recording_msid") or it.get("recording_msid")
        cover_art_url = add.get("cover_art_url")

        thumbnail_url, track_url = build_thumbnail_and_track_urls(
            release_mbid=release_mbid,
            cover_art_url=cover_art_url,
            recording_mbid=recording_mbid,
            recording_msid=recording_msid,
        )

        listens.append(
            {
                "time": t,
                "track_name": md.get("track_name"),
                "artist_name": md.get("artist_name"),
                "release_name": md.get("release_name"),
                "release_mbid": release_mbid,
                "recording_mbid": recording_mbid,
                "recording_msid": recording_msid,
                "artist_mbids": artist_mbids,
                "thumbnail_url": thumbnail_url,
                "track_url": track_url,
            }
        )

    listens.sort(key=lambda x: x["time"])
    return listens


def match(points, listens):
    """
    Match GPX points to listens, but only emit one point when the song changes.

    Logic:
    - Compress consecutive listens with the same "song key"
      (recording_mbid / recording_msid / track+artist).
    - For each such listen, pick the closest GPX point in time.
    """

    if not points or not listens:
        return []

    # Prebuild list of point times for bisect
    point_times = [p["time"] for p in points]

    def song_key(l):
        return (
            l.get("recording_mbid"),
            l.get("recording_msid"),
            l.get("track_name"),
            l.get("artist_name"),
        )

    # Keep only listens where the song actually changes
    filtered_listens = []
    last_key = None
    for l in listens:
        k = song_key(l)
        if k != last_key:
            filtered_listens.append(l)
            last_key = k

    out = []

    from bisect import bisect_left

    for l in filtered_listens:
        idx = bisect_left(point_times, l["time"])
        candidates = []
        if idx > 0:
            candidates.append(points[idx - 1])
        if idx < len(points):
            candidates.append(points[idx])

        if not candidates:
            continue

        best_p = min(
            candidates,
            key=lambda p: abs((p["time"] - l["time"]).total_seconds()),
        )

        dt = abs((best_p["time"] - l["time"]).total_seconds())
        if dt <= MAX_GAP_SECONDS:
            out.append((best_p, l))

    return out


def clean_props(d: dict) -> dict:
    """Return a copy without None values."""
    return {k: v for k, v in d.items() if v is not None}


def to_geojson(pairs):
    feats = []

    for p, l in pairs:
        base_props = {
            "gpx_time": p["time"].isoformat(),
            "gpx_elevation": p["ele"],
            "track_name": l.get("track_name"),
            "artist_name": l.get("artist_name"),
            "release_name": l.get("release_name"),
            "listened_at": l["time"].isoformat(),
            "source": "ListenBrainz",
            # important IDs
            "release_mbid": l.get("release_mbid"),
            # optional IDs
            "recording_mbid": l.get("recording_mbid"),
            "recording_msid": l.get("recording_msid"),
            "artist_mbids": l.get("artist_mbids"),
            # links and media
            "thumbnail_url": l.get("thumbnail_url"),
            "track_url": l.get("track_url"),
        }

        props = clean_props(base_props)

        feats.append(
            {
                "type": "Feature",
                "geometry": {
                    "type": "Point",
                    "coordinates": [p["lon"], p["lat"]],
                },
                "properties": props,
            }
        )

    return {"type": "FeatureCollection", "features": feats}


def main():
    if len(sys.argv) != 3:
        print("Usage: uv run gpx_lb.py <track.gpx> <out.geojson>")
        sys.exit(1)

    gpx_path = sys.argv[1]
    out_path = sys.argv[2]

    pts = load_gpx_points(gpx_path)
    if not pts:
        print("No GPX points with timestamps.")
        sys.exit(1)

    start_ts = int(pts[0]["time"].timestamp())
    listens = fetch_listens(start_ts - MAX_GAP_SECONDS)

    if not listens:
        print("No listens returned from ListenBrainz for that period.")
        sys.exit(0)

    pairs = match(pts, listens)
    gj = to_geojson(pairs)

    with open(out_path, "w", encoding="utf-8") as f:
        json.dump(gj, f, ensure_ascii=False, indent=2)

    print(f"Created {len(gj['features'])} features → {out_path}")


if __name__ == "__main__":
    main()

Loved the talks by David MacMillan (chemistry 2021) and James Robinson (economic sciences 2024). Both were inspiring. There were other talks too, but these two were the stars of the day. There was also a pleasant surprise for the audience in the form of Kavita Krishnamurthy and L Subramaniam performing to kick off the day.

Quote of the day “Ask yourself, Do you think it’s cool”.

{
  "queue_name": "",     // The name of the queue the message is being pushed on
  "actor_name": "",         // The name of the actor that should handle this message
  "args": [],              // A list of positional arguments that are passed to the actor
  "kwargs": {},                // A dictionary of keyword arguments that are passed to the actor
  "options": {},               // Arbitrary options that are used by middleware. Leave this empty
  "message_id": "unique-id",   // A UUID4 value representing the message's unique id in the system
  "message_timestamp": 0,      // The UNIX timestamp in milliseconds representing when the message was first enqueued
}

This feature enables us to call the Job from any programming language that can communicate with the Broker and format a JSON payload. Let’s examine an example Actor from our previous blog post, called send_welcome_email. which takes two arguments, user_id and msg in that order. Now, let’s create a payload to call send_welcome_email

{
  "queue_name": "default",
  "actor_name": "send_welcome_email",
  "args": [
    1234,
    "Message for email send to redis directly"
  ],
  "kwargs": {},
  "options": {
    "redis_message_id": "f3bcdcb4-1e18-41fa-9190-bf34d77a8fbe"
  },
  "message_id": "425564d6-f762-455f-b8dd-69c9da1dd8fd",
  "message_timestamp": 1762113802478
}

In our example, we are using Redis as the Broker. The default queue is called default. We also need to send redis_message_id, as part of options, that is a different UUID4 value than message_id. You also need to make sure it’s unique. We are using a generated UUIDs here. Additionally, you can see that the args contain the arguments passed to the Actor. The remaining attributes are self-explanatory.

Now we need to post it to the Redis broker, to a default namespace called dramatiq. Dramatiq stores the message in a hash (HSET) and then pushes (RPUSH) redis_message_id of the message to a queue. The hash key is dramatiq:default.msgs and the filed is the redis_message_id. For the queue the key is dramatiq:default.

Then get to redict-cli. I am running redict using docker/podman, I will just shell into it

 podman  exec -it redict sh

Then call the command on db 1, as that is the redis DB I am using.

# PART 1 - Add message to hash
redict-cli -n 1 HSET dramatiq:default.msgs f3bcdcb4-1e18-41fa-9190-bf34d77a8fbe '{
  "queue_name": "default",
  "actor_name": "send_welcome_email",
  "args": [
    1234,
    "Message for email send to redis directly"
  ],
  "kwargs": {},
  "options": {
    "redis_message_id": "f3bcdcb4-1e18-41fa-9190-bf34d77a8fbe"
  },
  "message_id": "425564d6-f762-455f-b8dd-69c9da1dd8fd",
  "message_timestamp": 1762113802478
}'

# PART 2 - Push the id to queue
redict-cli -n 1 RPUSH dramatiq:default f3bcdcb4-1e18-41fa-9190-bf34d77a8fbe

That’s it. That should run the Job on one of the workers. Below, I have written a simple Python script that posts a similar message directly to Redis.

# /// script
# requires-python = ">=3.10"
# dependencies = [
#   "redis"
# ]
# ///
"""
Usage:
  uv run send_tasks_directly.py
"""
import redis, json, uuid, time

# connect to redis
r = redis.Redis(host='localhost', port=6379, db=1)
namespace = "dramatiq"
queue_name = "default"
redis_message_id = str(uuid.uuid4())
msg = {
  "queue_name": queue_name,
  "actor_name": "send_welcome_email",
  "args": [1234, "Message for email send to redis directly"],
  "kwargs": {},
  "options": {"redis_message_id": redis_message_id},
  "message_id": str(uuid.uuid4()),
  "message_timestamp": int(time.time() * 1000)
}
payload = json.dumps(msg, separators=(",", ":"))

r.hset(f"{namespace}:{queue_name}.msgs", redis_message_id, payload)
r.rpush(f"{namespace}:{queue_name}", redis_message_id)

I can do the same from Rust (run using rust-script) as well.

//! ```cargo
//! [dependencies]
//! redis = "0.25"
//! serde = { version = "1", features = ["derive"] }
//! serde_json = "1"
//! uuid = { version = "1", features = ["v4"] }
//! chrono = "0.4"
//! ```

// Run using
// rust-script send_tasks_directly.rs

use chrono::Utc;
use redis::Commands;
use redis::ConnectionLike;
use serde::Serialize;
use uuid::Uuid;

#[derive(Serialize)]
struct Message {
    queue_name: String,
    actor_name: String,
    args: Vec<serde_json::Value>,
    kwargs: serde_json::Value,
    options: serde_json::Value,
    message_id: String,
    message_timestamp: i64,
}
fn main() -> redis::RedisResult<()> {
    let client = redis::Client::open("redis://127.0.0.1:6379/1")?;
    let mut con = client.get_connection()?;

    let namespace = "dramatiq";
    let queue_name = "default";
    let redis_message_id = Uuid::new_v4().to_string();
    println!("redis_message_id '{}'", redis_message_id);

    let msg = Message {
        queue_name: queue_name.into(),
        actor_name: "send_welcome_email".into(),
        args: vec![
            serde_json::json!(1234),
            serde_json::json!("Message for email send to redis directly from Rust"),
        ],
        kwargs: serde_json::json!({}),
        options: serde_json::json!({ "redis_message_id": redis_message_id }),
        message_id: Uuid::new_v4().to_string(),
        message_timestamp: Utc::now().timestamp_millis(),
    };

    let payload = serde_json::to_string_pretty(&msg).unwrap();
    println!("payload '{}'", payload);

    let hash_key = format!("{}:{}", namespace, queue_name.to_string() + ".msgs");
    let list_key = format!("{}:{}", namespace, queue_name);

    con.hset::<_, _, _, ()>(&hash_key, &redis_message_id, payload)?;
    println!("Sent message to hash_key '{}'", hash_key);

    let stored_value: String = con.hget(&hash_key, &redis_message_id)?;
    println!("🔍 Retrieved from Redis:\n{}", stored_value);

    con.req_packed_command(&redis::cmd("PING").get_packed_command())?;
    con.rpush::<_, _, ()>(&list_key, &redis_message_id)?;

    println!("Sent message to Redis queue list_key '{}'", list_key);
    Ok(())
}

Isn’t that nifty?

Question: Why redis_message_id when there is already message_id at the Dramatiq level?

We are manually adding redis_message_id here because we are performing the same work that RedisBroker does internally. If we use the standard APIs, then it’s done internally automatically. Internally, redis_message_id is generated as a unique ID every time a message is retried, whereas the message_id remains the same. So if a message is tried only once, there would be a single message_id and redis_message_id. But if it’s retried five times, then there will be a single message_id and five different redis_message_id.