smserver package

Subpackages

Submodules

smserver.ability module

The Ability module provide helper for checking the permission

class smserver.ability.Ability

Bases: object

The Ablitity class provide easy method for checking if someone can do a given action

static can(action, level=0)

Return true if the action is possible with the given level

Parameters:
  • action (Permissions) – Action to test
  • level (int) – Level for this action
static cannot(action, level=0)

Return true if the action is not possible with the given level

Parameters:
  • action (Permissions) – Action to test
  • level (int) – Level for this action
class smserver.ability.Permissions

Bases: enum.Enum

List of available permissions

The value of each member is the level required for the given action

smserver.authplugin module

class smserver.authplugin.AuthPlugin(server, autocreate)

Bases: object

login(user, password)

smserver.chathelper module

smserver.chathelper.chat_color(color)

Stepmania chat color format

Parameters:color (str) – Color in hex format
Example: 
>>> chat_color("cecece")
'|c0cecece'
smserver.chathelper.nick_color(nick)

Generate a color from a given string.

Parameters:nick (str) – Input string
Returns:A hex color
Return type:str 
>>> nick_color("A")
'005ad0'

>>> nick_color("Hello!")
'f20c1a'
smserver.chathelper.with_color(message, color=None)

Add a color to a message, for Stepmania Chat.

If no color is provided, it guess the color with nick_color

Parameters:
  • message (str) – Message to color
  • color (str) – Color in hex format
Example: 
>>> with_color("Hello")
'|c0f20c0cHello|c0ffffff'

>>> with_color("Hello", "cecece")
'|c0cececeHello|c0ffffff'

smserver.chatplugin module

This module add the class needed for creating custom chat command

Example:

Here’s a simple ChatPlugin which will send a HelloWorld on use:

class ChatHelloWorld(ChatPlugin):
    helper = "Display Hello World"
    command = "hello"

    def __call__(self, serv, message):
        serv.send_message("Hello world", to="me")
class smserver.chatplugin.ChatPlugin

Bases: object

Inherit from this class to add a command in the chat.

can(serv)

Method call each time somenone try to run this command

Parameters:serv (StepmaniaController) – The StepmaniaController instance
Returns:True if authorize False if not
Return type:bool
command = None

The command to use to call this function

Return type:str
helper = ''

Text that will be show when calling the help command

Return type:str
permission = None

Permission needed for this command (see ability)

Return type:smserver.ability.Permissions
room = False

Specify here if the command need to be execute in a room

Return type:bool

smserver.conf module

class smserver.conf.Conf(*args)

Bases: dict

Configuration dictionnary.

It use yaml configuration file by default, and some option can be change by passing it in argument (with argparse)

static add_to_conf(conf, arg, value, replace=True)

Add the given value to the conf

Parameters:
  • conf (dict) – The configuration dictionnary
  • arg (str) – The path of the key, with dot notation
  • value – Value to assign
  • replace (bool) – If the key already exist replace it
Example: 
>>> conf = {}
>>> Conf.add_to_conf(conf, "root.key", "value")
>>> print(conf)
{'root': {'key': 'value'}}

>>> Conf.add_to_conf(conf, "root.key", "new_value", replace=False)
>>> print(conf)
{'root': {'key': 'value'}}

>>> Conf.add_to_conf(conf, "root.key", "new_value", replace=True)
>>> print(conf)
{'root': {'key': 'new_value'}}
parser = ArgumentParser(prog='sphinx-build', usage=None, description='Stepmania configuration', formatter_class=<class 'argparse.HelpFormatter'>, conflict_handler='resolve', add_help=True)
reload()

Reload the configuration file

smserver.database module

class smserver.database.DataBase(type='sqlite', database=None, user=None, password=None, host=None, port=None, driver=None)

Bases: object

The DataBase class hold information about a given database.

It’s a wrapper around SQLAlchemy database creation. By default, it create a in memory sqlite database.

Parameters:
  • type (str) – Type of database (sqlite, mysql, postgresql, ...)
  • database (str) – Name of the database (file for sqlite)
  • user (str) – User of the database
  • password (str) – Password of the database
  • host (str) – Location of the database (localhost for local)
  • port (int) – Port of the database
  • driver (str) – Driver for communication with the database.
