Skip to main content

​
Overview

These are complete, production-ready plugins from the Sakuya AC codebase. Study these examples to learn plugin development patterns and best practices.

​
Chat Filter Plugin

This plugin moderates chat messages to make them safe for children by filtering profanity and inappropriate content. 
"""
Enables moderation of words in chat to make it safe for children
"""

from lib.PacketManager.packets import FSNETCMD_TEXTMESSAGE
import re

# To disable moderation completely, set ENABLED = False
# To allow some profanity, set ABOVE_13 = True

ENABLED = True
ABOVE_13 = False

class Plugin:
    def __init__(self):
        self.plugin_manager = None
        if ABOVE_13:
            # Moderate less for teens/adults - only severe language
            moderation_string = r'\b(b[i!]tch|cunt|whore|slut|bastard|nude|naked|porn|penis|vagina|boobs|dick|hate|murder|die|n[i!]gg?(er|a|as|ers)|f[a@]gg?ot?|kike|chink|retard|racist|racism|rape|rapist|molest|https?:\/\/\S+|www\.\S+)\b'
        else:
            # Stricter moderation for children
            moderation_string = r'\b(f[u\*]ck(?:er|ing)?|sh[i!]t(?:ty|head)?|b[i!]tch(?:es|y)?|cunt|ass(?:hole|wipe|hat)?|' \
                                r'whore|slut(?:ty)?|damn(?:it)?|hell|piss(?:ed)?|bastard|nude|naked|' \
                                r'sex(?:ual)?|porn(?:ography)?|penis|vagina|boobs?|tits?|titties|dick|' \
                                r'cock|pussy|boner|' \
                                r'horny|jerk(?:\s*off)?|masturbat(?:e|ion)|cum(?:ming)?|' \
                                r'hate|kill|murder|die|stupid|idiot|dumb(?:ass)?|moron|' \
                                r'n[i!]gg?(?:er|a|as|ers)|f[a@]gg?ot?|kike|chink|spic|wetback|' \
                                r'retard(?:ed)?|racist|racism|nazi|' \
                                r'rape(?:d|s)?|rapist|molest(?:er|ed)?|pedophile|' \
                                r'bl[o0][w0]job|handjob|' \
                                r'wtf|stfu|gtfo|lmfao|' \
                                r'https?:\/\/\S+|www\.\S+)\b'

        self.filter_regex = re.compile(moderation_string, re.IGNORECASE)

    def register(self, plugin_manager):
        self.plugin_manager = plugin_manager
        self.plugin_manager.register_hook('on_chat', self.on_chat)

    @staticmethod
    def censor_match_dynamic_length(match_obj):
        """
        This function is called by re.sub for each match.
        It returns a string of '#' characters matching the length of the found word.
        """
        matched_word = match_obj.group(0)
        return '#' * len(matched_word)

    def on_chat(self, data, player, message_to_client, message_to_server):
        decode = FSNETCMD_TEXTMESSAGE(data)
        msg = decode.raw_message

        # Check if the message contains filtered content
        censored_text = self.filter_regex.sub(self.censor_match_dynamic_length, msg)

        if censored_text != msg:
            # Create a new message with censored content
            message = FSNETCMD_TEXTMESSAGE.encode(censored_text, with_size=True)
            message_to_server.append(message)
            return False
        return True
Key features:
  • Uses regex to match inappropriate words
  • Configurable strictness with ABOVE_13 flag
  • Replaces filtered words with # characters
  • Returns False to block original message and sends censored version

​
Over-G Damage Plugin

This plugin damages aircraft that exceed G-force limits, adding realism to flight mechanics. 
"""This plugin will cause damage to the aircraft if
it exceeds the g-force limit set in the config file."""
from lib.PacketManager.packets import FSNETCMD_GETDAMAGE, FSNETCMD_TEXTMESSAGE
from config import G_LIM
import time

# CONFIG
ENABLED = True
INTERVAL = 0.3  # Interval in seconds before G Limit check is enforced

class Plugin:
    def __init__(self):
        self.plugin_manager = None

    def register(self, plugin_manager):
        self.plugin_manager = plugin_manager
        self.plugin_manager.register_hook('on_flight_data', self.on_receive)

    def on_receive(self, data, player, messages_to_client, *args):
        if ENABLED:
            if abs(player.aircraft.last_packet.g_value) > G_LIM:
                if time.time() - player.aircraft.last_over_g_message > INTERVAL:
                    player.aircraft.last_over_g_message = time.time()
                    
                    # Send damage packet
                    damage_packet = FSNETCMD_GETDAMAGE.encode(
                        player.aircraft.id,
                        1, 1,
                        player.aircraft.id,
                        1, 11, 0, True
                    )
                    
                    # Send warning message
                    warning_message = FSNETCMD_TEXTMESSAGE.encode(
                        f"You are exceeding the G Limit! gValue = {player.aircraft.last_packet.g_value}!",
                        True
                    )
                    
                    messages_to_client.append(damage_packet)
                    messages_to_client.append(warning_message)
        return True
