"""Miscellaneous functions for the mudpy engine."""
-# Copyright (c) 2004-2016 Jeremy Stanley <fungi@yuggoth.org>. Permission
+# Copyright (c) 2004-2017 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.
"""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)
+ Element(self.key, self.universe, self.origin)
del(self)
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.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)
- 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]
+ 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(r"""^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(",")]
+ for x in self.key.split(".")[-1].split(",")]
offsets = dict(
(x,
self.universe.contents["mudpy.movement.%s" % x].get("vector")
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()
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 = []
del self.files[data_filename]
# start loading from the initial file
- mudpy.data.DataFile(self.filename, self)
+ mudpy.data.Data(self.filename, self)
+
+ # 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", {}))
+
+ # 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])
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:
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 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 ("mudpy.limit" in universe.contents and self.account.subkey in
universe.contents["mudpy.limit"].get("admins")):
- self.account.set("administrator", "True")
+ self.account.set("administrator", True)
+ log("Administrator %s authenticated." %
+ self.account.get("name"), 2)
+ else:
+ # log("User %s authenticated." % self.account.get("name"), 2)
+ log("User %s authenticated." % self.account.subkey, 2)
def show_menu(self):
"""Send the user their current menu."""
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)
def delete_avatar(self, avatar):
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 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"
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
log(
"User " +
actor.owner.account.get("name") + " reloaded the world.",
- 8
+ 6
)
# set a flag to reload
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]
+ if parameters in universe.groups["command"]:
+ command = universe.groups["command"][parameters]
else:
command = None
if see_also:
really_see_also = ""
for item in see_also:
- if item in universe.categories["command"]:
- command = universe.categories["command"][item]
+ if item in universe.groups["command"]:
+ command = universe.groups["command"][item]
if actor.can_run(command):
if really_see_also:
really_see_also += ", "
# 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 = list(universe.groups["command"].keys())
sorted_commands.sort()
for item in sorted_commands:
- command = universe.categories["command"][item]
+ command = universe.groups["command"][item]
if actor.can_run(command):
description = command.get("description")
if not description:
if not parameters:
message = "What do you want to show?"
elif arguments[0] == "time":
- message = universe.categories["internal"]["counters"].get(
+ message = universe.groups["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] == "groups":
+ message = "These are the element groups:$(eol)"
+ groups = list(universe.groups.keys())
+ groups.sort()
+ for group in groups:
+ message += "$(eol) $(grn)" + group + "$(nrm)"
elif arguments[0] == "files":
message = "These are the current files containing the universe:$(eol)"
- filenames = list(universe.files.keys())
- filenames.sort()
+ filenames = sorted(universe.files)
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":
+ message += ("$(eol) $(red)(%s) $(grn)%s$(nrm)" %
+ (status, filename))
+ if universe.files[filename].flags:
+ message += (" $(yel)[%s]$(nrm)" %
+ ",".join(universe.files[filename].flags))
+ elif arguments[0] == "group":
if len(arguments) != 2:
- message = "You must specify one category."
- elif arguments[1] in universe.categories:
+ message = "You must specify one group."
+ elif arguments[1] in universe.groups:
message = ('These are the elements in the "' + arguments[1]
- + '" category:$(eol)')
+ + '" group:$(eol)')
elements = [
(
- universe.categories[arguments[1]][x].key
- ) for x in universe.categories[arguments[1]].keys()
+ universe.groups[arguments[1]][x].key
+ ) for x in universe.groups[arguments[1]].keys()
]
elements.sort()
for element in elements:
message += "$(eol) $(grn)" + element + "$(nrm)"
else:
- message = 'Category "' + arguments[1] + '" does not exist.'
+ message = 'Group "' + 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]
+ message = ('These are the nodes in the "' + arguments[1]
+ '" file:$(eol)')
- elements = universe.files[arguments[1]].data.keys()
- elements.sort()
+ elements = sorted(universe.files[arguments[1]].data)
for element in elements:
message += "$(eol) $(grn)" + element + "$(nrm)"
else:
- message = 'Category "' + arguments[1] + '" does not exist.'
+ message = 'File "%s" does not exist.' % arguments[1]
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
+ + '" element (in "' + element.origin.source
+ '"):$(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])))
+ message += ("$(eol) $(grn)%s: $(red)%s$(nrm)" %
+ (facet, str(facets[facet])))
else:
message = 'Element "' + arguments[1] + '" does not exist.'
elif arguments[0] == "result":
' 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)
+ Element(element, universe, filename)
log(logline, 6)
elif len(arguments) > 2:
message = "You can only specify an element and a filename."
else:
try:
universe.contents[element].set(facet, value)
+ except PermissionError:
+ message = ('The "%s" element is kept in read-only file '
+ '"%s" and cannot be altered.' %
+ (element, universe.contents[
+ element].origin.source))
except ValueError:
message = ('Value "%s" of type "%s" cannot be coerced '
'to the correct datatype for facet "%s".' %
projects / mudpy.git / blobdiff