class Element:
"""An element of the universe."""
- def __init__(self, key, universe, origin=""):
- """Default values for the in-memory element variables."""
+ def __init__(self, key, universe, filename=None):
+ """Set up a new element."""
+
+ # not owned by a user by default (used for avatars)
self.owner = None
+
+ # no contents in here by default
self.contents = {}
+
+ # an event queue for the element
+ self.events = {}
+
+ # keep track of our key name
self.key = key
+
+ # 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.subkey = self.key
if not self.category in universe.categories: self.category = "other"
universe.categories[self.category][self.subkey] = self
- self.origin = origin
- if not self.origin: self.origin = universe.default_origins[self.category]
- if not isabs(self.origin):
- self.origin = abspath(self.origin)
+
+ # get an appropriate filename for the origin
+ if not filename: filename = universe.default_origins[self.category]
+ if not isabs(filename): filename = abspath(filename)
+
+ # add the file if it doesn't exist yet
+ if not filename in universe.files: DataFile(filename, universe)
+
+ # record a pointer to the origin file
+ self.origin = universe.files[filename]
+
+ # add a data section to the origin if necessary
+ if not self.origin.data.has_section(self.key):
+ self.origin.data.add_section(self.key)
+
+ # add this element to the universe contents
universe.contents[self.key] = self
- if not self.origin in universe.files:
- DataFile(self.origin, universe)
- if not universe.files[self.origin].data.has_section(self.key):
- universe.files[self.origin].data.add_section(self.key)
+
def destroy(self):
"""Remove an element from the universe and destroy it."""
- log("Destroying: " + self.key + ".", 2)
- universe.files[self.origin].data.remove_section(self.key)
+ self.origin.data.remove_section(self.key)
del universe.categories[self.category][self.subkey]
del universe.contents[self.key]
del self
- def delete(self, facet):
- """Delete a facet from the element."""
- if universe.files[self.origin].data.has_option(self.key, facet):
- universe.files[self.origin].data.remove_option(self.key, facet)
def facets(self):
"""Return a list of non-inherited facets for this element."""
- if self.key in universe.files[self.origin].data.sections():
- return universe.files[self.origin].data.options(self.key)
+ if self.key in self.origin.data.sections():
+ return self.origin.data.options(self.key)
else: return []
def has_facet(self, facet):
"""Return whether the non-inherited facet exists."""
return facet in self.facets()
def remove_facet(self, facet):
"""Remove a facet from the element."""
- if self.has_facet(facet): universe.files[self.origin].data.remove_option(self.key, facet)
+ if self.has_facet(facet):
+ self.origin.data.remove_option(self.key, facet)
+ self.origin.modified = True
def ancestry(self):
"""Return a list of the element's inheritance lineage."""
if self.has_facet("inherit"):
def get(self, facet, default=None):
"""Retrieve values."""
if default is None: default = ""
- if universe.files[self.origin].data.has_option(self.key, facet):
- return universe.files[self.origin].data.get(self.key, facet)
+ if self.origin.data.has_option(self.key, facet):
+ return self.origin.data.get(self.key, facet)
elif self.has_facet("inherit"):
for ancestor in self.ancestry():
if universe.contents[ancestor].has_facet(facet):
def getboolean(self, facet, default=None):
"""Retrieve values as boolean type."""
if default is None: default=False
- if universe.files[self.origin].data.has_option(self.key, facet):
- return universe.files[self.origin].data.getboolean(self.key, facet)
+ if self.origin.data.has_option(self.key, facet):
+ return self.origin.data.getboolean(self.key, facet)
elif self.has_facet("inherit"):
for ancestor in self.ancestry():
if universe.contents[ancestor].has_facet(facet):
def getint(self, facet, default=None):
"""Return values as int/long type."""
if default is None: default = 0
- if universe.files[self.origin].data.has_option(self.key, facet):
- return universe.files[self.origin].data.getint(self.key, facet)
+ if self.origin.data.has_option(self.key, facet):
+ return self.origin.data.getint(self.key, facet)
elif self.has_facet("inherit"):
for ancestor in self.ancestry():
if universe.contents[ancestor].has_facet(facet):
def getfloat(self, facet, default=None):
"""Return values as float type."""
if default is None: default = 0.0
- if universe.files[self.origin].data.has_option(self.key, facet):
- return universe.files[self.origin].data.getfloat(self.key, facet)
+ if self.origin.data.has_option(self.key, facet):
+ return self.origin.data.getfloat(self.key, facet)
elif self.has_facet("inherit"):
for ancestor in self.ancestry():
if universe.contents[ancestor].has_facet(facet):
else: return default
def set(self, facet, value):
"""Set values."""
- if type(value) is long: value = str(value)
- elif not type(value) is str: value = repr(value)
- universe.files[self.origin].data.set(self.key, facet, value)
+ if not self.has_facet(facet) or not self.get(facet) == value:
+ if type(value) is long: value = str(value)
+ elif not type(value) is str: value = repr(value)
+ self.origin.data.set(self.key, facet, value)
+ self.origin.modified = True
def append(self, facet, value):
"""Append value tp a list."""
if type(value) is long: value = str(value)
newlist = self.getlist(facet)
newlist.append(value)
self.set(facet, newlist)
+
+ def new_event(self, action, when=None):
+ """Create, attach and enqueue an event element."""
+
+ # if when isn't specified, that means now
+ if not when: when = universe.get_time()
+
+ # events are elements themselves
+ event = Element("event:" + self.key + ":" + counter)
+
def send(self, message, eol="$(eol)"):
"""Convenience method to pass messages to an owner."""
if self.owner: self.owner.send(message, eol)
+
+ 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 universe.categories["command"].values(): result = False
+
+ # avatars of administrators can run any command
+ elif self.owner and self.owner.account.getboolean("administrator"): result = True
+
+ # everyone can run non-administrative commands
+ elif not command.getboolean("administrative"): result = True
+
+ # otherwise the command cannot be run by this actor
+ else: result = False
+
+ # pass back the result
+ return result
+
def go_to(self, location):
"""Relocate the element to a specific location."""
current = self.get("location")
class DataFile:
"""A file containing universe elements."""
def __init__(self, filename, universe):
+ self.modified = False
self.data = RawConfigParser()
if access(filename, R_OK): self.data.read(filename)
self.filename = filename
if not isabs(include_file):
include_file = path_join(dirname(filename), include_file)
DataFile(include_file, universe)
+ def is_writeable(self):
+ """Returns True if the __control__ read_only is False."""
+ return not self.data.has_option("__control__", "read_only") or not self.data.getboolean("__control__", "read_only")
def save(self):
"""Write the data, if necessary."""
- # when there is content or the file exists, but is not read-only
- if ( self.data.sections() or exists(self.filename) ) and not ( self.data.has_option("__control__", "read_only") and self.data.getboolean("__control__", "read_only") ):
+ # when modified, writeable and has content or the file exists
+ if self.modified and self.is_writeable() and ( self.data.sections() or exists(self.filename) ):
# make parent directories if necessary
if not exists(dirname(self.filename)):
file_descriptor.flush()
file_descriptor.close()
+ # unset the modified flag
+ self.modified = False
+
class Universe:
"""The universe."""
def __init__(self, filename=""):
self.default_origins = {}
self.files = {}
self.private_files = []
- self.loglist = []
+ self.loglines = {}
+ self.pending_events_long = {}
+ self.pending_events_short = {}
self.userlist = []
self.terminate_world = False
self.reload_modules = False
# note that we're now ready for user connections
log("Waiting for connection(s)...")
+ def get_time(self):
+ """Convenience method to get the elapsed time counter."""
+ return self.categories["internal"]["counters"].getint("elapsed")
+
class User:
"""This is a connected user."""
# replace the input with our cleaned-up text
self.partial_input = text
- 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 universe.categories["command"].values(): result = False
-
- # administrators can run any command
- elif self.account.getboolean("administrator"): result = True
-
- # everyone can run non-administrative commands
- elif not command.getboolean("administrative"): result = True
-
- # otherwise the command cannot be run by this user
- else: result = False
-
- # pass back the result
- return result
-
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(): counter += 1
self.avatar = Element("actor:avatar:" + self.account.get("name") + ":" + str(counter), universe)
- self.avatar.append("inherit", "template:actor")
+ self.avatar.append("inherit", "archetype:avatar")
self.account.append("avatars", self.avatar.key)
def delete_avatar(self, avatar):
# add to the recent log list
for line in lines:
- while 0 < len(universe.loglist) >= max_log_lines: del universe.loglist[0]
- universe.loglist.append(timestamp + " " + line)
+ while 0 < len(universe.loglines) >= max_log_lines: del universe.loglines[0]
+ universe.loglines.append((level, timestamp + " " + line))
+
+def get_loglines(level, start, stop=0):
+ """Return a specific range of loglines filtered by level."""
+
+ # begin with a blank message
+ message = ""
+
+ # filter the log lines
+ loglines = filter(lambda x,y: x>=level, universe.loglines)
+
+ # we need this in several places
+ count = len(loglines)
+
+ # don't proceed if there are no lines
+ if count:
+
+ # can't start before the begining or at the end
+ if start > count: start = count
+ if start < 1: start = 1
+
+ # can't stop before we start
+ if stop >= start: stop = start - 1
+
+ # some preamble
+ message += "There are " + str(len(universe.loglist))
+ message += " log lines in memory and " + str(count)
+ message += " at or above level " + str(level) + "."
+ message += " The lines from " + str(stop)
+ message += " to " + str(start) + " are:$(eol)$(eol)"
+
+ # add the text from the selected lines
+ for line in loglines[-start:-stop]:
+ message += " " + line[1] + "$(eol)"
+
+ # pass it back
+ return message
def wrap_ansi_text(text, width):
"""Wrap text with arbitrary width while ignoring ANSI colors."""
else: command = None
# if it's allowed, do it
- if user.can_run(command): exec(command.get("action"))
+ if user.avatar.can_run(command): exec(command.get("action"))
# otherwise, give an error
- elif command_name: command_error(user, input_data)
+ elif command_name: command_error(user.avatar, input_data)
# if no input, just idle back with a prompt
else: user.send("", just_prompt=True)
-def command_halt(user, parameters):
+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 " + user.account.get("name") + " halted the world."
+ # 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)
+ # let everyone know
+ broadcast(message, add_prompt=False)
+ log(message, 8)
- # set a flag to terminate the world
- universe.terminate_world = True
+ # set a flag to terminate the world
+ universe.terminate_world = True
-def command_reload(user):
+def command_reload(actor):
"""Reload all code modules, configs and data."""
+ if actor.owner:
- # let the user know and log
- user.send("Reloading all code modules, configs and data.")
- log("User " + user.account.get("name") + " reloaded the world.", 8)
+ # 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_modules = True
+ # set a flag to reload
+ universe.reload_modules = True
-def command_quit(user):
+def command_quit(actor):
"""Leave the world and go back to the main menu."""
- user.deactivate_avatar()
- user.state = "main_utility"
+ if actor.owner:
+ actor.owner.deactivate_avatar()
+ actor.owner.state = "main_utility"
-def command_help(user, parameters):
+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:
+ if parameters and actor.owner:
# is the command word one for which we have data?
if parameters in universe.categories["command"]:
else: command = None
# only for allowed commands
- if user.can_run(command):
+ if actor.can_run(command):
# add a description if provided
description = command.get("description")
sorted_commands.sort()
for item in sorted_commands:
command = universe.categories["command"][item]
- if user.can_run(command):
+ if actor.can_run(command):
description = command.get("description")
if not description:
description = "(no short description provided)"
output += "$(eol)Enter \"help COMMAND\" for help on a command named \"COMMAND\"."
# send the accumulated output to the user
- user.send(output)
+ actor.send(output)
-def command_move(user, parameters):
+def command_move(actor, parameters):
"""Move the avatar in a given direction."""
- if parameters in universe.contents[user.avatar.get("location")].portals():
- user.avatar.move_direction(parameters)
- else: user.send("You cannot go that way.")
+ if parameters in universe.contents[actor.get("location")].portals():
+ actor.move_direction(parameters)
+ else: actor.send("You cannot go that way.")
-def command_look(user, parameters):
+def command_look(actor, parameters):
"""Look around."""
- if parameters: user.send("You look at or in anything yet.")
- else: user.avatar.look_at(user.avatar.get("location"))
+ if parameters: actor.send("You can't look at or in anything yet.")
+ else: actor.look_at(actor.get("location"))
-def command_say(user, parameters):
+def command_say(actor, parameters):
"""Speak to others in the same room."""
# check for replacement macros
- if replace_macros(user, parameters, True) != parameters:
- user.send("You cannot speak $_(replacement macros).")
+ if replace_macros(actor.owner, parameters, True) != parameters:
+ actor.send("You cannot speak $_(replacement macros).")
# the user entered a message
elif parameters:
message = message.replace(" " + word + " ", " " + word.capitalize() + " ")
# tell the room
- user.avatar.echo_to_location(user.avatar.get("name") + " " + action + "s, \"" + message + "\"")
- user.send("You " + action + ", \"" + message + "\"")
+ actor.echo_to_location(actor.get("name") + " " + action + "s, \"" + message + "\"")
+ actor.send("You " + action + ", \"" + message + "\"")
# there was no message
else:
- user.send("What do you want to say?")
+ actor.send("What do you want to say?")
-def command_show(user, parameters):
+def command_show(actor, parameters):
"""Show program data."""
message = ""
if parameters.find(" ") < 1:
elif arguments[0] == "log":
if match("^\d+$", arguments[1]) and int(arguments[1]) > 0:
linecount = int(arguments[1])
- if linecount > len(universe.loglist): linecount = len(universe.loglist)
+ if linecount > len(universe.loglines): linecount = len(universe.loglist)
message = "There are " + str(len(universe.loglist)) + " log lines in memory."
message += " The most recent " + str(linecount) + " lines are:$(eol)$(eol)"
for line in universe.loglist[-linecount:]:
if not message:
if parameters: message = "I don't know what \"" + parameters + "\" is."
else: message = "What do you want to show?"
- user.send(message)
+ actor.send(message)
-def command_create(user, parameters):
+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 element in universe.contents: message = "The \"" + element + "\" element already exists."
else:
message = "You create \"" + element + "\" within the universe."
- logline = user.account.get("name") + " created an element: " + element
+ logline = actor.owner.account.get("name") + " created an element: " + element
if filename:
logline += " in file " + filename
if filename not in universe.files:
Element(element, universe, filename)
log(logline, 6)
elif len(arguments) > 2: message = "You can only specify an element and a filename."
- user.send(message)
+ actor.send(message)
-def command_destroy(user, parameters):
+def command_destroy(actor, parameters):
"""Destroy an element if it exists."""
- 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."
+ if actor.owner:
+ if not parameters: message = "You must specify an element to destroy."
else:
- universe.contents[parameters].destroy()
- message = "You destroy \"" + parameters + "\" within the universe."
- log(user.account.get("name") + " destroyed an element: " + parameters, 6)
- user.send(message)
+ 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(user, parameters):
+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:
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."
- user.send(message)
+ actor.send(message)
-def command_delete(user, parameters):
+def command_delete(actor, parameters):
"""Delete a facet from an element."""
if not parameters: message = "You must specify an element and a facet."
else:
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].delete(facet)
+ universe.contents[element].remove_facet(facet)
message = "You have successfully deleted the \"" + facet + "\" facet of element \"" + element + "\". Try \"show element " + element + "\" for verification."
- user.send(message)
+ actor.send(message)
-def command_error(user, input_data):
+def command_error(actor, input_data):
"""Generic error for an unrecognized command word."""
# 90% of the time use a generic error
message = "Arglebargle, glop-glyf!?!"
# send the error message
- user.send(message)
+ actor.send(message)
# if there is no universe, create an empty one
if not "universe" in locals(): universe = Universe()
projects / mudpy.git / blobdiff