piqueserver package

Module contents


piqueserver.banpublish module

class piqueserver.banpublish.PublishResource(factory)[source]

Bases: twisted.web.resource.Resource

getChild(name, request)[source]

Retrieve a ‘child’ resource from me.

Implement this to create dynamic resource generation – resources which are always available may be registered with self.putChild().

This will not be called if the class-level variable ‘isLeaf’ is set in your subclass; instead, the ‘postpath’ attribute of the request will be left as a list of the remaining path elements.

For example, the URL /foo/bar/baz will normally be:

| site.resource.getChild('foo').getChild('bar').getChild('baz').

However, if the resource returned by ‘bar’ has isLeaf set to true, then the getChild call will never be made on it.

Parameters and return value have the same meaning and requirements as those defined by L{IResource.getChildWithDefault}.

class piqueserver.banpublish.PublishServer(protocol, port)[source]

Bases: object


piqueserver.bansubscribe module

class piqueserver.bansubscribe.BanManager(protocol)[source]

Bases: object

bans = None
fetch_filtered_bans(url: str, whitelist: List[str])[source]

piqueserver.commands module

exception piqueserver.commands.CommandError[source]

Bases: Exception

class piqueserver.commands.CommandHelp(description, usage, info)

Bases: tuple


Alias for field number 0


Alias for field number 2


Alias for field number 1

exception piqueserver.commands.PermissionDenied[source]

Bases: Exception

piqueserver.commands.add(func: Callable) → None[source]

Function to add a command from scripts. Deprecated

piqueserver.commands.add_rights(user_type: str, command_name: str) → None[source]

Give the user type a new right

>>> add_rights("knights", "say_ni")
piqueserver.commands.admin(func: Callable) → Callable[source]

Shorthand for @restrict(“admin”). Mainly exists for backwards compatibility with pyspades scripts.

>>> @admin
... @command()
... def some_command(x):
...     pass
piqueserver.commands.alias(name: str) → Callable[source]

add a new alias to a command. Deprecated

piqueserver.commands.command(name=None, *aliases, admin_only=False) → Callable[[Callable], Callable][source]

Register a new command.

The command will be accessible as /function_name unless a name or alias is specified.

>>> @command()
... def some_command(x):
...     pass

Optional names and aliases:

>>> @command("name", "alias1", "alias2")
... def some_command(x):
...     pass
piqueserver.commands.format_command_error(command_func: Callable, message: str, exception: Exception = None) → str[source]

format a help message for a given command

piqueserver.commands.get_command_help(command_func: Callable) → piqueserver.commands.CommandHelp[source]
piqueserver.commands.get_player(protocol, value: str, spectators=True)[source]

Get a player connection object by name or ID.

IDs are formatted as: “#12”

If no player with the specified name is found, it will try to find a player who’s name contains the string

piqueserver.commands.get_rights(user_type: str) → List[str][source]

Get a list of rights a specific user type has.

>>> add_rights("knights", "say_ni")
>>> get_rights("knights")
>>> get_rights("arthur")
piqueserver.commands.get_team(connection, value)[source]
piqueserver.commands.handle_command(connection, command, parameters)[source]

Public facing function to run a command, given the connection, a command name, and a list of parameters.

Will log the command.

piqueserver.commands.handle_input(connection, user_input)[source]
piqueserver.commands.has_permission(f, connection)[source]
piqueserver.commands.join_arguments(arg, default=None)[source]
piqueserver.commands.name(name: str) → Callable[source]

Give the command a new name. Deprecated

piqueserver.commands.player_only(func: Callable)[source]

This recorator restricts a command to only be runnable by players connected to the server, not via other places such as, say, the console.

>>> @command()
... @player_only
... def some_command(x):
...     pass
piqueserver.commands.restrict(*user_types) → Callable[source]

restrict the command to only be used by a specific type of user

>>> @restrict("admin", "guard")
... @command()
... def some_command(x):
...     pass
piqueserver.commands.target_player(func: Callable)[source]
This decorator converts first argument of a command to a piqueserver.FeatureConnection.
It’s intended for commands which accept single argument for target player eg. /fly [player]. It implicitly uses invoker as target if no arguments are provided. It uses first argument are player name or id for targetting. It forces non-player invokers to provide player argument.
>>> @command()
... @target_player
... def fly(connection, target):
...     target.fly = True
...     pass
piqueserver.commands.update_rights(rights: Dict[KT, VT])[source]

Update the rights of all users according to the input dictionary. This is currently only here for when the config needs to be reloaded.

>>> update_rights({"knights": ["say_ni"]})
>>> get_rights("knights")

