| """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() |
|
|