create_tables()

Create all the table in the DataBase if they don’t exist.

engine

SQLAlchemy engine assosiate with the DataBase

recreate_tables()

Drop and recreate all the table in the DataBase

session

Return a new SQLAlchemy Session

session_scope()

Provide a transactional scope around a series of operations.

classmethod test_db()

Return a inmemory database

smserver.logger module

class smserver.logger.Logger(options=None)

Bases: object

logger

smserver.pluginmanager module

exception smserver.pluginmanager.PluginError

Bases: Exception

class smserver.pluginmanager.PluginManager(plugin_class, paths=None, directory=None, plugin_file=None, force_reload=False)

Bases: list

all_paths(directory)
classmethod get_plugin(path, plugin_classes, default=None, force_reload=False)
static import_plugin(path, plugin_classes, force_reload=False)
init(*opt)
load(force_reload=False)
class smserver.pluginmanager.StepmaniaPlugin(server)

Bases: object

on_createroom(session, serv, packet)
on_enterroom(session, serv, packet)
on_login(session, serv, packet)
on_nscattack(session, serv, packet)
on_nsccm(session, serv, packet)
on_nsccuul(session, serv, packet)
on_nscformatted(session, serv, packet)
on_nscgon(session, serv, packet)
on_nscgsr(session, serv, packet)
on_nscgsu(session, serv, packet)
on_nschello(session, serv, packet)
on_nscping(session, serv, packet)
on_nscpingr(session, serv, packet)
on_nscrsg(session, serv, packet)
on_nscsu(session, serv, packet)
on_nscuopts(session, serv, packet)
on_nsscsms(session, serv, packet)
on_nssmonl(session, serv, packet)
on_packet(session, serv, packet)
on_roominfo(session, serv, packet)
on_xmlpacket(session, serv, packet)

smserver.sdnotify module

Provide sd_notify wrapper to notify service manager about start-up completion and other service

class smserver.sdnotify.SDNotify

Bases: object

Notify service manager about start-up completion and other service status changes

Example: 
>>> sd_notify = SDNotify()
>>> sd_notify.ready()
notify(state)

Notify may be called by a service to notify the service manager about state changes.

It can be used to send arbitrary information, encoded in an environment-block-like string.

Most importantly, it can be used for start-up completion notification.

ready()

Tells the service manager that service startup is finished.

This is only used by systemd if the service definition file has Type=notify set.

reloading()

Tells the service manager that the service is reloading its configuration.

This is useful to allow the service manager to track the service’s internal state, and present it to the user.

Note that a service that sends this notification must also send a ready notification when it completed reloading its configuration.

status(status)

Passes a single-line UTF-8 status string back to the service manager that describes the service state.

This is free-form and can be used for various purposes:

stopping()

Tells the service manager that the service is beginning its shutdown.

This is useful to allow the service manager to track the service’s internal state, and present it to the user.

watchdog()

Tells the service manager to update the watchdog timestamp.

This is the keep-alive ping that services need to issue in regular intervals if WatchdogSec= is enabled for it.

smserver.server module

class smserver.server.StepmaniaServer(config=None)

Bases: smserver.smutils.smthread.StepmaniaServer

It’s the main class of the server. It will start a new thread for each configured server.

To start the server you will generally use:

from smserver import conf, server

config = conf.Conf(*sys.argv[1:])

server.StepmaniaServer(config).start()
add_connection(*opt)
disconnect_user(user_id)

Disconnect the given user

enter_room(room, user_id=None, conn=None)

Make a user enter in a given room

Parameters:
  • room (smserver.models.room.Room) – Room where the user have to enter
  • user_id (int) – User id which enter the room
  • conn (smserver.smutils.smconn.StepmaniaConnection) – Connection which enter the room
handle_packet(session, serv, packet)

Handle the given packet for a specific connection.

It will launch every controllers that fetch the packet requirement and try to run every plugins.

leave_room(room, user_id=None, conn=None)

Make a user leave a given room

Parameters:
  • room (smserver.models.room.Room) – Room where the user have to leave
  • user_id (int) – User id which leave the room
  • conn (smserver.smutils.smconn.StepmaniaConnection) – Connection which leave the room
on_disconnect(*opt)
on_packet(*opt)
reload()