piqueserver.config module

class piqueserver.config.ConfigStore[source]

Bases: object

Configuration store that manages global configuration.

Usage example:

>>> config = ConfigStore()
>>> config.load_from_dict({'key1': 'value1'})
>>> option1 = config.option('key1', default='nothing',
...                         validate=lambda x: len(x) > 0)
>>> section1 = config.section('section1')
>>> nested_option = section1.option('key1', default=0)
>>> print(nested_option.get())
>>> option1.set('hello')
>>> # underlying dictionary structure will look like
>>> # {
>>> #   'key1': 'hello',
>>> #   'section1': {
>>> #     'key1': 0
>>> #   }
>>> # }

Return the subset of the underlying dictionary that doesn’t have any corresponding registered options.

dump_to_file(fobj, format_='TOML')[source]

Writes the current configuration to a file-like objection, with the format specified by format_.


Load from a dictionary object directly.

load_from_file(fobj, format_='TOML')[source]

Clear the current configuration and load new configuration from a file-like object in a supported format.

option(name, default=None, cast=None, validate=None)[source]

Register and return a new option object.


Register and return a new section object.


Load from a dictionary object directly.

update_from_file(fobj, format_='TOML')[source]

Updates the configuration from a file-like object. Useful for overwriting/updating part of the config without touching the rest.

piqueserver.config.cast_duration(d) → int[source]

casts duration(1min, 1hr) into seconds. If input is an int it returns that unmodified.

piqueserver.console module

class piqueserver.console.ConsoleInput(protocol)[source]

Bases: twisted.protocols.basic.LineReceiver

admin = True
delimiter = b'\n'

Override this for when each line is received.

@param line: The line which was received with the delimiter removed. @type line: C{bytes}

name = 'Console'
send_chat(value: str, _)[source]
send_lines(lines: List[str])[source]

piqueserver.irc module

class piqueserver.irc.IRCBot[source]

Bases: twisted.words.protocols.irc.IRCClient

irc_NICK(prefix, params)[source]

Called when a user changes their nickname.


Called when I finish joining a channel.