Key features:
  • Monitors on_flight_data hook continuously
  • Uses timing to prevent spam
  • Sends both damage packet and warning message
  • Returns True to forward original flight data

​
Refuel Plugin

Enables air-to-air refueling between players with matching IFF. 
"""
This plugin enables you to perform air to air refueling.
"""
from ast import alias
import math
from lib.PacketManager.packets import FSNETCMD_AIRCMD, FSNETCMD_AIRPLANESTATE
from lib import YSchat
import asyncio

ENABLED = True
REFUEL_RADIUS = 500  # in meters
FUEL_RATE = 50  # in kg/s

refuelers = {}  # Players offering fuel
refueling = {}  # Players receiving fuel

class Plugin:
    def __init__(self):
        self.plugin_manager = None

    def register(self, plugin_manager):
        self.plugin_manager = plugin_manager
        self.plugin_manager.register_command('refuel', self.refuel, 
                                             "Allows you to get refueled", 
                                             alias="r")
        self.plugin_manager.register_command('refueler', self.refueler, 
                                             "Allows other players to refuel from you", 
                                             alias="rf")
        self.plugin_manager.register_hook('on_flight_data', self.on_flight_data)
        self.plugin_manager.register_hook('on_unjoin', self.on_unjoin)

    def refueler(self, full_message, player, message_to_client, message_to_server):
        if player in refueling:
            msg = YSchat.message("You cannot be refueled while being a refueler")
        elif player in refuelers:
            msg = YSchat.message("You are no more refueling!")
            refuelers.pop(player)
        else:
            msg = YSchat.message(f"You are now refueling other players! Within {REFUEL_RADIUS}m and same IFF")
            refuelers[player] = [0, [], False]
        message_to_client.append(msg)

    def refuel(self, full_message, player, message_to_client, message_to_server):
        if player in refuelers:
            msg = YSchat.message("You cannot be a refueler while being refueled")
        elif player in refueling:
            msg = YSchat.message("You are no more refueling!")
            refueling.pop(player)
        else:
            msg = YSchat.message(f"You can now be refueled! Within {REFUEL_RADIUS}m and same IFF")
            refueling[player] = [0, []]
        message_to_client.append(msg)

    def on_flight_data(self, data, player, message_to_client, message_to_server):
        asyncio.create_task(self.refuel_logic(data, player))
        return True

    def on_unjoin(self, data, player, message_to_client, message_to_server):
        if player in refuelers:
            refuelers.pop(player)
        elif player in refueling:
            refueling.pop(player)
        return True

    @staticmethod
    def in_range(refueler_cord:list, refueled_cord:list):
        dist = math.sqrt(
            (refueler_cord[0]-refueled_cord[0])**2 + 
            (refueler_cord[1]-refueled_cord[1])**2 + 
            (refueler_cord[2]-refueled_cord[2])**2
        )
        return dist < REFUEL_RADIUS

    async def refuel_logic(self, data, player):
        decode = FSNETCMD_AIRPLANESTATE(data)
        k = player.streamWriterObject
        
        if player in refueling:
            for refueler in refuelers:
                if refueler.iff == player.iff:
                    refueling[player] = [decode.fuel, decode.position]
                    
                    if self.in_range(refuelers[refueler][1], decode.position):
                        refuelers[refueler][2] = True
                        max_lim = int((player.aircraft.initial_config['WEIGFUEL'])[:-2])
                        
                        if decode.fuel < max_lim:
                            k.write(FSNETCMD_AIRCMD.set_command(
                                player.aircraft.id, 
                                "INITFUEL", 
                                f"{decode.fuel+FUEL_RATE}kg", 
                                True
                            ))
                        elif decode.fuel >= max_lim:
                            msg = YSchat.message("The plane is full!")
                            refuelers[refueler][2] = False
                            k.write(msg)

                        if refuelers[refueler][0] < FUEL_RATE:
                            msg = YSchat.message("The refueler ran out of fuel!")
                            k.write(msg)
                    else:
                        refuelers[refueler][2] = False

        elif player in refuelers:
            refuelers[player] = [decode.fuel, decode.position, refuelers[player][2]]
            if decode.fuel-FUEL_RATE > 0 and refuelers[player][2]:
                fuel_cmd = FSNETCMD_AIRCMD.set_command(
                    player.aircraft.id, 
                    "INITFUEL", 
                    f"{decode.fuel-FUEL_RATE}kg", 
                    True
                )
                k.write(fuel_cmd)
