Minecraft-bridge/mc_wrapper.py

149 lines
5.2 KiB
Python
Raw Normal View History

2021-06-27 19:19:05 +00:00
from subprocess import Popen, PIPE
from typing import Union
import asyncio
import json
import sys
import re
from nbsr import NonBlockingStreamReader as NBSR
import config
2021-06-27 19:19:05 +00:00
# run the shell as a subprocess:
p = Popen(sys.argv[1:],
stdin = PIPE, stdout = PIPE, stderr = PIPE, shell = False)
# wrap p.stdout with a NonBlockingStreamReader object:
nbsr = NBSR(p.stdout)
async def start(client, mc_channel):
2023-10-01 12:07:13 +00:00
"""
Start reading from the Minecraft subprocess.
"""
2021-06-27 19:19:05 +00:00
await asyncio.sleep(3)
while True:
output = nbsr.readline(0.1)
# 0.1 secs to let the shell output the result
if not output:
await asyncio.sleep(1)
else:
try:
sentence = output.decode("utf-8").strip()
except UnicodeDecodeError:
print("Could not decode sentence:")
print(output)
else:
print(sentence)
plain_text, sentence = process_message(sentence)
if sentence is not None:
print("[Matrix] " + plain_text)
# Send terminal message to Matrix
await client.room_send(
room_id=mc_channel,
message_type="m.room.message",
content = {
"msgtype": "m.text",
"body": plain_text,
"format": "org.matrix.custom.html",
"formatted_body": sentence
}
)
server_live = False
def process_message(sentence : str) -> Union[str, None]:
2023-10-01 12:07:13 +00:00
"""
Process a message that is sent to stdout in the Minecraft terminal.
If this function deems it relevant, it returns a string that can be
sent to Matrix.
"""
2021-06-27 19:19:05 +00:00
global server_live
if (match := re.fullmatch(
r"\[[\d:]+\] \[Server thread\/INFO\]: Preparing level \"(.+)\""
, sentence)):
level, = match.groups()
return f"Preparing level {level}...", "<strong>Preparing level \"" + level + "\"...</strong>"
if re.fullmatch(
r"\[[\d:]+\] \[Server thread\/INFO\]: Done \(\d+.?\d*s\)! For help, type \"help\"",
sentence):
server_live = True
return f"The Minecraft server is live. The server is reacable at <code>{config.SERVER_IP}</code>.", f"The minecraft server is live. The server is reacable at <code>{config.SERVER_IP}</code>."
2021-06-27 19:19:05 +00:00
if re.fullmatch(
r"\[[\d:]+\] \[Server thread\/INFO\]: Stopping server",
sentence):
return "The server has stopped.", "<strong>The server has stopped.</strong>"
if not server_live:
return None, None
if (match := re.fullmatch(
r"\[[\d:]+\] \[Server thread\/INFO\]: ([A-Za-z0-9_]{3,16}) joined the game",
sentence)):
username, = match.groups()
return username + " joined the Minecraft server", (
"<strong>" + username + " joined the Minecraft server</strong>")
if (match := re.fullmatch(
r"\[[\d:]+\] \[Server thread\/INFO\]: ([A-Za-z0-9_]{3,16}) left the game",
sentence)):
username, = match.groups()
return username + " left the Minecraft server", (
2023-10-01 12:07:13 +00:00
"<strong>" + username + " left the Minecraft server</strong>"
)
2021-06-27 19:19:05 +00:00
if (match := re.fullmatch(
r"\[[\d:]+\] \[Server thread\/INFO\]: <([A-Za-z0-9_]{3,16})> (.+)",
sentence)):
username, message = match.groups()
return username + ": " + message, (
"<strong>" + username + "</strong>: " + message
)
if (match := re.fullmatch(
r"\[[\d:]+\] \[Server thread\/INFO\]: ([A-Za-z0-9_]{3,16}) ([\w\[\]\-\. !?,]+)",
sentence)):
message = " ".join(match.groups())
return message, "<strong>" + message + "</strong>"
return None, None
2023-10-01 12:07:13 +00:00
def reply_to_mc(message : str, display_name : str, sender : str):
2021-06-27 19:19:05 +00:00
"""
Send something back to the Minecraft terminal.
"""
2023-10-01 12:07:13 +00:00
if sender in config.MATRIX_ADMINS and message.startswith('!'):
2021-06-27 19:19:05 +00:00
p.stdin.write((message[1:] + "\r\n").encode())
else:
p.stdin.write(
2023-10-01 12:07:13 +00:00
("execute as @a run tellraw @s " + format(sender, display_name, message) + "\r\n").encode()
2021-06-27 19:19:05 +00:00
)
p.stdin.flush()
2023-10-01 12:07:13 +00:00
def format(sender : str, display_name : str, message : str) -> str:
"""
Create a string used to format the user's message.
"""
start = [ "", dict(text="M", color="red" ) ]
end = [ dict(text=f" <{display_name}> {message}") ]
options = config.at(['matrix', 'alternative_platforms']) or {}
for platform, details in options.items():
try:
regex = details['match']
text = details['text']
color = details['color']
except KeyError:
print("WARNING: Platform `" + platform + "` is missing some configurations.")
else:
if re.fullmatch(regex, sender):
start.append(dict(text=text, color=color))
return json.dumps(start + end)
2021-06-27 19:19:05 +00:00
if __name__ == '__main__':
asyncio.run(start())