channel has the starting character (C{‘#’}, C{’&’}, C{‘!’}, or C{‘+’}) intact.


Called when I have left a channel.

channel has the starting character (C{‘#’}, C{’&’}, C{‘!’}, or C{‘+’}) intact.

me(msg, do_filter=False)[source]
modeChanged(user, irc_channel, *arg, **kw)[source]
name = None

str(object=’’) -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to ‘strict’.

ops = None
privmsg(user, irc_channel, *arg, **kw)[source]
send(msg, do_filter=False)[source]
send_chat(value: str, _)[source]
send_lines(lines: List[str])[source]

Called after successfully signing on to the server.

unaliased_name = None
userKicked(kickee, irc_channel, kicker, message)[source]

Called when I observe someone else being kicked from a channel.

userLeft(user, irc_channel, *arg, **kw)[source]
userQuit(user, message)[source]

Called when I see another user disconnect from the network.

voices = None
class piqueserver.irc.IRCClientFactory(server, config)[source]

Bases: twisted.internet.protocol.ClientFactory

admin = None
aliases = None
bot = None

Create an instance of a subclass of Protocol.

The returned instance will handle input on an incoming server connection, and an attribute “factory” pointing to the creating factory.

Alternatively, L{None} may be returned to immediately close the new connection.

Override this method to alter how Protocol instances get created.

@param addr: an object implementing L{IAddress}

clientConnectionFailed(connector, reason)[source]

Called when a connection has failed to connect.

It may be useful to call connector.connect() - this will reconnect.

@type reason: L{twisted.python.failure.Failure}

clientConnectionLost(connector, reason)[source]

Called when an established connection is lost.

It may be useful to call connector.connect() - this will reconnect.

@type reason: L{twisted.python.failure.Failure}

colors = True
failed_reconnect_delay = 60
lost_reconnect_delay = 20

alias of IRCBot

rights = None

Called when a connection has been started.

You can call connector.stopConnecting() to stop the connection attempt.

@param connector: a Connector object.

user_types = None
class piqueserver.irc.IRCRelay(protocol, config)[source]

Bases: object

factory = None
me(*arg, **kw)[source]
send(*arg, **kw)[source]
piqueserver.irc.alias(connection, value=None)[source]

This decorator rewrites the username of incoming messages to strip the ident and rejects it if the source channel is not equal to the channel the bot is in


piqueserver.map module

class piqueserver.map.Map(rot_info: piqueserver.map.RotationInfo, load_dir: str)[source]

Bases: object

apply_script(protocol, connection, config)[source]
load_information(rot_info: piqueserver.map.RotationInfo, load_dir: str) → None[source]
exception piqueserver.map.MapNotFound(the_map)[source]

Bases: Exception

class piqueserver.map.RotationInfo(name: str = 'pyspades')[source]

Bases: object

get_map_filename(load_dir: str) → str[source]
get_meta_filename(load_dir: str) → str[source]
get_seed() → int[source]
seed = None
piqueserver.map.check_rotation(maps: List[Union[str, RotationInfo]], load_dir: Optional[str] = None) → List[piqueserver.map.RotationInfo][source]

Checks if provided maps exist in maps dir. and returns an array of RotationInfo objects for those maps. Raises MapNotFound exception if maps are not found.

piqueserver.networkdict module

class piqueserver.networkdict.NetworkDict[source]

Bases: object

pop(*arg, **kw)[source]

remove a key from the networkdict and return the removed items


piqueserver.player module

class piqueserver.player.FeatureConnection(*args, **kwargs)[source]

Bases: pyspades.player.ServerConnection

ban(reason=None, duration=None)[source]
get_spawn_location() → Tuple[int, int, int][source]
kick(reason=None, silent=False)[source]
on_animation_update(jump: bool, crouch: bool, sneak: bool, sprint: bool) → Tuple[bool, bool, bool, bool][source]
on_block_build(x: int, y: int, z: int) → None[source]
on_block_build_attempt(x: int, y: int, z: int) → bool[source]
on_block_destroy(x: int, y: int, z: int, mode: int) → bool[source]
on_block_removed(x: int, y: int, z: int) → None[source]
on_chat(value: str, global_message: bool) → Union[str, bool][source]

notifies when the server receives a chat message

return False to block sending the message

on_command(command: str, parameters: List[str]) → None[source]
on_connect() → None[source]
on_disconnect() → None[source]
on_fall(damage: int) → Optional[bool][source]
on_grenade(time_left: float) → None[source]
on_hit(hit_amount: float, player: piqueserver.player.FeatureConnection, _type: int, grenade: pyspades.world.Grenade) → Optional[bool][source]
on_join() → None[source]
on_kill(killer: Optional[FeatureConnection], _type: int, grenade: None) → None[source]
on_line_build(points) → None[source]
on_line_build_attempt(points) → bool[source]
on_login(name: str) → None[source]
on_reset() → None[source]
on_team_join(team: pyspades.team.Team) → Optional[bool][source]
on_user_login(user_type, verbose=True)[source]
send_lines(lines: List[str]) → None[source]

piqueserver.run module

piqueserver.run.copytree(src, dst)[source]

A re-implementation of shutil.copytree that doesn’t fail if dst already exists. Other properties: Doesn’t over-write if src/dst files don’t differ. Creates a backup of dst file before over-writing.


piqueserver.scheduler module

class piqueserver.scheduler.Scheduler(protocol)[source]

Bases: object

call_end(*arg, **kw)[source]
call_later(*arg, **kw)[source]
loop_call(delay, func, *arg, **kw)[source]

piqueserver.server module

pyspades - default/featured server

class piqueserver.server.FeatureProtocol(interface: bytes, config_dict: Dict[str, Any])[source]

Bases: pyspades.server.ServerProtocol

add_ban(ip, reason, duration, name=None)[source]

Ban an ip with an optional reason and duration in seconds. If duration is None, ban is permanent.

advance_call = None
advance_rotation(message: Optional[str] = None) → twisted.internet.defer.Deferred[source]

Advances to the next map in the rotation. If message is provided it will send it to the chat, waits for 10 seconds and then advances.

Returns:Deferred that fires when the map has been loaded
balanced_teams = None
ban_manager = None
ban_publish = None
bans = None
broadcast_chat(value, global_message=True, sender=None, team=None, irc=False)[source]

Send a chat message to many users

building = True
call_end(delay: int, func: Callable, *arg, **kw) → piqueserver.utils._async.EndCall[source]
connectTCP(*arg, **kw)[source]

alias of piqueserver.player.FeatureConnection

data_received(peer: enet.Peer, packet: enet.Packet) → None[source]
default_fog = (128, 232, 255)
everyone_is_admin = False
format(value: str, extra: Optional[Dict[str, str]] = None) → str[source]
format_lines(value: List[str]) → List[str][source]
game_mode = None
get_advance_time() → float[source]
get_external_ip(ip_getter: str) → Iterator[twisted.internet.defer.Deferred][source]
get_mode_name() → str[source]
global_chat = True
god_blocks = None
identifier = None
interface = None
ip = None
irc_relay = None
irc_say(msg: str, me: bool = False) → None[source]
is_indestructable(x: int, y: int, z: int) → bool[source]
killing = True
last_time = None
listenTCP(*arg, **kw) → twisted.internet.tcp.Port[source]
make_map(rot_info: piqueserver.map.RotationInfo) → twisted.internet.defer.Deferred[source]

Creates and returns a Map object from rotation info in a new thread

Returns:Deferred that resolves to a Map object.
map_info = None
master = False
master_reconnect_call = None
on_advance(map_name: str) → None[source]
on_ban(connection, reason, duration)[source]
on_ban_attempt(connection, reason, duration)[source]
on_map_change(the_map: pyspades.vxl.VXLData) → None[source]
planned_map = None
player_memory = None
receive_callback(address: enet.Address, data: bytes) → int[source]

This hook receives the raw UDP data before it is processed by enet

remote_console = None
send_chat(*args, **kwargs)[source]

Deprecated: see broadcast_chat

set_map_name(rot_info: piqueserver.map.RotationInfo) → None[source]

Sets the map by its name.

set_map_rotation(maps: List[str]) → None[source]

Over-writes the current map rotation with provided one. FeatureProtocol.advance_rotation still needs to be called to actually change the map,

set_server_name(name: str) → None[source]
set_time_limit(time_limit: Optional[int] = None, additive: bool = False) → Optional[int][source]

Notifies players and disconnects them before a shutdown.

spawns = None

alias of FeatureTeam

time_announce_schedule = None
timestamps = None
update_format() → None[source]

Called when the map (or other variables) have been updated

user_blocks = None

remove any bans that might have expired. This takes a while, so it is split up over the event loop


Starts a loop for check_for_releases and updates self.new_release.

class piqueserver.server.FeatureTeam(team_id: int, name: str, color: Tuple[int, int, int], spectator: bool, protocol: pyspades.protocol.BaseProtocol)[source]

Bases: pyspades.team.Team

get_entity_location(entity_id: int) → Tuple[int, int, int][source]
locked = False
piqueserver.server.ensure_dir_exists(filename: str) → None[source]
piqueserver.server.run() → None[source]

runs the server


piqueserver.ssh module

class piqueserver.ssh.RemoteConsole(server)[source]

Bases: object

piqueserver.ssh.create_remote_factory(namespace, users)[source]

piqueserver.statistics module

class piqueserver.statistics.StatsClient[source]

Bases: piqueserver.statistics.StatsProtocol


Called when a connection is made.

This may be considered the initializer of the protocol, because it is called when the connection is completed. For clients, this is called once the connection to the server has been established; for servers, this is called after an accept() call stops blocking and a socket has been received. If you need to send any greeting or initial message, do it here.

login_defers = None
login_user(name, password)[source]
server = None
class piqueserver.statistics.StatsClientFactory(name, password, callback)[source]

Bases: twisted.internet.protocol.ReconnectingClientFactory

maxDelay = 20

alias of StatsClient

class piqueserver.statistics.StatsFactory(password)[source]

Bases: twisted.internet.protocol.ServerFactory


alias of StatsServer

class piqueserver.statistics.StatsProtocol[source]

Bases: twisted.protocols.basic.Int16StringReceiver


Override this for notification when each complete string is received.

@param string: The complete string which was received with all
framing (length prefix, etc) removed.

@type string: C{bytes}

class piqueserver.statistics.StatsServer[source]

Bases: piqueserver.statistics.StatsProtocol

check_user(name, password)[source]

Called when the connection is shut down.

Clear any circular references here, and any external references to this Protocol. The connection has been closed.

@type reason: L{twisted.python.failure.Failure}


Called when a connection is made.

This may be considered the initializer of the protocol, because it is called when the connection is completed. For clients, this is called once the connection to the server has been established; for servers, this is called after an accept() call stops blocking and a socket has been received. If you need to send any greeting or initial message, do it here.

piqueserver.statistics.connect_statistics(host, port, name, password, callback, interface='')[source]

piqueserver.statusserver module

class piqueserver.statusserver.AccessLogger(logger: logging.Logger, log_format: str)[source]

Bases: aiohttp.abc.AbstractAccessLogger

log(request, response, time)[source]

Emit log to logger.

class piqueserver.statusserver.StatusServer(protocol)[source]

Bases: object


Starts the status server on configured host/port


Updates cached overview


Gathers data on current server/game state from protocol class

piqueserver.statusserver.set_default_headers(request, response)[source]

piqueserver.version module