Made a python script to create posts via meshtastic messages. If you can see this, the test worked! Posting from the park with no mobile data _
It took 3 walks to the park due to bugs and unhandled exceptions but it worked!
The python script connects to a node that is left at home. The bot can detect and process a message with the following format:
PST Com: Meshtastic Sub: Posting via Meshtastic Bod: Made a python script to create posts via meshtastic messages. If you can see this, the test worked! Posting from the park with no mobile data ^_^
It will parse the message and create a post. So, as long as I can reach my home’s node I am able to create a post.
Nice! When you get a chance, I would love to see that code! We have a system right now that does local weather and events once a day.
For sure. It is quite basic and I am not proud of the hacky method I used to “parse” the message, but it might be useful for someone looking for a simple way to interface with a meshtastic device over TCP (onReceive) and the Lemmy API (createPost).
import json import re import meshtastic import meshtastic.tcp_interface from pubsub import pub import time import os import requests INSTANCE_API = "https://mander.xyz/api/v3" MESHTEST_LEMMY_JWT = 'Cookie retrieved from browser' def createPost(community_name, subject, body): url = f"{INSTANCE_API}/post" getCommunity = requests.get(f"{INSTANCE_API}/community?name={community_name}").json() communityId = getCommunity['community_view']['community']['id'] data = { "community_id": communityId, "name": subject, "body": body} headers = { "Authorization": f"Bearer {MESHTEST_LEMMY_JWT}", "Content-Type": "application/json" } response = requests.post(url, headers=headers, json=data) return response MESHTASTIC_NODE_IP = "Local IP of the base node connected to WiFi" def sstrip(text): return re.sub(r'(?:\s|,)*(sub|SUB|Sub|COM|com|Com|Bod|bod|BOD)(?:\s|,)*$', '', text.strip()) def processMessage(message): blocks = message.split(':') # Splits the message into blocks, but will also split smiley faces ":)", bad method. try: for i in range(0,len(blocks)-1): if blocks[i][-3:].lower() == 'sub': subject = sstrip(blocks[i+1]) if blocks[i][-3:].lower() == 'com': community_name = sstrip(blocks[i+1]).lower() if blocks[i][-3:].lower() == 'bod': body = sstrip(blocks[i+1]) return community_name, subject, body except: return 'ERR','ERR','ERR' def onReceive(packet, interface): if 'decoded' in packet and 'payload' in packet['decoded']: try: message = packet['decoded']['payload'].decode('utf-8') sender = packet['from'] if 'Ping' in message: interface.sendText("Pong!", destinationId=sender) if message.split('\n')[0] == 'PST': try: community_name, subject, body = processMessage(message) response = createPost(community_name, subject, body) if response.ok: interface.sendText("Post created succesfully!", destinationId=sender) else: interface.sendText("Unable to create post!", destinationId=sender) except: interface.sendText("Exception triggered", destinationId=sender) except Exception as e: pass interface = meshtastic.tcp_interface.TCPInterface(hostname=MESHTASTIC_NODE_IP) pub.subscribe(onReceive, "meshtastic.receive") while True: time.sleep(2)
On a side note, I love how python looks on lemmy. This is awesome! I really like the way you parse the message, its readable.
if 'Ping' in message: interface.sendText("Pong!", destinationId=sender)
Im totally doing that to test out my node(s).
Great job on this. I may take some of this later.
Here is the weather code (fat fingered the name, cant be bothered to change): https://yuno.chrisco.me/git/michael/meshtastic_forceast/src/branch/main/main.py
Heavily influenced by something I found online a while back which I lost the reference to.
Im guessing you were a C/C# dev based on the function names.
Glad you like it :D
The ping is very useful. I know that there is a built-in range test, but sometimes I don’t need the test to be on all the time, nor do I want to set the frequency too high. Actually… This give me an idea, I can simply program a command to turn the range test off/on remotely.
That weather function is nice! The US makes available some nice weather APIs. I have a PinePhone and it has a weather module that relies on a US-based API, but I am not in the US. At least I can find out the weather in Oregon easily. I don’t know if there is some similar API in the Netherlands.
Im guessing you were a C/C# dev based on the function names.
I helped re-factor some C+±based micro-controller firmware recently and the original code was not following any convention, so I looked at a list of conventions and decided that ‘lower camel case’ looked like a nice one to pick. So, I have been developing a habit to stick to that. I do scientific r&d and only sometimes need to do a bit of programming, so I’m not sure if I’d call myself a dev!
It worked. Can you see replies too?
Wuhuu! Not yet! I do have a few ideas on how to implement that. Since the size of messages is limited to 200 characters, and since trying to transmit too much data via meshtastic wouldn’t be very efficient, I am brainstorming about how to implement notifications and fetching in a somewhat compatible manner.
One approach would be via some interface that displays the minimum amount of data (for example, first few letters of a post’s title, community, username). The user would “fetch” specific pieces of data, which then gets stored into the phone’s memory and this way one can populate the application with content without overloading the mesh.
It’s not something I am too seriously considering actually making, I am just having a bit of fun.
Where you able to interpret the MQTT messages directly from the computer? I have been trying to transform the messages that are output when using mosquitto_sub into the original protobuf packets, but I am having issues with that step.
I think messages bled to msh because someone messaged my node to call me a wanker.
Ah, might be! 😅 Were you using a public MQTT server? Or were you downlinking a lot of toots and ‘spamming’ them via RF?
@Sal my own local MQTT, but there may have been forwarding
https://gist.github.com/itomato/5a422fcd475329e61e26b49ddb8c4aca
Cool, thanks for sharing!
Where I run into problems is in the following step:
def on_mqtt_message(client, userdata, msg): try: message = msg.payload.decode("utf-8", errors="replace")
If the message is published by a meshtastic client, then my “message” looks something like:
̴=����*!�_��i��}M������jUJC�'�!���c�^5yk�=C��gE��
���������LongFast?`So, I think that the UTF-8 decoding does not work on the raw payload, and that the payload needs to be processed into a mesh.protobuf. At least that is what I understand so far.
The messages that you published via MQTT were messages sent from the Meshtastic client, or were they messages that you posted using mosquitto_pub? If this did work on packets that the meshtastic client published to MQTT then I must be overlooking something… Again, thanks for sharing!
I took some inspiration from your code and looked into how to decode the mesh.protobuf packets and managed to decode the MQTT messages! From the raw byte stream I had to remove the first two bytes of the payload and then process it as a MeshPacket to get the data in its readable form.
Once processed, the payload from MQTT looks like this:
from: 1501080443 to: 3294953195 channel: 8 encrypted: "u\003\221n\354U\373\257\006[" id: 882625294 rx_time: 1739046557 hop_limit: 3 priority: HIGH hop_start: 3
So, looking at your code, I think that what happened was that the messages were being posted to Mastodon only due to the on_receive() function. The MQTT messages would also trigger this function call when the node downlinks the MQTT message.
What is useful about triggering the messages directly from the MQTT stream is that it is then possible to create a general application that listens to one or multiple MQTT servers without the need for a node to downlink.
Here is the MeshPacket code:
import mesh_pb2 import paho.mqtt.client as mqtt # MQTT Config MQTT_BROKER = "MQTT IP" MQTT_PORT = 1883 MQTT_TOPIC = "#" def onMessage(client, userdata, msg): print(f"Received message on {msg.topic}") proto_msg = mesh_pb2.MeshPacket() print("Raw:\n\n") print(msg.payload) proto_msg.ParseFromString(msg.payload[2::]) print("\n\nParsed:\n\n") print(proto_msg) client = mqtt.Client() client.on_message = onMessage client.connect(MQTT_BROKER, MQTT_PORT, 60) client.subscribe(MQTT_TOPIC) client.loop_forever()
Interesting! Thanks, I need to continue testing/studying.