# This file is part of Elide, frontend to Lisien, a framework for life simulation games.
# Copyright (c) Zachary Spector, public@zacharyspector.com
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, version 3.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""Object to configure, start, and stop elide."""
import json
import os
import sys
from threading import Thread
from lisien.exc import OutOfTimelineError
if "KIVY_NO_ARGS" not in os.environ:
os.environ["KIVY_NO_ARGS"] = "1"
from kivy.app import App
from kivy.clock import Clock, triggered
from kivy.logger import Logger
from kivy.properties import (
AliasProperty,
BooleanProperty,
NumericProperty,
ObjectProperty,
StringProperty,
)
from kivy.resources import resource_add_path
from kivy.uix.screenmanager import NoTransition, ScreenManager
import elide
import elide.charsview
import elide.dialog
import elide.rulesview
import elide.screen
import elide.spritebuilder
import elide.statcfg
import elide.stores
import elide.timestream
from elide.graph.arrow import GraphArrow
from elide.graph.board import GraphBoard
from elide.grid.board import GridBoard
from lisien.proxy import (
CharStatProxy,
EngineProcessManager,
PlaceProxy,
ThingProxy,
)
resource_add_path(elide.__path__[0] + "/assets")
resource_add_path(elide.__path__[0] + "/assets/rltiles")
resource_add_path(elide.__path__[0] + "/assets/kenney1bit")
def trigger(func):
return triggered()(func)
[docs]
class ELiDEApp(App):
"""Extensible lisien Development Environment."""
title = "elide"
branch = StringProperty("trunk")
turn = NumericProperty(0)
tick = NumericProperty(0)
character = ObjectProperty()
selection = ObjectProperty(None, allownone=True)
selected_proxy = ObjectProperty()
selected_proxy_name = StringProperty("")
statcfg = ObjectProperty()
edit_locked = BooleanProperty(False)
simulate_button_down = BooleanProperty(False)
def on_selection(self, *_):
Logger.debug("App: {} selected".format(self.selection))
def on_selected_proxy(self, *_):
if hasattr(self.selected_proxy, "name"):
self.selected_proxy_name = str(self.selected_proxy.name)
return
selected_proxy = self.selected_proxy
assert hasattr(selected_proxy, "origin"), "{} has no origin".format(
type(selected_proxy)
)
assert hasattr(selected_proxy, "destination"), (
"{} has no destination".format(type(selected_proxy))
)
origin = selected_proxy.origin
destination = selected_proxy.destination
self.selected_proxy_name = (
str(origin.name) + "->" + str(destination.name)
)
def _get_character_name(self, *_):
if self.character is None:
return
return self.character.name
def _set_character_name(self, name):
if self.character.name != name:
self.character = self.engine.character[name]
character_name = AliasProperty(
_get_character_name, _set_character_name, bind=("character",)
)
def _pull_time(self, *_):
if not hasattr(self, "engine"):
Clock.schedule_once(self._pull_time, 0)
return
branch, turn, tick = self.engine._btt()
self.branch = branch
self.turn = turn
self.tick = tick
pull_time = trigger(_pull_time)
def _really_time_travel(self, branch, turn, tick):
try:
self.engine._set_btt(
branch, turn, tick, cb=self._update_from_time_travel
)
except OutOfTimelineError as ex:
Logger.warning(
f"App: couldn't time travel to {(branch, turn, tick)}: "
+ ex.args[0],
exc_info=ex,
)
(self.branch, self.turn, self.tick) = (
ex.branch_from,
ex.turn_from,
ex.tick_from,
)
finally:
self.edit_locked = False
del self._time_travel_thread
def time_travel(self, branch, turn, tick=None):
if hasattr(self, "_time_travel_thread"):
return
self.edit_locked = True
self._time_travel_thread = Thread(
target=self._really_time_travel, args=(branch, turn, tick)
)
self._time_travel_thread.start()
def _really_time_travel_to_tick(self, tick):
try:
self.engine._set_btt(
self.branch, self.turn, tick, cb=self._update_from_time_travel
)
except OutOfTimelineError as ex:
Logger.warning(
f"App: couldn't time travel to {(self.branch, self.turn, tick)}: "
+ ex.args[0],
exc_info=ex,
)
(self.branch, self.turn, self.tick) = (
ex.branch_from,
ex.turn_from,
ex.tick_from,
)
finally:
self.edit_locked = False
del self._time_travel_thread
def time_travel_to_tick(self, tick):
self._time_travel_thread = Thread(
target=self._really_time_travel_to_tick, args=(tick,)
)
self._time_travel_thread.start()
def _update_from_time_travel(
self, command, branch, turn, tick, result, **kwargs
):
(self.branch, self.turn, self.tick) = (branch, turn, tick)
self.mainscreen.update_from_time_travel(
command, branch, turn, tick, result, **kwargs
)
[docs]
def set_tick(self, t):
"""Set my tick to the given value, cast to an integer."""
self.tick = int(t)
[docs]
def set_turn(self, t):
"""Set the turn to the given value, cast to an integer"""
self.turn = int(t)
[docs]
def select_character(self, char):
"""Change my ``character`` to the selected character object if they
aren't the same.
"""
if char == self.character:
return
self.character = char
[docs]
def build_config(self, config):
"""Set config defaults"""
for sec in "lisien", "elide":
config.adddefaultsection(sec)
config.setdefaults(
"lisien",
{
"language": "eng",
"logfile": "lisien.log",
"loglevel": "debug",
"replayfile": "",
},
)
config.setdefaults(
"elide",
{
"debugger": "no",
"inspector": "no",
"user_kv": "yes",
"play_speed": "1",
"thing_graphics": json.dumps(
[
("Kenney: 1 bit", "kenney1bit.atlas"),
("RLTiles: Body", "base.atlas"),
("RLTiles: Basic clothes", "body.atlas"),
("RLTiles: Armwear", "arm.atlas"),
("RLTiles: Legwear", "leg.atlas"),
("RLTiles: Right hand", "hand1.atlas"),
("RLTiles: Left hand", "hand2.atlas"),
("RLTiles: Boots", "boot.atlas"),
("RLTiles: Hair", "hair.atlas"),
("RLTiles: Beard", "beard.atlas"),
("RLTiles: Headwear", "head.atlas"),
]
),
"place_graphics": json.dumps(
[
("Kenney: 1 bit", "kenney1bit.atlas"),
("RLTiles: Dungeon", "dungeon.atlas"),
("RLTiles: Floor", "floor.atlas"),
]
),
},
)
config.write()
[docs]
def build(self):
self.icon = "icon_24px.png"
config = self.config
if config["elide"]["debugger"] == "yes":
import pdb
pdb.set_trace()
self.manager = ScreenManager(transition=NoTransition())
if config["elide"]["inspector"] == "yes":
from kivy.core.window import Window
from kivy.modules import inspector
inspector.create_inspector(Window, self.manager)
self._add_screens()
return self.manager
def _pull_lang(self, *_, **kwargs):
self.strings.language = kwargs["language"]
def _pull_chars(self, *_, **__):
self.chars.names = list(self.engine.character)
def _pull_time_from_signal(self, *_, then, now):
self.branch, self.turn, self.tick = now
self.mainscreen.ids.turnscroll.value = self.turn
[docs]
def start_subprocess(self, path=None, *_):
"""Start the lisien core and get a proxy to it
Must be called before ``init_board``
"""
if hasattr(self, "_started"):
raise ChildProcessError("Subprocess already running")
config = self.config
enkw = {
"logger": Logger,
"threaded_triggers": False,
"do_game_start": getattr(self, "do_game_start", False),
}
workers = config["lisien"].get("workers", "")
if workers:
enkw["workers"] = workers
if config["lisien"].get("logfile"):
enkw["logfile"] = config["lisien"]["logfile"]
if config["lisien"].get("loglevel"):
enkw["loglevel"] = config["lisien"]["loglevel"]
if config["lisien"].get("replayfile"):
self._replayfile = open(config["lisien"].get("replayfile"), "at")
enkw["replay_file"] = self._replayfile
if path is not None and os.path.isdir(path):
startdir = path
elif os.path.isdir(sys.argv[-1]):
startdir = sys.argv[-1]
else:
startdir = None
self.procman = EngineProcessManager()
self.engine = engine = self.procman.start(startdir, **enkw)
self.pull_time()
self.engine.time.connect(self._pull_time_from_signal, weak=False)
self.engine.character.connect(self._pull_chars, weak=False)
self.strings.store = self.engine.string
self._started = True
return engine
trigger_start_subprocess = trigger(start_subprocess)
[docs]
def init_board(self, *_):
"""Get the board widgets initialized to display the game state
Must be called after start_subprocess
"""
if "boardchar" not in self.engine.eternal:
if "physical" in self.engine.character:
self.engine.eternal["boardchar"] = self.engine.character[
"physical"
]
else:
chara = self.engine.eternal["boardchar"] = (
self.engine.new_character("physical")
)
self.chars.names = list(self.engine.character)
self.mainscreen.graphboards = {
name: GraphBoard(character=char)
for name, char in self.engine.character.items()
}
self.mainscreen.gridboards = {
name: GridBoard(character=char)
for name, char in self.engine.character.items()
}
self.select_character(self.engine.eternal["boardchar"])
self.selected_proxy = self._get_selected_proxy()
def _add_screens(self, *_):
def toggler(screenname):
def tog(*_):
if self.manager.current == screenname:
self.manager.current = "main"
else:
self.manager.current = screenname
return tog
config = self.config
self.mainmenu = elide.menu.DirPicker(toggle=toggler("mainmenu"))
self.pawncfg = elide.spritebuilder.PawnConfigScreen(
toggle=toggler("pawncfg"),
data=json.loads(config["elide"]["thing_graphics"]),
)
self.spotcfg = elide.spritebuilder.SpotConfigScreen(
toggle=toggler("spotcfg"),
data=json.loads(config["elide"]["place_graphics"]),
)
self.statcfg = elide.statcfg.StatScreen(toggle=toggler("statcfg"))
self.rules = elide.rulesview.RulesScreen(toggle=toggler("rules"))
self.charrules = elide.rulesview.CharacterRulesScreen(
character=self.character, toggle=toggler("charrules")
)
self.bind(character=self.charrules.setter("character"))
self.chars = elide.charsview.CharactersScreen(
toggle=toggler("chars"), new_board=self.new_board
)
self.bind(character_name=self.chars.setter("character_name"))
def chars_push_character_name(*_):
self.unbind(character_name=self.chars.setter("character_name"))
self.character_name = self.chars.character_name
self.bind(character_name=self.chars.setter("character_name"))
self.chars.push_character_name = chars_push_character_name
self.strings = elide.stores.StringsEdScreen(toggle=toggler("strings"))
self.funcs = elide.stores.FuncsEdScreen(
name="funcs", toggle=toggler("funcs")
)
self.bind(selected_proxy=self.statcfg.setter("proxy"))
self.timestream = elide.timestream.TimestreamScreen(
name="timestream", toggle=toggler("timestream")
)
self.mainscreen = elide.screen.MainScreen(
use_kv=config["elide"]["user_kv"] == "yes",
play_speed=int(config["elide"]["play_speed"]),
)
if self.mainscreen.statlist:
self.statcfg.statlist = self.mainscreen.statlist
self.mainscreen.bind(statlist=self.statcfg.setter("statlist"))
self.bind(
selection=self.refresh_selected_proxy,
character=self.refresh_selected_proxy,
)
for wid in (
self.mainmenu,
self.mainscreen,
self.pawncfg,
self.spotcfg,
self.statcfg,
self.rules,
self.charrules,
self.chars,
self.strings,
self.funcs,
self.timestream,
):
self.manager.add_widget(wid)
if (
(os.environ["KIVY_NO_ARGS"] or sys.argv[-2] == "-")
and os.path.exists(sys.argv[-1])
and os.path.isdir(sys.argv[-1])
):
self.mainmenu.open(os.path.abspath(sys.argv[-1]))
[docs]
def update_calendar(self, calendar, past_turns=1, future_turns=5):
"""Fill in a calendar widget with actual simulation data"""
startturn = self.turn - past_turns
endturn = self.turn + future_turns
stats = [
stat
for stat in self.selected_proxy
if isinstance(stat, str)
and not stat.startswith("_")
and stat not in ("character", "name", "units", "wallpaper")
]
if "_config" in self.selected_proxy:
stats.append("_config")
if isinstance(self.selected_proxy, CharStatProxy):
sched_entity = self.engine.character[self.selected_proxy.name]
else:
sched_entity = self.selected_proxy
calendar.entity = sched_entity
if startturn == endturn == self.turn:
# It's the "calendar" that's actually just the current stats
# of the selected entity, on the left side of elide
schedule = {stat: [self.selected_proxy[stat]] for stat in stats}
else:
schedule = (
self.engine.handle(
"get_schedule",
entity=sched_entity,
stats=stats,
beginning=startturn,
end=endturn,
),
)
calendar.from_schedule(schedule, start_turn=startturn)
def _set_language(self, lang):
self.engine.string.language = lang
def _get_selected_proxy(self):
if self.selection is None:
return self.character.stat
elif hasattr(self.selection, "proxy"):
return self.selection.proxy
elif hasattr(self.selection, "origin") and hasattr(
self.selection, "destination"
):
return self.character.portal[self.selection.origin.name][
self.selection.destination.name
]
else:
raise ValueError("Invalid selection: {}".format(self.selection))
def refresh_selected_proxy(self, *_):
self.selected_proxy = self._get_selected_proxy()
def on_character_name(self, *_):
if not hasattr(self, "engine"):
Clock.schedule_once(self.on_character_name, 0)
return
self.engine.eternal["boardchar"] = self.engine.character[
self.character_name
]
def on_character(self, *_):
if not hasattr(self, "mainscreen"):
Clock.schedule_once(self.on_character, 0)
return
if hasattr(self, "_oldchar"):
self.mainscreen.graphboards[self._oldchar.name].unbind(
selection=self.setter("selection")
)
self.mainscreen.gridboards[self._oldchar.name].unbind(
selection=self.setter("selection")
)
self.selection = None
self.mainscreen.graphboards[self.character.name].bind(
selection=self.setter("selection")
)
self.mainscreen.gridboards[self.character.name].bind(
selection=self.setter("selection")
)
[docs]
def on_pause(self):
"""Sync the database with the current state of the game."""
if hasattr(self, "engine"):
self.engine.commit()
self.strings.save()
self.funcs.save()
[docs]
def on_stop(self, *largs):
"""Sync the database, wrap up the game, and halt."""
if hasattr(self, "stopped"):
return
self.stopped = True
self.strings.save()
self.funcs.save()
if hasattr(self, "procman"):
self.procman.shutdown()
if hasattr(self, "engine"):
del self.engine
if hasattr(self, "_replayfile"):
self._replayfile.close()
[docs]
def delete_selection(self):
"""Delete both the selected widget and whatever it represents."""
selection = self.selection
if selection is None:
return
if isinstance(selection, GraphArrow):
self.mainscreen.boardview.board.rm_arrow(
selection.origin.name, selection.destination.name
)
selection.character.portal[selection.origin.name][
selection.destination.name
].delete()
elif isinstance(selection.proxy, PlaceProxy):
charn = selection.board.character.name
self.mainscreen.graphboards[charn].rm_spot(selection.name)
gridb = self.mainscreen.gridboards[charn]
if selection.name in gridb.spot:
gridb.rm_spot(selection.name)
selection.proxy.delete()
else:
assert isinstance(selection.proxy, ThingProxy)
charn = selection.board.character.name
self.mainscreen.graphboards[charn].rm_pawn(selection.name)
self.mainscreen.gridboards[charn].rm_pawn(selection.name)
selection.proxy.delete()
self.selection = None
[docs]
def new_board(self, name):
"""Make a graph for a character name, and switch to it."""
char = self.engine.character[name]
self.mainscreen.graphboards[name] = GraphBoard(character=char)
self.mainscreen.gridboards[name] = GridBoard(character=char)
self.character = char
def on_edit_locked(self, *_):
Logger.debug(
"ELiDEApp: "
+ ("edit locked" if self.edit_locked else "edit unlocked")
)