Reload configuration files

send_message(message, room=None, conn=None, func=None)

Send a chat message (default to all)

Parameters:
send_sd_running_status()

Send running status to systemd

send_user_list(room)

Send a NSCUUL packet to update the use list for a given room

start()

Start all the threads

stop()

Close all the threads

smserver.server.main()
smserver.server.with_session(func)

Wrap the function with a sqlalchemy session.

Use:

@with_session
def func_with_session(self, session):
    pass

Only work with instance methods of StepmaniaServer class

smserver.stepmania_controller module

This module add the class use to handle smpacket.

All the controllers placed in the controllers folder will be loaded. You can also add your own controller in the plugins folder.

Example:

Here’s a simple Controller which will send a HelloWorld on every PingMessage:

class ControllerHelloWorld(StepmaniaController):
    command = smpacket.SMClientCommand.NSCPing
    require_login = False

    def handle(self):
        serv.send_message("Hello world", to="me")
class smserver.stepmania_controller.StepmaniaController(server, conn, packet, session)

Bases: object

Inherit from this class to add an action every time you will see a specific packet.

A new instance of StepmaniaController is instantiate on each incoming packet.

Parameters:
active_users

Return the list of connected user’s object which are still online.

can(action, room_id=None)

Return True if this connection can do the specified action

Parameters:
Returns:

True if the action in authorized
cannot(action, room_id=None)

Return True if this connection cannot do the specified action

Parameters:
Returns:

True if the action in unauthorized
colored_user_repr(room_id=None)

Colored textual representation of the users connected on this connection. Use it when sending chat message

Parameters:room_id (int) – The ID of the room
command = None

The command to handle in this controller.

Return type:smserver.smutils.smpacket.SMClientCommand
handle()

This method is call on every incoming packet.

Do all the stuff you need here.

level(room_id=None)

The maximum level of the users in this connection

Parameters:room_id (int) – The ID of the room.
Returns:Level of the user
Return type:int
require_login = False

Specify if the user has to be log in.

Return type:bool
room

The room object where the user is.

Return None if the user is not in a room

room_users

Return the list of user currently in the same room

send(packet)

Send the specified packet to the current connection

Parameters:packet (smserver.smutils.smpacket.SMPacket) – The packet to send
Returns:nothing
send_message(message, to=None, room_id=None)

Send a chat message

Parameters:
  • message (str) – The message to send.
  • to (str) – Send the message to ? (room, all, me). Default to room
  • room_id (int) – A specific room. Default to the room connection.
send_user_message(message, to=None)

Same as send_message but prepend the user repr to the message

Parameters:
  • message (str) – The message to send.
  • to (str) – Send the message to ? (room, all, me). Default to room
sendall(packet)

Send the specified packet to all the connections on the server

Parameters:packet (smserver.smutils.smpacket.SMPacket) – The packet to send
Returns:nothing
sendingame(room_id, packet)

Send the specified packet to all the players in the specified room which have send an NSCGSR packet

Parameters:
Returns:

nothing
sendplayers(room_id, packet)

Send the specified packet to all the players in the specified room (Not spectator)

Parameters:
Returns:

nothing
sendroom(room, packet)

Send the specified packet to all the connection in the specified room

Parameters:
Returns:

nothing
song()

Return the song object the last song the user have selected.

Return None if there is no such song.

user_repr(room_id=None)

Textual representation of the users connected on this connection.

Parameters:room_id (int) – The ID of the room
users

Return the list of connected user’s object.

smserver.watcher module

class smserver.watcher.PeriodicMethods

Bases: object

Decorator to indicate the periodicity of a methods

class smserver.watcher.StepmaniaWatcher(server)

Bases: threading.Thread

Secondary thread, hold by the main server. Call periodically each function with a ‘periodicmethod decorator’

UDP_IP = '255.255.255.255'
UDP_PORT = 8765
check_end_game(*opts)
check_song_start(session, room_id, room_conns)
force_run()
room_still_in_game(room)
run()
scoreboard_update(*opts)
sdnotify_watchdog(*opts)
send_game_start(*opts)
send_ping(*opts)
send_scoreboard(room, session)
send_udp(*opts)
stop()

End the loop

Module contents

Stepmania Server implementation