+++ /dev/null
-"""Miscellaneous objects for the MUFF Engine"""
-
-# Copyright (c) 2005 mudpy, Jeremy Stanley <fungi@yuggoth.org>, all rights reserved.
-# Licensed per terms in the LICENSE file distributed with this software.
-
-import ConfigParser
-
-# used by several functions for random calls
-import random
-
-# random_name uses string.strip
-import string
-
-# the log function uses time.asctime for creating timestamps
-import time
-
-# hack to load all modules in the muff package
-import muff
-for module in muff.__all__:
- exec("import " + module)
-
-def broadcast(message):
- """Send a message to all connected users."""
- for each_user in muffvars.userlist: each_user.send("$(eol)" + message)
-
-def log(message):
- """Log a message."""
-
- # the time in posix log timestamp format
- timestamp = time.asctime()[4:19]
-
- # send the timestamp and message to standard output
- print(timestamp + " " + message)
-
-def wrap_ansi_text(text, width):
- """Wrap text with arbitrary width while ignoring ANSI colors."""
-
- # the current position in the entire text string, including all
- # characters, printable or otherwise
- absolute_position = 0
-
- # the current text position relative to the begining of the line,
- # ignoring color escape sequences
- relative_position = 0
-
- # whether the current character is part of a color escape sequence
- escape = False
-
- # iterate over each character from the begining of the text
- for each_character in text:
-
- # the current character is the escape character
- if each_character == chr(27):
- escape = True
-
- # the current character is within an escape sequence
- elif escape:
-
- # the current character is m, which terminates the
- # current escape sequence
- if each_character == "m":
- escape = False
-
- # the current character is a newline, so reset the relative
- # position (start a new line)
- elif each_character == "\n":
- relative_position = 0
-
- # the current character meets the requested maximum line width,
- # so we need to backtrack and find a space at which to wrap
- elif relative_position == width:
-
- # distance of the current character examined from the
- # relative position
- wrap_offset = 0
-
- # count backwards until we find a space
- while text[absolute_position - wrap_offset] != " ":
- wrap_offset += 1
-
- # insert an eol in place of the space
- text = text[:absolute_position - wrap_offset] + "\r\n" + text[absolute_position - wrap_offset + 1:]
-
- # increase the absolute position because an eol is two
- # characters but the space it replaced was only one
- absolute_position += 1
-
- # now we're at the begining of a new line, plus the
- # number of characters wrapped from the previous line
- relative_position = wrap_offset
-
- # as long as the character is not a carriage return and the
- # other above conditions haven't been met, count it as a
- # printable character
- elif each_character != "\r":
- relative_position += 1
-
- # increase the absolute position for every character
- absolute_position += 1
-
- # return the newly-wrapped text
- return text
-
-def weighted_choice(data):
- """Takes a dict weighted by value and returns a random key."""
-
- # this will hold our expanded list of keys from the data
- expanded = []
-
- # create thee expanded list of keys
- for key in data.keys():
- for count in range(data[key]):
- expanded.append(key)
-
- # return one at random
- return random.choice(expanded)
-
-def random_name():
- """Returns a random character name."""
-
- # the vowels and consonants needed to create romaji syllables
- vowels = [ "a", "i", "u", "e", "o" ]
- consonants = ["'", "k", "z", "s", "sh", "z", "j", "t", "ch", "ts", "d", "n", "h", "f", "m", "y", "r", "w" ]
-
- # this dict will hold our weighted list of syllables
- syllables = {}
-
- # generate the list with an even weighting
- for consonant in consonants:
- for vowel in vowels:
- syllables[consonant + vowel] = 1
-
- # we'll build the name into this string
- name = ""
-
- # create a name of random length from the syllables
- for syllable in range(random.randrange(2, 6)):
- name += weighted_choice(syllables)
-
- # strip any leading quotemark, capitalize and return the name
- return string.strip(name, "'").capitalize()
-
-def replace_macros(user, text, is_input=False):
- """Replaces macros in text output."""
- while True:
- macro_start = string.find(text, "$(")
- if macro_start == -1: break
- macro_end = string.find(text, ")", macro_start) + 1
- macro = text[macro_start:macro_end]
- if macro in muffvars.macros.keys():
- text = string.replace(text, macro, muffvars.macros[macro])
-
- # the user's account name
- elif macro == "$(account)":
- text = string.replace(text, macro, user.name)
-
- # third person subjective pronoun
- elif macro == "$(tpsp)":
- if user.avatar.get("gender") == "male":
- text = string.replace(text, macro, "he")
- elif user.avatar.get("gender") == "female":
- text = string.replace(text, macro, "she")
- else:
- text = string.replace(text, macro, "it")
-
- # third person objective pronoun
- elif macro == "$(tpop)":
- if user.avatar.get("gender") == "male":
- text = string.replace(text, macro, "him")
- elif user.avatar.get("gender") == "female":
- text = string.replace(text, macro, "her")
- else:
- text = string.replace(text, macro, "it")
-
- # third person possessive pronoun
- elif macro == "$(tppp)":
- if user.avatar.get("gender") == "male":
- text = string.replace(text, macro, "his")
- elif user.avatar.get("gender") == "female":
- text = string.replace(text, macro, "hers")
- else:
- text = string.replace(text, macro, "its")
-
- # if we get here, log and replace it with null
- else:
- text = string.replace(text, macro, "")
- if not is_input:
- log("Unexpected replacement macro " + macro + " encountered.")
-
- # replace the look-like-a-macro sequence
- text = string.replace(text, "$_(", "$(")
-
- return text
-
-def check_time(frequency):
- """Check for a factor of the current increment count."""
- if type(frequency) is str:
- frequency = muffconf.getint("time", frequency)
- return not muffuniv.universe.internals["counters"].getint("elapsed") % frequency
-
-def on_pulse():
- """The things which should happen on each pulse, aside from reloads."""
-
- # open the listening socket if it hasn't been already
- if not muffvars.newsocket: muffsock.initialize_server_socket()
-
- # assign a user if a new connection is waiting
- user = muffsock.check_for_connection(muffvars.newsocket)
- if user: muffvars.userlist.append(user)
-
- # iterate over the connected users
- for user in muffvars.userlist: user.pulse()
-
- # update the log every now and then
- if check_time("frequency_log"):
- log(repr(len(muffvars.userlist)) + " connection(s)")
-
- # periodically save everything
- if check_time("frequency_save"):
- for user in muffvars.userlist: user.save()
- muffuniv.universe.save()
-
- # pause for a configurable amount of time (decimal seconds)
- time.sleep(muffconf.getfloat("time", "increment"))
-
- # increment the elapsed increment counter
- muffuniv.universe.internals["counters"].set("elapsed", muffuniv.universe.internals["counters"].getint("elapsed") + 1)
-
-def reload_data():
- """Reload data into new persistent objects."""
-
- # reload the users
- temporary_userlist = []
- for user in muffvars.userlist: temporary_userlist.append(user)
- for user in temporary_userlist: user.reload()
- del(temporary_userlist)
-
projects / mudpy.git / blobdiff