"""Miscellaneous functions for the mudpy engine."""
-# Copyright (c) 2004-2016 Jeremy Stanley <fungi@yuggoth.org>. Permission
-# to use, copy, modify, and distribute this software is granted under
-# terms provided in the LICENSE file distributed with this software.
+# Copyright (c) 2004-2018 mudpy authors. Permission to use, copy,
+# modify, and distribute this software is granted under terms
+# provided in the LICENSE file distributed with this software.
import codecs
import os
"""An element of the universe."""
- def __init__(self, key, universe, filename=None, old_style=False):
+ def __init__(self, key, universe, origin=None):
"""Set up a new element."""
- # TODO(fungi): This can be removed after the transition is complete
- self.old_style = old_style
-
# keep track of our key name
self.key = key
# keep track of what universe it's loading into
self.universe = universe
- # clone attributes if this is replacing another element
- if self.old_style and self.key in self.universe.contents:
- old_element = self.universe.contents[self.key]
- for attribute in vars(old_element).keys():
- exec("self." + attribute + " = old_element." + attribute)
- if self.owner:
- self.owner.avatar = self
-
- # i guess this is a new element then
- else:
+ # set of facet keys from the universe
+ self.facethash = dict()
- # set of facet keys from the universe
- self.facethash = dict()
+ # not owned by a user by default (used for avatars)
+ self.owner = None
- # not owned by a user by default (used for avatars)
- self.owner = None
+ # no contents in here by default
+ self.contents = {}
- # no contents in here by default
- self.contents = {}
+ if self.key.find(".") > 0:
+ self.group, self.subkey = self.key.split(".")[-2:]
+ else:
+ self.group = "other"
+ self.subkey = self.key
+ if self.group not in self.universe.groups:
+ self.universe.groups[self.group] = {}
- # parse out appropriate category and subkey names, add to list
- if self.key.find(":") > 0:
- self.category, self.subkey = self.key.split(":", 1)
- else:
- self.category = "other"
- self.subkey = self.key
- if self.category not in self.universe.categories:
- self.category = "other"
- self.subkey = self.key
-
- # get an appropriate filename for the origin
- if not filename:
- filename = self.universe.default_origins[self.category]
- if not os.path.isabs(filename):
- filename = os.path.abspath(filename)
-
- # add the file if it doesn't exist yet
- if filename not in self.universe.files:
- mudpy.data.DataFile(filename, self.universe)
+ # get an appropriate origin
+ if not origin:
+ self.universe.add_group(self.group)
+ origin = self.universe.files[
+ self.universe.origins[self.group]["fallback"]]
# record or reset a pointer to the origin file
- self.origin = self.universe.files[filename]
-
- # add a data section to the origin if necessary
- if self.key not in self.origin.data:
- self.origin.data[self.key] = {}
+ self.origin = self.universe.files[origin.source]
# add or replace this element in the universe
self.universe.contents[self.key] = self
- self.universe.categories[self.category][self.subkey] = self
+ self.universe.groups[self.group][self.subkey] = self
def reload(self):
"""Create a new element and replace this one."""
- Element(self.key, self.universe, self.origin.filename,
- old_style=self.old_style)
- del(self)
+ args = (self.key, self.universe, self.origin)
+ self.destroy()
+ Element(*args)
def destroy(self):
"""Remove an element from the universe and destroy it."""
- del(self.origin.data[self.key])
- del self.universe.categories[self.category][self.subkey]
+ for facet in dict(self.facethash):
+ self.remove_facet(facet)
+ del self.universe.groups[self.group][self.subkey]
del self.universe.contents[self.key]
del self
def facets(self):
"""Return a list of non-inherited facets for this element."""
- if self.old_style:
- try:
- return self.origin.data[self.key].keys()
- except (AttributeError, KeyError):
- return []
- else:
- return self.facethash
+ return self.facethash
def has_facet(self, facet):
"""Return whether the non-inherited facet exists."""
def remove_facet(self, facet):
"""Remove a facet from the element."""
- if self.has_facet(facet):
- del(self.origin.data[self.key][facet])
- self.origin.modified = True
+ if ".".join((self.key, facet)) in self.origin.data:
+ del self.origin.data[".".join((self.key, facet))]
+ if facet in self.facethash:
+ del self.facethash[facet]
+ self.origin.modified = True
def ancestry(self):
"""Return a list of the element's inheritance lineage."""
if default is None:
default = ""
try:
- if self.old_style:
- return self.origin.data[self.key][facet]
- else:
- return self.origin.data[".".join((self.key, facet))]
+ return self.origin.data[".".join((self.key, facet))]
except (KeyError, TypeError):
pass
if self.has_facet("inherit"):
def set(self, facet, value):
"""Set values."""
- if not self.has_facet(facet) or not self.get(facet) == value:
- if self.old_style:
- if self.key not in self.origin.data:
- self.origin.data[self.key] = {}
- self.origin.data[self.key][facet] = value
- else:
- node = ".".join((self.key, facet))
- self.origin.data[node] = value
- self.facethash[node] = self.origin.data[node]
+ if not self.origin.is_writeable() and not self.universe.loading:
+ # break if there is an attempt to update an element from a
+ # read-only file, unless the universe is in the midst of loading
+ # updated data from files
+ raise PermissionError("Altering elements in read-only files is "
+ "disallowed")
+ # Coerce some values to appropriate data types
+ # TODO(fungi) Move these to a separate validation mechanism
+ if facet in ["loglevel"]:
+ value = int(value)
+ elif facet in ["administrator"]:
+ value = bool(value)
+
+ # The canonical node for this facet within its origin
+ node = ".".join((self.key, facet))
+
+ if node not in self.origin.data or self.origin.data[node] != value:
+ # Be careful to only update the origin's contents when required,
+ # since that affects whether the backing file gets written
+ self.origin.data[node] = value
self.origin.modified = True
+ # Make sure this facet is included in the element's facets
+ self.facethash[facet] = self.origin.data[node]
+
def append(self, facet, value):
"""Append value to a list."""
newlist = self.get(facet)
def can_run(self, command):
"""Check if the user can run this command object."""
- # has to be in the commands category
- if command not in self.universe.categories["command"].values():
+ # has to be in the commands group
+ if command not in self.universe.groups["command"].values():
result = False
# avatars of administrators can run any command
def portals(self):
"""Map the portal directions for an area to neighbors."""
portals = {}
- if re.match("""^area:-?\d+,-?\d+,-?\d+$""", self.key):
+ if re.match(r"""^area\.-?\d+,-?\d+,-?\d+$""", self.key):
coordinates = [(int(x))
- for x in self.key.split(":")[1].split(",")]
- offsets = dict((x,
- self.universe.contents["mudpy.movement.%s" % x].get("vector")
- ) for x in self.universe.directions)
+ for x in self.key.split(".")[-1].split(",")]
+ offsets = dict(
+ (x,
+ self.universe.contents["mudpy.movement.%s" % x].get("vector")
+ ) for x in self.universe.directions)
for portal in self.get("gridlinks"):
adjacent = map(lambda c, o: c + o,
coordinates, offsets[portal])
- neighbor = "area:" + ",".join(
+ neighbor = "area." + ",".join(
[(str(x)) for x in adjacent]
)
if neighbor in self.universe.contents:
def __init__(self, filename="", load=False):
"""Initialize the universe."""
- self.categories = {}
+ self.groups = {}
self.contents = {}
- self.default_origins = {}
self.directions = set()
+ self.loading = False
self.loglines = []
- self.private_files = []
+ self.origins = {}
self.reload_flag = False
self.setup_loglines = []
self.startdir = os.getcwd()
self.terminate_flag = False
self.userlist = []
+ self.versions = None
if not filename:
possible_filenames = [
"etc/mudpy.yaml",
def load(self):
"""Load universe data from persistent storage."""
+ # while loading, it's safe to update elements from read-only files
+ self.loading = True
+
# it's possible for this to enter before logging configuration is read
pending_loglines = []
- # the files dict must exist and filename needs to be read-only
- if not hasattr(
- self, "files"
- ) or not (
- self.filename in self.files and self.files[
- self.filename
- ].is_writeable()
- ):
+ # start populating the (re)files dict from the base config
+ self.files = {}
+ mudpy.data.Data(self.filename, self)
- # clear out all read-only files
- if hasattr(self, "files"):
- for data_filename in list(self.files.keys()):
- if not self.files[data_filename].is_writeable():
- del self.files[data_filename]
+ # load default storage locations for groups
+ if hasattr(self, "contents") and "mudpy.filing" in self.contents:
+ self.origins.update(self.contents["mudpy.filing"].get(
+ "groups", {}))
- # start loading from the initial file
- mudpy.data.DataFile(self.filename, self)
+ # add some builtin groups we know we'll need
+ for group in ("account", "actor", "internal"):
+ self.add_group(group)
# make a list of inactive avatars
inactive_avatars = []
- for account in self.categories["account"].values():
+ for account in self.groups.get("account", {}).values():
for avatar in account.get("avatars"):
try:
inactive_avatars.append(self.contents[avatar])
except KeyError:
pending_loglines.append((
- "Missing avatar \"%s\", possible data corruption" %
+ 'Missing avatar "%s", possible data corruption' %
avatar, 6))
for user in self.userlist:
if user.avatar in inactive_avatars:
inactive_avatars.remove(user.avatar)
- # go through all elements to clear out inactive avatar locations
- for element in self.contents.values():
- area = element.get("location")
- if element in inactive_avatars and area:
- if area in self.contents and element.key in self.contents[
- area
- ].contents:
- del self.contents[area].contents[element.key]
- element.set("default_location", area)
- element.remove_facet("location")
-
# another pass to straighten out all the element contents
for element in self.contents.values():
element.update_location()
element.clean_contents()
+
+ # done loading, so disallow updating elements from read-only files
+ self.loading = False
+
return pending_loglines
def new(self):
def get_time(self):
"""Convenience method to get the elapsed time counter."""
- return self.categories["internal"]["counters"].get("elapsed")
+ return self.groups["internal"]["counters"].get("elapsed")
+
+ def add_group(self, group, fallback=None):
+ """Set up group tracking/metadata."""
+ if group not in self.origins:
+ self.origins[group] = {}
+ if not fallback:
+ fallback = mudpy.data.find_file(
+ ".".join((group, "yaml")), universe=self)
+ if "fallback" not in self.origins[group]:
+ self.origins[group]["fallback"] = fallback
+ flags = self.origins[group].get("flags", None)
+ if fallback not in self.files:
+ mudpy.data.Data(fallback, self, flags=flags)
class User:
self.output_queue = []
self.partial_input = b""
self.password_tries = 0
- self.state = "initial"
+ self.state = "telopt_negotiation"
self.telopts = {}
+ self.universe = universe
def quit(self):
"""Log, close the connection and remove."""
if self.account:
- name = self.account.get("name")
- else:
- name = ""
- if name:
- message = "User " + name
+ name = self.account.get("name", self)
else:
- message = "An unnamed user"
- message += " logged out."
- log(message, 2)
+ name = self
+ log("Logging out %s" % name, 2)
self.deactivate_avatar()
self.connection.close()
self.remove()
def check_idle(self):
"""Warn or disconnect idle users as appropriate."""
idletime = universe.get_time() - self.last_input
- linkdead_dict = universe.categories["internal"]["time"].get(
- "linkdead"
- )
+ linkdead_dict = universe.contents[
+ "mudpy.timing.idle.disconnect"].facets()
if self.state in linkdead_dict:
linkdead_state = self.state
else:
log(logline, 2)
self.state = "disconnecting"
self.menu_seen = False
- idle_dict = universe.categories["internal"]["time"].get("idle")
+ idle_dict = universe.contents["mudpy.timing.idle.warn"].facets()
if self.state in idle_dict:
idle_state = self.state
else:
def reload(self):
"""Save, load a new user and relocate the connection."""
+ # copy old attributes
+ attributes = self.__dict__
+
# get out of the list
self.remove()
+ # get rid of the old user object
+ del(self)
+
# create a new user object
new_user = User()
# set everything equivalent
- for attribute in vars(self).keys():
- exec("new_user." + attribute + " = self." + attribute)
+ new_user.__dict__ = attributes
# the avatar needs a new owner
if new_user.avatar:
+ new_user.account = universe.contents[new_user.account.key]
+ new_user.avatar = universe.contents[new_user.avatar.key]
new_user.avatar.owner = new_user
# add it to the list
universe.userlist.append(new_user)
- # get rid of the old user object
- del(self)
-
def replace_old_connections(self):
"""Disconnect active users with the same name."""
def authenticate(self):
"""Flag the user as authenticated and disconnect duplicates."""
if self.state is not "authenticated":
- log("User " + self.account.get("name") + " logged in.", 2)
self.authenticated = True
- if self.account.subkey in universe.contents["mudpy.limit"].get(
- "admins"):
- self.account.set("administrator", "True")
+ if ("mudpy.limit" in universe.contents and self.account.subkey in
+ universe.contents["mudpy.limit"].get("admins")):
+ self.account.set("administrator", True)
+ log("Administrator %s authenticated." %
+ self.account.get("name"), 2)
+ else:
+ log("User %s authenticated for account %s." % (
+ self, self.account.subkey), 2)
def show_menu(self):
"""Send the user their current menu."""
def remove(self):
"""Remove a user from the list of connected users."""
+ log("Disconnecting account %s." % self, 0)
universe.userlist.remove(self)
def send(
self.check_idle()
# if output is paused, decrement the counter
- if self.state == "initial":
+ if self.state == "telopt_negotiation":
if self.negotiation_pause:
self.negotiation_pause -= 1
else:
if self.output_queue:
try:
self.connection.send(self.output_queue[0])
- except BrokenPipeError:
+ except (BrokenPipeError, ConnectionResetError):
if self.account and self.account.get("name"):
account = self.account.get("name")
else:
account = "an unknown user"
self.state = "disconnecting"
- log("Broken pipe sending to %s." % account, 7)
+ log("Disconnected while sending to %s." % account, 7)
del self.output_queue[0]
def enqueue_input(self):
mudpy.telnet.negotiate_telnet_options(self)
# separate multiple input lines
- new_input_lines = self.partial_input.split(b"\n")
+ new_input_lines = self.partial_input.split(b"\r\0")
+ if len(new_input_lines) == 1:
+ new_input_lines = new_input_lines[0].split(b"\r\n")
# if input doesn't end in a newline, replace the
# held partial input with the last line of it
- if not self.partial_input.endswith(b"\n"):
+ if not (
+ self.partial_input.endswith(b"\r\0") or
+ self.partial_input.endswith(b"\r\n")):
self.partial_input = new_input_lines.pop()
# otherwise, chop off the extra null input and reset
line = line.strip()
# log non-printable characters remaining
- if mudpy.telnet.is_enabled(self, mudpy.telnet.TELOPT_BINARY,
- mudpy.telnet.HIM):
+ if not mudpy.telnet.is_enabled(
+ self, mudpy.telnet.TELOPT_BINARY, mudpy.telnet.HIM):
asciiline = bytes([x for x in line if 32 <= x <= 126])
if line != asciiline:
logline = "Non-ASCII characters from "
try:
line = line.decode("utf-8")
except UnicodeDecodeError:
- logline = "Non-UTF-8 characters from "
+ logline = "Non-UTF-8 sequence from "
if self.account and self.account.get("name"):
logline += self.account.get("name") + ": "
else:
def new_avatar(self):
"""Instantiate a new, unconfigured avatar for this user."""
counter = 0
- while "avatar:" + self.account.get("name") + ":" + str(
- counter
- ) in universe.categories["actor"].keys():
+ while ("avatar_%s_%s" % (self.account.get("name"), counter)
+ in universe.groups.get("actor", {}).keys()):
counter += 1
self.avatar = Element(
- "actor:avatar:" + self.account.get("name") + ":" + str(
- counter
- ),
- universe, old_style=True
- )
- self.avatar.append("inherit", "archetype:avatar")
+ "actor.avatar_%s_%s" % (self.account.get("name"), counter),
+ universe)
+ self.avatar.append("inherit", "archetype.avatar")
self.account.append("avatars", self.avatar.key)
+ log("Created new avatar %s for user %s." % (
+ self.avatar.key, self.account.get("name")), 0)
def delete_avatar(self, avatar):
"""Remove an avatar from the world and from the user's list."""
if self.avatar is universe.contents[avatar]:
self.avatar = None
+ log("Deleting avatar %s for user %s." % (
+ avatar, self.account.get("name")), 0)
universe.contents[avatar].destroy()
avatars = self.account.get("avatars")
avatars.remove(avatar)
self.account.get("avatars")[index]]
self.avatar.owner = self
self.state = "active"
+ log("Activated avatar %s (%s)." % (
+ self.avatar.get("name"), self.avatar.key), 0)
self.avatar.go_home()
def deactivate_avatar(self):
"""Have the active avatar leave the world."""
if self.avatar:
+ log("Deactivating avatar %s (%s) for user %s." % (
+ self.avatar.get("name"), self.avatar.key,
+ self.account.get("name")), 0)
current = self.avatar.get("location")
if current:
self.avatar.set("default_location", current)
"""Destroy the user and associated avatars."""
for avatar in self.account.get("avatars"):
self.delete_avatar(avatar)
+ log("Destroying account %s for user %s." % (
+ self.account.get("name"), self), 0)
self.account.destroy()
def list_avatar_names(self):
try:
avatars.append(universe.contents[avatar].get("name"))
except KeyError:
- log("Missing avatar \"%s\", possible data corruption." %
+ log('Missing avatar "%s", possible data corruption.' %
avatar, 6)
return avatars
"""Log a message."""
# a couple references we need
- file_name = universe.contents["mudpy.log"].get("file")
- max_log_lines = universe.contents["mudpy.log"].get("lines")
- syslog_name = universe.contents["mudpy.log"].get("syslog")
+ if "mudpy.log" in universe.contents:
+ file_name = universe.contents["mudpy.log"].get("file", "")
+ max_log_lines = universe.contents["mudpy.log"].get("lines", 0)
+ syslog_name = universe.contents["mudpy.log"].get("syslog", "")
+ else:
+ file_name = ""
+ max_log_lines = 0
+ syslog_name = ""
timestamp = time.asctime()[4:19]
# turn the message into a list of nonempty lines
if file_name:
if not os.path.isabs(file_name):
file_name = os.path.join(universe.startdir, file_name)
+ os.makedirs(os.path.dirname(file_name), exist_ok=True)
file_descriptor = codecs.open(file_name, "a", "utf-8")
for line in lines:
file_descriptor.write(timestamp + " " + line + "\n")
file_descriptor.close()
# send the timestamp and line to standard output
- if universe.contents["mudpy.log"].get("stdout"):
+ if ("mudpy.log" in universe.contents and
+ universe.contents["mudpy.log"].get("stdout")):
for line in lines:
print(timestamp + " " + line)
# ignoring color escape sequences
rel_pos = 0
- # the absolute position of the most recent whitespace character
- last_whitespace = 0
+ # the absolute and relative positions of the most recent whitespace
+ # character
+ last_abs_whitespace = 0
+ last_rel_whitespace = 0
# whether the current character is part of a color escape sequence
escape = False
# the current character is the escape character
if each_character == "\x1b" and not escape:
escape = True
+ rel_pos -= 1
# the current character is within an escape sequence
elif escape:
-
- # the current character is m, which terminates the
- # escape sequence
+ rel_pos -= 1
if each_character == "m":
+ # the current character is m, which terminates the
+ # escape sequence
escape = False
+ # the current character is a space
+ elif each_character == " ":
+ last_abs_whitespace = abs_pos
+ last_rel_whitespace = rel_pos
+
# the current character is a newline, so reset the relative
- # position (start a new line)
+ # position too (start a new line)
elif each_character == "\n":
rel_pos = 0
- last_whitespace = abs_pos
-
- # the current character meets the requested maximum line width,
- # so we need to backtrack and find a space at which to wrap;
- # special care is taken to avoid an off-by-one in case the
- # current character is a double-width glyph
- elif each_character != "\r" and (
- rel_pos >= width or (
- rel_pos >= width - 1 and glyph_columns(
- each_character
- ) == 2
- )
- ):
+ last_abs_whitespace = abs_pos
+ last_rel_whitespace = rel_pos
- # it's always possible we landed on whitespace
- if unicodedata.category(each_character) in ("Cc", "Zs"):
- last_whitespace = abs_pos
+ # the current character meets the requested maximum line width, so we
+ # need to wrap unless the current word is wider than the terminal (in
+ # which case we let it do the wrapping instead)
+ if last_rel_whitespace != 0 and (rel_pos > width or (
+ rel_pos > width - 1 and glyph_columns(each_character) == 2)):
- # insert an eol in place of the space
- text = text[:last_whitespace] + "\r\n" + text[last_whitespace + 1:]
+ # insert an eol in place of the last space
+ text = (text[:last_abs_whitespace] + "\r\n" +
+ text[last_abs_whitespace + 1:])
# increase the absolute position because an eol is two
# characters but the space it replaced was only one
# now we're at the begining of a new line, plus the
# number of characters wrapped from the previous line
- rel_pos = 0
- for remaining_characters in text[last_whitespace:abs_pos]:
- rel_pos += glyph_columns(remaining_characters)
+ rel_pos -= last_rel_whitespace
+ last_rel_whitespace = 0
# 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":
rel_pos += glyph_columns(each_character)
- if unicodedata.category(each_character) in ("Cc", "Zs"):
- last_whitespace = abs_pos
+ if each_character in (" ", "\n"):
+ last_abs_whitespace = abs_pos
+ last_rel_whitespace = rel_pos
# increase the absolute position for every character
abs_pos += 1
replacement = replacement[:-2]
else:
replacement = ""
- log("Couldn't read included " + incfile + " file.", 6)
+ log("Couldn't read included " + incfile + " file.", 7)
# if we get here, log and replace it with null
else:
user.pulse()
# add an element for counters if it doesn't exist
- if "counters" not in universe.categories["internal"]:
- universe.categories["internal"]["counters"] = Element(
- "internal:counters", universe, old_style=True
- )
+ if "counters" not in universe.groups.get("internal", {}):
+ Element("internal.counters", universe)
# update the log every now and then
- if not universe.categories["internal"]["counters"].get("mark"):
+ if not universe.groups["internal"]["counters"].get("mark"):
log(str(len(universe.userlist)) + " connection(s)")
- universe.categories["internal"]["counters"].set(
- "mark", universe.categories["internal"]["time"].get(
- "frequency_log"
- )
+ universe.groups["internal"]["counters"].set(
+ "mark", universe.contents["mudpy.timing"].get("status")
)
else:
- universe.categories["internal"]["counters"].set(
- "mark", universe.categories["internal"]["counters"].get(
+ universe.groups["internal"]["counters"].set(
+ "mark", universe.groups["internal"]["counters"].get(
"mark"
) - 1
)
# periodically save everything
- if not universe.categories["internal"]["counters"].get("save"):
+ if not universe.groups["internal"]["counters"].get("save"):
universe.save()
- universe.categories["internal"]["counters"].set(
- "save", universe.categories["internal"]["time"].get(
- "frequency_save"
- )
+ universe.groups["internal"]["counters"].set(
+ "save", universe.contents["mudpy.timing"].get("save")
)
else:
- universe.categories["internal"]["counters"].set(
- "save", universe.categories["internal"]["counters"].get(
+ universe.groups["internal"]["counters"].set(
+ "save", universe.groups["internal"]["counters"].get(
"save"
) - 1
)
# pause for a configurable amount of time (decimal seconds)
- time.sleep(universe.categories["internal"]
- ["time"].get("increment"))
+ time.sleep(universe.contents["mudpy.timing"].get("increment"))
# increase the elapsed increment counter
- universe.categories["internal"]["counters"].set(
- "elapsed", universe.categories["internal"]["counters"].get(
+ universe.groups["internal"]["counters"].set(
+ "elapsed", universe.groups["internal"]["counters"].get(
"elapsed", 0
) + 1
)
def reload_data():
"""Reload all relevant objects."""
- for user in universe.userlist[:]:
- user.reload()
- for element in universe.contents.values():
- if element.origin.is_writeable():
- element.reload()
+ universe.save()
+ old_userlist = universe.userlist[:]
+ for element in list(universe.contents.values()):
+ element.destroy()
universe.load()
+ for user in old_userlist:
+ user.reload()
def check_for_connection(listening_socket):
return None
# note that we got one
- log("Connection from " + address[0], 2)
+ log("New connection from %s." % address[0], 2)
# disable blocking so we can proceed whether or not we can send/receive
connection.setblocking(0)
# create a new user object
user = User()
+ log("Instantiated %s for %s." % (user, address[0]), 0)
# associate this connection with it
user.connection = connection
def menu_echo_on(state):
"""True if echo is on, false if it is off."""
- return universe.categories["menu"][state].get("echo", True)
+ return universe.groups["menu"][state].get("echo", True)
def get_echo_message(state):
def get_default_menu_choice(state):
"""Return the default choice for a menu."""
- return universe.categories["menu"][state].get("default")
+ return universe.groups["menu"][state].get("default")
def get_formatted_default_menu_choice(state):
# try to get an error message matching the condition
# and current state
- description = universe.categories[
+ description = universe.groups[
"menu"][state].get("error_" + error)
if not description:
description = "That is not a valid choice..."
else:
# try to get a menu description for the current state
- description = universe.categories["menu"][state].get("description")
+ description = universe.groups["menu"][state].get("description")
# return the description or error message
if description:
def get_menu_prompt(state):
"""Try to get a prompt, if it was defined."""
- prompt = universe.categories["menu"][state].get("prompt")
+ prompt = universe.groups["menu"][state].get("prompt")
if prompt:
prompt += " "
return prompt
def get_menu_choices(user):
"""Return a dict of choice:meaning."""
- menu = universe.categories["menu"][user.state]
+ menu = universe.groups["menu"][user.state]
create_choices = menu.get("create")
if create_choices:
choices = eval(create_choices)
creates = {}
for facet in menu.facets():
if facet.startswith("demand_") and not eval(
- universe.categories["menu"][user.state].get(facet)
+ universe.groups["menu"][user.state].get(facet)
):
ignores.append(facet.split("_", 2)[1])
elif facet.startswith("create_"):
def get_menu_branches(state):
"""Return a dict of choice:branch."""
branches = {}
- for facet in universe.categories["menu"][state].facets():
+ for facet in universe.groups["menu"][state].facets():
if facet.startswith("branch_"):
branches[
facet.split("_", 2)[1]
- ] = universe.categories["menu"][state].get(facet)
+ ] = universe.groups["menu"][state].get(facet)
return branches
def get_default_branch(state):
"""Return the default branch."""
- return universe.categories["menu"][state].get("branch")
+ return universe.groups["menu"][state].get("branch")
def get_choice_branch(user, choice):
def get_menu_actions(state):
"""Return a dict of choice:branch."""
actions = {}
- for facet in universe.categories["menu"][state].facets():
+ for facet in universe.groups["menu"][state].facets():
if facet.startswith("action_"):
actions[
facet.split("_", 2)[1]
- ] = universe.categories["menu"][state].get(facet)
+ ] = universe.groups["menu"][state].get(facet)
return actions
def get_default_action(state):
"""Return the default action."""
- return universe.categories["menu"][state].get("action")
+ return universe.groups["menu"][state].get("action")
def get_choice_action(user, choice):
user.error = "bad_name"
# if that account exists, time to request a password
- elif name in universe.categories["account"]:
- user.account = universe.categories["account"][name]
+ elif name in universe.groups.get("account", {}):
+ user.account = universe.groups["account"][name]
user.state = "checking_password"
# otherwise, this could be a brand new user
else:
- user.account = Element("account:" + name, universe, old_style=True)
+ user.account = Element("account.%s" % name, universe)
user.account.set("name", name)
log("New user: " + name, 2)
user.state = "checking_new_account_name"
# get the next waiting line of input
input_data = user.input_queue.pop(0)
+ if "mudpy.limit" in universe.contents:
+ max_password_tries = universe.contents["mudpy.limit"].get(
+ "password_tries", 3)
+ else:
+ max_password_tries = 3
+
# does the hashed input equal the stored hash?
if mudpy.password.verify(input_data, user.account.get("passhash")):
user.state = "main_utility"
# if at first your hashes don't match, try, try again
- elif user.password_tries < universe.contents["mudpy.limit"].get(
- "password_tries") - 1:
+ elif user.password_tries < max_password_tries - 1:
user.password_tries += 1
user.error = "incorrect"
# get the next waiting line of input
input_data = user.input_queue.pop(0)
+ if "mudpy.limit" in universe.contents:
+ max_password_tries = universe.contents["mudpy.limit"].get(
+ "password_tries", 3)
+ else:
+ max_password_tries = 3
+
# make sure the password is strong--at least one upper, one lower and
# one digit, seven or more characters in length
if len(input_data) > 6 and len(
user.state = "verifying_new_password"
# the password was weak, try again if you haven't tried too many times
- elif user.password_tries < universe.contents["mudpy.limit"].get(
- "password_tries") - 1:
+ elif user.password_tries < max_password_tries - 1:
user.password_tries += 1
user.error = "weak"
# get the next waiting line of input
input_data = user.input_queue.pop(0)
+ if "mudpy.limit" in universe.contents:
+ max_password_tries = universe.contents["mudpy.limit"].get(
+ "password_tries", 3)
+ else:
+ max_password_tries = 3
+
# hash the input and match it to storage
if mudpy.password.verify(input_data, user.account.get("passhash")):
user.authenticate()
# go back to entering the new password as long as you haven't tried
# too many times
- elif user.password_tries < universe.contents["mudpy.limit"].get(
- "password_tries") - 1:
+ elif user.password_tries < max_password_tries - 1:
user.password_tries += 1
user.error = "differs"
user.state = "entering_new_password"
command_name = command_name.lower()
# the command matches a command word for which we have data
- if command_name in universe.categories["command"]:
- command = universe.categories["command"][command_name]
+ if command_name in universe.groups["command"]:
+ command = universe.groups["command"][command_name]
else:
command = None
# otherwise, give an error
elif command_name:
- command_error(actor, input_data)
+ mudpy.command.error(actor, input_data)
# if no input, just idle back with a prompt
else:
user.send("", just_prompt=True)
-def command_halt(actor, parameters):
- """Halt the world."""
- if actor.owner:
-
- # see if there's a message or use a generic one
- if parameters:
- message = "Halting: " + parameters
- else:
- message = "User " + actor.owner.account.get(
- "name"
- ) + " halted the world."
-
- # let everyone know
- broadcast(message, add_prompt=False)
- log(message, 8)
-
- # set a flag to terminate the world
- universe.terminate_flag = True
-
-
-def command_reload(actor):
- """Reload all code modules, configs and data."""
- if actor.owner:
-
- # let the user know and log
- actor.send("Reloading all code modules, configs and data.")
- log(
- "User " +
- actor.owner.account.get("name") + " reloaded the world.",
- 8
- )
-
- # set a flag to reload
- universe.reload_flag = True
-
-
-def command_quit(actor):
- """Leave the world and go back to the main menu."""
- if actor.owner:
- actor.owner.state = "main_utility"
- actor.owner.deactivate_avatar()
-
-
-def command_help(actor, parameters):
- """List available commands and provide help for commands."""
-
- # did the user ask for help on a specific command word?
- if parameters and actor.owner:
-
- # is the command word one for which we have data?
- if parameters in universe.categories["command"]:
- command = universe.categories["command"][parameters]
- else:
- command = None
-
- # only for allowed commands
- if actor.can_run(command):
-
- # add a description if provided
- description = command.get("description")
- if not description:
- description = "(no short description provided)"
- if command.get("administrative"):
- output = "$(red)"
- else:
- output = "$(grn)"
- output += parameters + "$(nrm) - " + description + "$(eol)$(eol)"
-
- # add the help text if provided
- help_text = command.get("help")
- if not help_text:
- help_text = "No help is provided for this command."
- output += help_text
-
- # list related commands
- see_also = command.get("see_also")
- if see_also:
- really_see_also = ""
- for item in see_also:
- if item in universe.categories["command"]:
- command = universe.categories["command"][item]
- if actor.can_run(command):
- if really_see_also:
- really_see_also += ", "
- if command.get("administrative"):
- really_see_also += "$(red)"
- else:
- really_see_also += "$(grn)"
- really_see_also += item + "$(nrm)"
- if really_see_also:
- output += "$(eol)$(eol)See also: " + really_see_also
-
- # no data for the requested command word
- else:
- output = "That is not an available command."
-
- # no specific command word was indicated
- else:
-
- # give a sorted list of commands with descriptions if provided
- output = "These are the commands available to you:$(eol)$(eol)"
- sorted_commands = list(universe.categories["command"].keys())
- sorted_commands.sort()
- for item in sorted_commands:
- command = universe.categories["command"][item]
- if actor.can_run(command):
- description = command.get("description")
- if not description:
- description = "(no short description provided)"
- if command.get("administrative"):
- output += " $(red)"
- else:
- output += " $(grn)"
- output += item + "$(nrm) - " + description + "$(eol)"
- output += ("$(eol)Enter \"help COMMAND\" for help on a command "
- "named \"COMMAND\".")
-
- # send the accumulated output to the user
- actor.send(output)
-
-
-def command_move(actor, parameters):
- """Move the avatar in a given direction."""
- if parameters in universe.contents[actor.get("location")].portals():
- actor.move_direction(parameters)
- else:
- actor.send("You cannot go that way.")
-
-
-def command_look(actor, parameters):
- """Look around."""
- if parameters:
- actor.send("You can't look at or in anything yet.")
- else:
- actor.look_at(actor.get("location"))
-
-
-def command_say(actor, parameters):
- """Speak to others in the same area."""
-
- # check for replacement macros and escape them
- parameters = escape_macros(parameters)
-
- # if the message is wrapped in quotes, remove them and leave contents
- # intact
- if parameters.startswith("\"") and parameters.endswith("\""):
- message = parameters[1:-1]
- literal = True
-
- # otherwise, get rid of stray quote marks on the ends of the message
- else:
- message = parameters.strip("\"'`")
- literal = False
-
- # the user entered a message
- if message:
-
- # match the punctuation used, if any, to an action
- actions = universe.contents["mudpy.linguistic"].get(
- "actions"
- )
- default_punctuation = (
- universe.contents["mudpy.linguistic"].get(
- "default_punctuation"))
- action = ""
-
- # reverse sort punctuation options so the longest match wins
- for mark in sorted(actions.keys(), reverse=True):
- if not literal and message.endswith(mark):
- action = actions[mark]
- break
-
- # add punctuation if needed
- if not action:
- action = actions[default_punctuation]
- if message and not (
- literal or unicodedata.category(message[-1]) == "Po"
- ):
- message += default_punctuation
-
- # failsafe checks to avoid unwanted reformatting and null strings
- if message and not literal:
-
- # decapitalize the first letter to improve matching
- message = message[0].lower() + message[1:]
-
- # iterate over all words in message, replacing typos
- typos = universe.contents["mudpy.linguistic"].get(
- "typos"
- )
- words = message.split()
- for index in range(len(words)):
- word = words[index]
- while unicodedata.category(word[0]) == "Po":
- word = word[1:]
- while unicodedata.category(word[-1]) == "Po":
- word = word[:-1]
- if word in typos.keys():
- words[index] = words[index].replace(word, typos[word])
- message = " ".join(words)
-
- # capitalize the first letter
- message = message[0].upper() + message[1:]
-
- # tell the area
- if message:
- actor.echo_to_location(
- actor.get("name") + " " + action + "s, \"" + message + "\""
- )
- actor.send("You " + action + ", \"" + message + "\"")
-
- # there was no message
- else:
- actor.send("What do you want to say?")
-
-
-def command_chat(actor):
- """Toggle chat mode."""
- mode = actor.get("mode")
- if not mode:
- actor.set("mode", "chat")
- actor.send("Entering chat mode (use $(grn)!chat$(nrm) to exit).")
- elif mode == "chat":
- actor.remove_facet("mode")
- actor.send("Exiting chat mode.")
- else:
- actor.send("Sorry, but you're already busy with something else!")
-
-
-def command_show(actor, parameters):
- """Show program data."""
- message = ""
- arguments = parameters.split()
- if not parameters:
- message = "What do you want to show?"
- elif arguments[0] == "time":
- message = universe.categories["internal"]["counters"].get(
- "elapsed"
- ) + " increments elapsed since the world was created."
- elif arguments[0] == "categories":
- message = "These are the element categories:$(eol)"
- categories = list(universe.categories.keys())
- categories.sort()
- for category in categories:
- message += "$(eol) $(grn)" + category + "$(nrm)"
- elif arguments[0] == "files":
- message = "These are the current files containing the universe:$(eol)"
- filenames = list(universe.files.keys())
- filenames.sort()
- for filename in filenames:
- if universe.files[filename].is_writeable():
- status = "rw"
- else:
- status = "ro"
- message += ("$(eol) $(red)(" + status + ") $(grn)" + filename
- + "$(nrm)")
- elif arguments[0] == "category":
- if len(arguments) != 2:
- message = "You must specify one category."
- elif arguments[1] in universe.categories:
- message = ("These are the elements in the \"" + arguments[1]
- + "\" category:$(eol)")
- elements = [
- (
- universe.categories[arguments[1]][x].key
- ) for x in universe.categories[arguments[1]].keys()
- ]
- elements.sort()
- for element in elements:
- message += "$(eol) $(grn)" + element + "$(nrm)"
- else:
- message = "Category \"" + arguments[1] + "\" does not exist."
- elif arguments[0] == "file":
- if len(arguments) != 2:
- message = "You must specify one file."
- elif arguments[1] in universe.files:
- message = ("These are the elements in the \"" + arguments[1]
- + "\" file:$(eol)")
- elements = universe.files[arguments[1]].data.keys()
- elements.sort()
- for element in elements:
- message += "$(eol) $(grn)" + element + "$(nrm)"
- else:
- message = "Category \"" + arguments[1] + "\" does not exist."
- elif arguments[0] == "element":
- if len(arguments) != 2:
- message = "You must specify one element."
- elif arguments[1].strip(".") in universe.contents:
- element = universe.contents[arguments[1].strip(".")]
- message = ("These are the properties of the \"" + arguments[1]
- + "\" element (in \"" + element.origin.filename
- + "\"):$(eol)")
- facets = element.facets()
- for facet in sorted(facets):
- if element.old_style:
- message += ("$(eol) $(grn)%s: $(red)%s$(nrm)" %
- (facet, escape_macros(element.get(facet))))
- else:
- message += ("$(eol) $(grn)%s: $(red)%s$(nrm)" %
- (facet, str(facets[facet])))
- else:
- message = "Element \"" + arguments[1] + "\" does not exist."
- elif arguments[0] == "result":
- if len(arguments) < 2:
- message = "You need to specify an expression."
- else:
- try:
- message = repr(eval(" ".join(arguments[1:])))
- except Exception as e:
- message = ("$(red)Your expression raised an exception...$(eol)"
- "$(eol)$(bld)%s$(nrm)" % e)
- elif arguments[0] == "log":
- if len(arguments) == 4:
- if re.match("^\d+$", arguments[3]) and int(arguments[3]) >= 0:
- stop = int(arguments[3])
- else:
- stop = -1
- else:
- stop = 0
- if len(arguments) >= 3:
- if re.match("^\d+$", arguments[2]) and int(arguments[2]) > 0:
- start = int(arguments[2])
- else:
- start = -1
- else:
- start = 10
- if len(arguments) >= 2:
- if (re.match("^\d+$", arguments[1])
- and 0 <= int(arguments[1]) <= 9):
- level = int(arguments[1])
- else:
- level = -1
- elif 0 <= actor.owner.account.get("loglevel", 0) <= 9:
- level = actor.owner.account.get("loglevel", 0)
- else:
- level = 1
- if level > -1 and start > -1 and stop > -1:
- message = get_loglines(level, start, stop)
- else:
- message = ("When specified, level must be 0-9 (default 1), "
- "start and stop must be >=1 (default 10 and 1).")
- else:
- message = "I don't know what \"" + parameters + "\" is."
- actor.send(message)
-
-
-def command_create(actor, parameters):
- """Create an element if it does not exist."""
- if not parameters:
- message = "You must at least specify an element to create."
- elif not actor.owner:
- message = ""
- else:
- arguments = parameters.split()
- if len(arguments) == 1:
- arguments.append("")
- if len(arguments) == 2:
- element, filename = arguments
- if element in universe.contents:
- message = "The \"" + element + "\" element already exists."
- else:
- message = ("You create \"" +
- element + "\" within the universe.")
- logline = actor.owner.account.get(
- "name"
- ) + " created an element: " + element
- if filename:
- logline += " in file " + filename
- if filename not in universe.files:
- message += (
- " Warning: \"" + filename + "\" is not yet "
- "included in any other file and will not be read "
- "on startup unless this is remedied.")
- Element(element, universe, filename, old_style=True)
- log(logline, 6)
- elif len(arguments) > 2:
- message = "You can only specify an element and a filename."
- actor.send(message)
-
-
-def command_destroy(actor, parameters):
- """Destroy an element if it exists."""
- if actor.owner:
- if not parameters:
- message = "You must specify an element to destroy."
- else:
- if parameters not in universe.contents:
- message = "The \"" + parameters + "\" element does not exist."
- else:
- universe.contents[parameters].destroy()
- message = ("You destroy \"" + parameters
- + "\" within the universe.")
- log(
- actor.owner.account.get(
- "name"
- ) + " destroyed an element: " + parameters,
- 6
- )
- actor.send(message)
-
-
-def command_set(actor, parameters):
- """Set a facet of an element."""
- if not parameters:
- message = "You must specify an element, a facet and a value."
- else:
- arguments = parameters.split(" ", 2)
- if len(arguments) == 1:
- message = ("What facet of element \"" + arguments[0]
- + "\" would you like to set?")
- elif len(arguments) == 2:
- message = ("What value would you like to set for the \"" +
- arguments[1] + "\" facet of the \"" + arguments[0]
- + "\" element?")
- else:
- element, facet, value = arguments
- if element not in universe.contents:
- message = "The \"" + element + "\" element does not exist."
- else:
- universe.contents[element].set(facet, value)
- message = ("You have successfully (re)set the \"" + facet
- + "\" facet of element \"" + element
- + "\". Try \"show element " +
- element + "\" for verification.")
- actor.send(message)
-
-
-def command_delete(actor, parameters):
- """Delete a facet from an element."""
- if not parameters:
- message = "You must specify an element and a facet."
- else:
- arguments = parameters.split(" ")
- if len(arguments) == 1:
- message = ("What facet of element \"" + arguments[0]
- + "\" would you like to delete?")
- elif len(arguments) != 2:
- message = "You may only specify an element and a facet."
- else:
- element, facet = arguments
- if element not in universe.contents:
- message = "The \"" + element + "\" element does not exist."
- elif facet not in universe.contents[element].facets():
- message = ("The \"" + element + "\" element has no \"" + facet
- + "\" facet.")
- else:
- universe.contents[element].remove_facet(facet)
- message = ("You have successfully deleted the \"" + facet
- + "\" facet of element \"" + element
- + "\". Try \"show element " +
- element + "\" for verification.")
- actor.send(message)
-
-
-def command_error(actor, input_data):
- """Generic error for an unrecognized command word."""
-
- # 90% of the time use a generic error
- if random.randrange(10):
- message = "I'm not sure what \"" + input_data + "\" means..."
-
- # 10% of the time use the classic diku error
- else:
- message = "Arglebargle, glop-glyf!?!"
-
- # send the error message
- actor.send(message)
-
-
def daemonize(universe):
"""Fork and disassociate from everything."""
# only if this is what we're configured to do
- if universe.contents["mudpy.process"].get("daemon"):
+ if "mudpy.process" in universe.contents and universe.contents[
+ "mudpy.process"].get("daemon"):
# log before we start forking around, so the terminal gets the message
log("Disassociating from the controlling terminal.")
"""Write a file containing the current process ID."""
pid = str(os.getpid())
log("Process ID: " + pid)
- file_name = universe.contents["mudpy.process"].get("pidfile")
+ if "mudpy.process" in universe.contents:
+ file_name = universe.contents["mudpy.process"].get("pidfile", "")
+ else:
+ file_name = ""
if file_name:
if not os.path.isabs(file_name):
file_name = os.path.join(universe.startdir, file_name)
+ os.makedirs(os.path.dirname(file_name), exist_ok=True)
file_descriptor = codecs.open(file_name, "w", "utf-8")
file_descriptor.write(pid + "\n")
file_descriptor.flush()
def remove_pidfile(universe):
"""Remove the file containing the current process ID."""
- file_name = universe.contents["mudpy.process"].get("pidfile")
+ if "mudpy.process" in universe.contents:
+ file_name = universe.contents["mudpy.process"].get("pidfile", "")
+ else:
+ file_name = ""
if file_name:
if not os.path.isabs(file_name):
file_name = os.path.join(universe.startdir, file_name)
log(*logline)
universe.setup_loglines = []
- # log an initial message
- log("Started mudpy with command line: " + " ".join(sys.argv))
-
# fork and disassociate
daemonize(universe)
# make the pidfile
create_pidfile(universe)
+ # load and store diagnostic info
+ universe.versions = mudpy.version.Versions("mudpy")
+
+ # log startup diagnostic messages
+ log("On %s at %s" % (universe.versions.python_version, sys.executable), 1)
+ log("Import path: %s" % ", ".join(sys.path), 1)
+ log("Installed dependencies: %s" % universe.versions.dependencies_text, 1)
+ log("Other python packages: %s" % universe.versions.environment_text, 1)
+ log("Started %s with command line: %s" % (
+ universe.versions.version, " ".join(sys.argv)), 1)
+
# pass the initialized universe back
return universe
projects / mudpy.git / blobdiff