"""Miscellaneous objects for the MUFF Engine"""
-# Copyright (c) 2005 mudpy, Jeremy Stanley <fungi@yuggoth.org>
-# All rights reserved.
-#
-# Redistribution and use in source and binary forms, with or without
-# modification, are permitted provided that the following conditions
-# are met:
-#
-# - Redistributions of source code must retain the above copyright
-# notice, this list of conditions and the following disclaimer.
-# - Redistributions in binary form must reproduce the above
-# copyright notice, this list of conditions and the following
-# disclaimer in the documentation and/or other materials provided
-# with the distribution.
-#
-# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
-# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
-# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
-# FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
-# COPYRIGHT HOLDERS OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
-# INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
-# BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
-# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
-# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
-# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
-# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
-# POSSIBILITY OF SUCH DAMAGE.
+# 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
+
+# used to match the 'L' at the end of a long int in repr_long
+import re
+
+# 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(output):
- for each_user in muffvars.userlist:
- each_user.send(output)
+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):
- relative_position = 0
+ """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
- escape = 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 = 1
+ 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 = 0
+ 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 repr_long(value):
+ string_value = repr(value)
+ if re.match('\d*L$', string_value): return string_value.strip("L")
+ else: return string_value
+
+def getlong(config, section, option):
+ try:
+ return int(config.get(section, option).strip("L"))
+ except ConfigParser.NoSectionError:
+ config.add_section(section)
+ return getlong(config, section, option)
+ except ConfigParser.NoOptionError:
+ setlong(config, section, option, 0)
+ return getlong(config, section, option)
+
+def setlong(config, section, option, value):
+ return config.set(section, option, repr_long(value))
+
+def replace_macros(user, text, is_input=False):
+ 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 getlong(muffvars.variable_data, "time", "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"):
+ muffvars.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
+ setlong(muffvars.variable_data, "time", "elapsed",
+ getlong(muffvars.variable_data, "time", "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