File size: 2,005 Bytes
031e289 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 | """Interactive four-channel session: /event NAME [INTENSITY], /wait SECONDS, /state."""
import argparse
import json
from codelock_affect import AffectEngine, ProjectConfig
from codelock_affect.model import load
from codelock_affect.session import AffectSession
from codelock_affect.vectors import VectorBank
def main():
p = argparse.ArgumentParser(description=__doc__)
p.add_argument("--model")
p.add_argument("--config", default="configs/default.json")
p.add_argument("--bank", default="vectors/multichannel")
p.add_argument("--state-only", action="store_true")
args = p.parse_args()
config = ProjectConfig.load(args.config)
engine = AffectEngine(config)
if not args.state_only:
model, tok = load(args.model or config.model)
session = AffectSession(model, tok, VectorBank.load(args.bank), engine)
print(__doc__ + " Use /quit to exit.")
while True:
try:
text = input("CODELOCK > ").strip()
if text == "/quit":
break
if text.startswith("/event "):
parts = text.split()
engine.observe(parts[1], float(parts[2]) if len(parts) > 2 else 1)
print(json.dumps(engine.levels.as_dict()))
continue
if text.startswith("/wait "):
engine.advance(float(text.split()[1]))
print(json.dumps(engine.levels.as_dict()))
continue
if text == "/state":
print(json.dumps(engine.snapshot(), indent=2))
continue
if text:
print(
"Use /event, /wait or /state in state-only mode."
if args.state_only
else session.respond(text)
)
except (EOFError, KeyboardInterrupt):
break
except (ValueError, KeyError, IndexError) as error:
print(f"Invalid command: {error}")
if __name__ == "__main__":
main()
|