Key features:
  • Uses commands to toggle refueling modes
  • Tracks player positions and fuel levels
  • Uses async task for refueling logic
  • Validates IFF matching and distance
  • Drains fuel from refueler while transferring

​
Radar Plugin

Implements fog-of-war radar that hides enemy aircraft beyond a certain range. 
"""
This plugin allows you to have radar features similar to
original WW3 server.
"""
import math
from lib.PacketManager.packets import FSNETCMD_AIRPLANESTATE
from lib.YSchat import send
import struct
import traceback

ENABLED = True
RADIUS = 9000  # in meters, detection range for different IFF

flying_players = {}

class Plugin:
    def __init__(self):
        self.plugin_manager = None

    def register(self, plugin_manager):
        self.plugin_manager = plugin_manager
        self.plugin_manager.register_hook('on_flight_data_server', self.on_flight_data_server)
        self.plugin_manager.register_hook('on_flight_data', self.on_flight_data)
        self.plugin_manager.register_hook('on_unjoin', self.on_unjoin)

    def on_flight_data(self, data, player, message_to_client, message_to_server):
        try:
            flying_players[player.aircraft.id][1] = FSNETCMD_AIRPLANESTATE(data).position
        except KeyError:
            return True
        return True

    def on_flight_data_server(self, data, player, message_to_client, message_to_server):
        try:
            if player.aircraft.id not in flying_players and player.aircraft.id != -1:
                flying_players[player.aircraft.id] = [player, []]
                return True
            elif player.aircraft.id == -1:
                return True

            decode = FSNETCMD_AIRPLANESTATE(data)

            if len(flying_players[player.aircraft.id][1]) == 0:
                return True
            elif decode.player_id == player.aircraft.id:
                return True
            elif flying_players[decode.player_id][0].iff == player.iff:
                return True
            elif self.in_range(flying_players[player.aircraft.id][1], decode.position):
                return True
            else:
                # Move enemy aircraft to arbitrary far away position
                position_data = struct.pack("3f", 10e8, 200, 10e8)
                if decode.packet_version == 4 or 5:
                    updated_data = data[:14] + position_data + data[26:]
                else:
                    updated_data = data[:16] + position_data + data[28:]

                player.streamWriterObject.write(send(updated_data))
                return False  # Block original packet
        except Exception as e:
            if isinstance(e, KeyError):
                return True
            traceback.print_exc()

    def on_unjoin(self, data, player, message_to_client, message_to_server):
        if player.aircraft.id in flying_players:
            flying_players.pop(player.aircraft.id)
        return True

    @staticmethod
    def in_range(pos1, pos2):
        dist = math.sqrt(
            (pos1[0] - pos2[0])**2 + 
            (pos1[1] - pos2[1])**2 + 
            (pos1[2] - pos2[2])**2
        )
        return dist <= RADIUS
Key features:
  • Hooks on_flight_data_server to intercept server broadcasts
  • Tracks all player positions
  • Modifies enemy positions in packets to hide them
  • Uses struct.pack to modify binary packet data
  • Returns False and writes modified packet directly

​
Weather Control Plugin

Allows players to control server weather through chat commands. 
"""This plugin will set the weather based on a chat message."""
from lib.PacketManager.packets import (
    FSNETCMD_SKYCOLOR, 
    FSNETCMD_FOGCOLOR, 
    FSNETCMD_TEXTMESSAGE, 
    FSNETCMD_ENVIRONMENT
)

ENABLED = True

