Simple Server
This commit is contained in:
Kyler
commit
cccc606212
|
|
@ -0,0 +1,225 @@
|
|||
# Yeahbut Apr 2024
|
||||
import certifi, os
|
||||
os.environ["SSL_CERT_FILE"] = certifi.where()
|
||||
|
||||
# Modified from https://raw.githubusercontent.com/barneygale/quarry/master/examples/server_chat_room_advanced.py
|
||||
|
||||
from typing import List
|
||||
|
||||
from twisted.internet import reactor
|
||||
from quarry.net.server import ServerFactory, ServerProtocol
|
||||
from quarry.types.uuid import UUID
|
||||
from quarry.data.data_packs import data_packs, dimension_types
|
||||
|
||||
|
||||
class YTDServerProtocol(ServerProtocol):
|
||||
|
||||
def player_joined(self):
|
||||
# Call super. This switches us to "play" mode, marks the player as
|
||||
# in-game, and does some logging.
|
||||
ServerProtocol.player_joined(self)
|
||||
|
||||
# Send join game packet
|
||||
self.factory.send_join_game(self)
|
||||
|
||||
# Send "Player Position and Look" packet
|
||||
self.send_packet(
|
||||
"player_position_and_look",
|
||||
self.buff_type.pack("dddff?",
|
||||
0, # x
|
||||
500, # y Must be >= build height to pass the "Loading Terrain" screen on 1.18.2
|
||||
0, # z
|
||||
0, # yaw
|
||||
0, # pitch
|
||||
0b00000), # flags
|
||||
self.buff_type.pack_varint(0), # teleport id
|
||||
self.buff_type.pack("?", True)) # Leave vehicle
|
||||
|
||||
# Start sending "Keep Alive" packets
|
||||
self.ticker.add_loop(20, self.update_keep_alive)
|
||||
|
||||
# Announce player join to other players
|
||||
self.factory.broadcast_player_join(self)
|
||||
|
||||
# Send full player list
|
||||
self.factory.send_player_list_add(self, self.factory.players)
|
||||
|
||||
def player_left(self):
|
||||
ServerProtocol.player_left(self)
|
||||
|
||||
# Announce player leave to other players
|
||||
self.factory.broadcast_player_leave(self)
|
||||
|
||||
def update_keep_alive(self):
|
||||
# Send a "Keep Alive" packet
|
||||
self.send_packet("keep_alive", self.buff_type.pack('Q', 0))
|
||||
|
||||
def packet_chat_message(self, buff):
|
||||
if self.protocol_mode != 'play':
|
||||
return
|
||||
|
||||
message = buff.unpack_string()
|
||||
|
||||
self.factory.broadcast_chat(message, self.uuid, self.display_name)
|
||||
|
||||
buff.discard()
|
||||
|
||||
def send_chunk(self, x, z, full, heightmap, sections, biomes):
|
||||
sections_data = self.bt.pack_chunk(sections)
|
||||
self.send_packet(
|
||||
'chunk_data',
|
||||
self.bt.pack('ii?', x, z, full),
|
||||
self.bt.pack_chunk_bitmask(sections),
|
||||
self.bt.pack_nbt(heightmap),
|
||||
self.bt.pack_array('I', biomes) if full else b'',
|
||||
self.bt.pack_varint(len(sections_data)),
|
||||
sections_data,
|
||||
self.bt.pack_varint(0),
|
||||
b''.join(self.bt.pack_nbt(entity) for entity in ()))
|
||||
|
||||
|
||||
class YTDServerFactory(ServerFactory):
|
||||
protocol = YTDServerProtocol
|
||||
motd = "YTD Custom Server (WIP)"
|
||||
force_protocol_version = 758
|
||||
|
||||
def send_join_game(self, player):
|
||||
# Build up fields for "Join Game" packet
|
||||
entity_id = 0
|
||||
max_players = 0
|
||||
hashed_seed = 42
|
||||
view_distance = 2
|
||||
simulation_distance = 2
|
||||
game_mode = 3
|
||||
prev_game_mode = 3
|
||||
is_hardcore = False
|
||||
is_respawn_screen = True
|
||||
is_reduced_debug = False
|
||||
is_debug = False
|
||||
is_flat = False
|
||||
|
||||
dimension_codec = data_packs[player.protocol_version]
|
||||
dimension_name = "minecraft:overworld"
|
||||
dimension_tag = dimension_types[player.protocol_version, dimension_name]
|
||||
world_count = 1
|
||||
world_name = "chat"
|
||||
|
||||
join_game = [
|
||||
player.buff_type.pack("i?Bb", entity_id, is_hardcore, game_mode, prev_game_mode),
|
||||
player.buff_type.pack_varint(world_count),
|
||||
player.buff_type.pack_string(world_name),
|
||||
player.buff_type.pack_nbt(dimension_codec),
|
||||
]
|
||||
|
||||
join_game.append(player.buff_type.pack_nbt(dimension_tag))
|
||||
|
||||
join_game.append(player.buff_type.pack_string(world_name))
|
||||
join_game.append(player.buff_type.pack("q", hashed_seed))
|
||||
join_game.append(player.buff_type.pack_varint(max_players))
|
||||
join_game.append(player.buff_type.pack_varint(view_distance)),
|
||||
|
||||
join_game.append(player.buff_type.pack_varint(simulation_distance))
|
||||
|
||||
join_game.append(player.buff_type.pack("????", is_reduced_debug, is_respawn_screen, is_debug, is_flat))
|
||||
|
||||
# Send "Join Game" packet
|
||||
player.send_packet("join_game", *join_game)
|
||||
|
||||
# Sends an unsigned chat message, using system messages on supporting clients
|
||||
def broadcast_chat(self, message: str, sender: UUID, sender_name: str):
|
||||
for player in self.players:
|
||||
if player.protocol_mode != 'play':
|
||||
continue
|
||||
|
||||
self.send_chat(player, message, sender, sender_name)
|
||||
|
||||
def send_chat(self, player: YTDServerProtocol, message: str, sender: UUID, sender_name: str):
|
||||
player.send_packet("chat_message",
|
||||
player.buff_type.pack_chat("<%s> %s" % (sender_name, message)),
|
||||
player.buff_type.pack('B', 0),
|
||||
player.buff_type.pack_uuid(sender))
|
||||
|
||||
# Sends a system message, falling back to chat messages on older clients
|
||||
def broadcast_system(self, message: str):
|
||||
for player in self.players:
|
||||
if player.protocol_mode != 'play':
|
||||
continue
|
||||
|
||||
self.send_system(player, message)
|
||||
|
||||
@staticmethod
|
||||
def send_system(player: YTDServerProtocol, message: str):
|
||||
player.send_packet("chat_message",
|
||||
player.buff_type.pack_chat(message),
|
||||
player.buff_type.pack('B', 0),
|
||||
player.buff_type.pack_uuid(UUID(int=0)))
|
||||
|
||||
# Announces player join
|
||||
def broadcast_player_join(self, joined: YTDServerProtocol):
|
||||
self.broadcast_system("\u00a7e%s has joined." % joined.display_name)
|
||||
self.broadcast_player_list_add(joined)
|
||||
|
||||
# Announces player leave
|
||||
def broadcast_player_leave(self, left: YTDServerProtocol):
|
||||
self.broadcast_system("\u00a7e%s has left." % left.display_name)
|
||||
self.broadcast_player_list_remove(left)
|
||||
|
||||
# Sends player list entry for new player to other players
|
||||
def broadcast_player_list_add(self, added: YTDServerProtocol):
|
||||
for player in self.players:
|
||||
# Exclude the added player, they will be sent the full player list separately
|
||||
if player.protocol_mode == 'play' and player != added:
|
||||
self.send_player_list_add(player, [added])
|
||||
|
||||
@staticmethod
|
||||
def send_player_list_add(player: YTDServerProtocol, added: List[YTDServerProtocol]):
|
||||
data = [
|
||||
player.buff_type.pack_varint(0), # Action - 0 = Player add
|
||||
player.buff_type.pack_varint(len(added)), # Player entry count
|
||||
]
|
||||
|
||||
for entry in added:
|
||||
if entry.protocol_mode != 'play':
|
||||
continue
|
||||
|
||||
data.append(player.buff_type.pack_uuid(entry.uuid)) # Player UUID
|
||||
data.append(player.buff_type.pack_string(entry.display_name)) # Player name
|
||||
data.append(player.buff_type.pack_varint(0)) # Empty properties list
|
||||
data.append(player.buff_type.pack_varint(3)) # Gamemode
|
||||
data.append(player.buff_type.pack_varint(0)) # Latency
|
||||
data.append(player.buff_type.pack('?', False)) # No display name
|
||||
|
||||
player.send_packet('player_list_item', *data)
|
||||
|
||||
# Sends player list update for leaving player to other players
|
||||
def broadcast_player_list_remove(self, removed: YTDServerProtocol):
|
||||
for player in self.players:
|
||||
if player.protocol_mode == 'play' and player != removed:
|
||||
player.send_packet('player_list_item',
|
||||
player.buff_type.pack_varint(4), # Action - 4 = Player remove
|
||||
player.buff_type.pack_varint(1), # Player entry count
|
||||
player.buff_type.pack_uuid(removed.uuid)) # Player UUID
|
||||
|
||||
|
||||
def main(argv):
|
||||
# Parse options
|
||||
import argparse
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("-a", "--host", default="", help="address to listen on")
|
||||
parser.add_argument("-p", "--port", default=25565, type=int, help="port to listen on")
|
||||
parser.add_argument("--offline", action="store_true", help="offline server")
|
||||
args = parser.parse_args(argv)
|
||||
|
||||
# Create factory
|
||||
factory = YTDServerFactory()
|
||||
|
||||
factory.online_mode = False#not args.offline
|
||||
|
||||
# Listen
|
||||
factory.listen(args.host, args.port)
|
||||
reactor.run()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import sys
|
||||
main(sys.argv[1:])
|
||||
Loading…
Reference in New Issue