class Plugin:
    def __init__(self):
        self.initialWeather = None
        self.plugin_manager = None

    def register(self, plugin_manager):
        self.plugin_manager = plugin_manager
        self.plugin_manager.register_hook('on_environment_server', self.on_environment)
        self.plugin_manager.register_command('fog', self.fog, 
                                             "Sets fog color. Usage: /fog r,g,b")
        self.plugin_manager.register_command('sky', self.sky, 
                                             "Sets sky color. Usage: /sky r,g,b")
        self.plugin_manager.register_command('time', self.time, 
                                             "Sets time. Usage: /time <day|night>")
        self.plugin_manager.register_command('visibility', self.visibility, 
                                             "Sets visibility: /vis <meters>", 
                                             "vis")

    def fog(self, full_message, player, message_to_client, message_to_server):
        try:
            args = list(map(int, full_message[4:].split(',')))
        except:
            message_to_client.append(
                FSNETCMD_TEXTMESSAGE.encode("Invalid fog argument, usage /fog r,g,b", True)
            )
            return
        fog_packet = FSNETCMD_FOGCOLOR.encode(args[0], args[1], args[2], True)
        message_to_client.append(fog_packet)

    def sky(self, full_message, player, message_to_client, message_to_server):
        try:
            args = list(map(int, full_message[4:].split(',')))
        except Exception as e:
            message_to_client.append(
                FSNETCMD_TEXTMESSAGE.encode("Invalid sky argument, usage /sky r,g,b", True)
            )
            return True
        sky_packet = FSNETCMD_SKYCOLOR.encode(args[0], args[1], args[2], True)
        message_to_client.append(sky_packet)
        return True

    def time(self, full_message, player, message_to_client, message_to_server):
        requested = full_message[6:].lower()
        if requested == 'day' and self.initialWeather:
            packet = FSNETCMD_ENVIRONMENT.set_time(self.initialWeather.buffer, False, True)
            self.initialWeather = FSNETCMD_ENVIRONMENT(packet[4:])
            message_to_client.append(packet)
        elif requested == 'night' and self.initialWeather:
            packet = FSNETCMD_ENVIRONMENT.set_time(self.initialWeather.buffer, True, True)
            self.initialWeather = FSNETCMD_ENVIRONMENT(packet[4:])
            message_to_client.append(packet)
        else:
            message_to_client.append(
                FSNETCMD_TEXTMESSAGE.encode(
                    "Invalid time argument, usage /time night or /time day", 
                    True
                )
            )
        return True

    def visibility(self, full_message, player, message_to_client, message_to_server):
        visibility = int(full_message[4:])
        packet = FSNETCMD_ENVIRONMENT.set_visibility(
            self.initialWeather.buffer, 
            visibility, 
            True
        )
        self.initialWeather = FSNETCMD_ENVIRONMENT(packet[4:])
        message_to_client.append(packet)
        return True

    def on_environment(self, data, *args):
        """Store the initial weather packet for the server."""
        self.initialWeather = FSNETCMD_ENVIRONMENT(data)
        return True
Key features:
  • Captures initial weather state on server connection
  • Multiple commands for different weather aspects
  • Validates command arguments with error messages
  • Modifies stored weather state for consistency

​
Crash on Ground Plugin

Prevents aircraft from despawning mid-air when killed, forcing them to crash into the ground. 
# This plugin renders the dead plane until it crashes in ground
# Thus disabling mid air despawning when dead
from lib.PacketManager.packets import (
    FSNETCMD_REMOVEAIRPLANE, 
    FSNETCMD_READBACK, 
    FSNETCMD_GETDAMAGE
)

ENABLED = True

class Plugin:
    def __init__(self):
        self.plugin_manager = None

    def register(self, plugin_manager):
        self.plugin_manager = plugin_manager
        self.plugin_manager.register_hook('on_remove_airplane_server', self.on_death)

    def on_death(self, data, player, message_to_client, message_to_server):
        if player.aircraft.id != -1:
            if player.aircraft.id == FSNETCMD_REMOVEAIRPLANE(data).object_id:
                return True
            else:
                decode = FSNETCMD_REMOVEAIRPLANE(data)
                
                # Send massive damage to kill the aircraft
                damage = FSNETCMD_GETDAMAGE.encode(
                    decode.object_id, 1, 1,
                    player.aircraft.id, 10000,
                    11, 0, True
                )
                message_to_client.append(damage)
                message_to_server.append(damage)
                
                # Request readback
                readback = FSNETCMD_READBACK.encode(2, player.aircraft.id, True)
                message_to_server.append(readback)
                
                return False  # Block removal packet
        else:
            return True
Key features:
  • Hooks on_remove_airplane_server
  • Replaces removal with massive damage
  • Blocks original removal packet
  • Forces aircraft to fall and crash naturally

​
Next Steps

Plugin Hooks

Learn about all available hooks

Chat Commands

Create custom chat commands

Plugin Structure

Understand plugin file structure

Packet Reference

YSFlight packet documentation

Build docs developers (and LLMs) love

Get started for freeTalk to us