need a pastebin or some shit ```python ## ====== src/quest_engine/__init__.py import json import os import argparse import pathlib import random import rfc8785 import tempfile import subprocess from abc import ABC, abstractmethod from pydantic import TypeAdapter import paho.mqtt.client as mqtt from .persistence import ( StorageEngine, JsonStorageEngine, iterate_jsonl_with_comments, ) from .visitors import ( TaskIdSetRemoverVisitor, TaskEmitRecordKeepAllVisitor, TaskDefinitionRecordCollector, TaskEmitRecordCompletedPreparerVisitor, ScoreAccumulatorVisitor, ) from .model import ( TaskDefinitionRecord, TaskEmitRecord, LogVisitor, ) MQTT_ADDRESS = os.environ.get("QUEST_ENGINE_MQTT_ADDRESS", "localhost") MQTT_USERNAME = os.environ.get("QUEST_ENGINE_MQTT_USERNAME", "") MQTT_PASSWORD = os.environ.get("QUEST_ENGINE_MQTT_PASSWORD", "") # TODO get home directory dynamically. DATA_PATH = os.environ.get("QUEST_ENGINE_DATA_PATH", "/home/sell/dev/life/todo-system/data.json") ACTION_SCANNER_UNIQUE_NAME = "ssmisa" # "qs-1c440c97" ACTION_SCANNER_TOPIC_STATE = f"{ACTION_SCANNER_UNIQUE_NAME}/profile/state" QUEST_SCANNER_UNIQUE_NAME = "ssmis" # "qs-1c440c97" QUEST_SCANNER_TOPIC_STATE = f"{QUEST_SCANNER_UNIQUE_NAME}/profile/state" def load_config() -> dict: with open(DATA_PATH, "r") as f: return json.load(f) parser = argparse.ArgumentParser('foo') parser.add_argument('--seed', type=int, default=None, help='Set seed for reproducibility') subparsers = parser.add_subparsers(dest='subcommand') load_qd_parser = subparsers.add_parser('load-quest-definitions', help='') load_qd_parser.add_argument('filename', type=pathlib.Path, help='') def _remove_emitted(storage, docs): task_id_candidates = [] for doc in docs: for subtask in doc.subtasks(): task_id_candidates.append(subtask._jcs_task_id_hash()) remaining_candidates = TaskIdSetRemoverVisitor.apply(storage, task_id_candidates) out = [] for doc in docs: this_task_id_set = set(subtask._jcs_task_id_hash() for subtask in doc.subtasks()) if this_task_id_set & remaining_candidates: out.append(doc) return out def _main_load_quest_definitions(options): assert options.subcommand == 'load-quest-definitions' local_rng = random.Random(options.seed) docs = [] with open(options.filename, 'r') as fh: for doc in iterate_jsonl_with_comments(fh): docs.append(TaskDefinitionRecord.model_validate(doc)) for idx, doc in enumerate(docs): pass # print(f"doc[{idx}] = {doc!r}") with tempfile.TemporaryDirectory() as td: tdp = pathlib.Path(td) se = JsonStorageEngine(tdp) assert docs doc = local_rng.sample(docs, 1)[0] for subtask in doc.subtasks(): se.emit_task(subtask) visitor = TaskEmitRecordKeepAllVisitor() se.visit_logs(visitor) subprocess.run(['cat', se._activation_record_log()], check=True) # print(f"seen: {visitor.items}") # reordered = local_rng.sample(docs, len(docs)) # _remove_emitted(storage, reordered) # for doc in docs: # for subtask in doc.subtasks(): # taskhash = subtask._jcs_task_id_hash() # print(f"subtask[#{taskhash}] = {subtask!r}") def main(): options = parser.parse_args() if options.subcommand == 'load-quest-definitions': _main_load_quest_definitions(options) else: raise ValueError("bad subcommand") def mqtt_loop(): config = load_config() client = mqtt.Client() engine = QuestEngine(config, client) def on_connect(client, userdata, flags, rc): print(f"Connected to MQTT broker (rc={rc})") client.subscribe(QUEST_SCANNER_TOPIC_STATE) client.subscribe(ACTION_SCANNER_UNIQUE_NAME) def on_message(client, userdata, msg): print(f"on_message(client={client!r}, userdata={userdata!r}, msg={msg!r})") try: payload = msg.payload.decode("utf-8").strip() print(f"Received: {payload}") engine.handle_input(msg.topic, payload) except Exception as e: print(f"Error handling message: {e}") import traceback traceback.print_exc() client.on_connect = on_connect client.on_message = on_message print(f"Quest Engine starting...") print(f" MQTT broker: {MQTT_ADDRESS}") print(f" Config: {DATA_PATH}") print(f" Database: {db.DB_PATH}") if MQTT_PASSWORD or MQTT_USERNAME: client.username_pw_set(MQTT_USERNAME, MQTT_PASSWORD) client.connect(MQTT_ADDRESS, 1883) client.loop_forever() ## ====== src/quest_engine/model.py import datetime import hashlib from dataclasses import dataclass from typing import Self, ClassVar from abc import ABC, abstractmethod import rfc8785 from pydantic import BaseModel, ConfigDict, ValidationError class MessageMetadata(BaseModel): model_config = ConfigDict(strict=True) record_id: str supersedes: list[str] # `record_id` that this record replaces. @dataclass(frozen=True) class KeyfunValue: identifier: str expires_at: datetime.datetime @classmethod def daily(cls, now: datetime.datetime) -> Self: return cls( now.strftime("%Y-%m-%d"), now + datetime.timedelta(days=2), ) @classmethod def weekly(cls, now: datetime.datetime) -> Self: year, week, _ = now.isocalendar() return cls( f"{year:04}-W{week:02}", now + datetime.timedelta(days=14), ) @classmethod def monthly(cls, now: datetime.datetime) -> Self: return cls( now.strftime("%Y-%m"), now + datetime.timedelta(days=62), ) KeyfunValue.BY_NAME = { "daily": KeyfunValue.daily, "weekly": KeyfunValue.weekly, "monthly": KeyfunValue.monthly, } class TaskDefinitionRecord(BaseModel): """ this is loaded from user config, and within TaskDefinitionFirstUseRecord """ WIRE_NAME: ClassVar[str] = "task-definition-record" model_config = ConfigDict(strict=True) id: str keyfun: str tier: int completion_award: int charge_count: int texts: list[str] def _jcs_hash(self): jcs_serialization = rfc8785.dumps([ 'TaskDefinitionRecord', self.model_dump(), ]) return hashlib.sha256(jcs_serialization).hexdigest() def metadata(self) -> MessageMetadata: return MessageMetadata(record_id=self._jcs_hash(), supersedes=[]) # def instantiate(self): # hash_id = self._jcs_hash() # return TaskDefinitionInstantiation( # hash_id=hash_id, # record=self, # ) def subtasks(self, now=None): if now is None: now = datetime.datetime.now() for subtask_idx, text in enumerate(self.texts): keyfun_value = KeyfunValue.BY_NAME[self.keyfun](now) for charge_idx in range(self.charge_count): yield TaskEmitRecord( task_definition=self, task_definition_hash=self._jcs_hash(), expires_at=keyfun_value.expires_at, keyfun_value=keyfun_value.identifier, subtask_idx=subtask_idx, charge_idx=charge_idx, text=text, ) # class TaskDefinitionInstantiation(BaseModel): # model_config = ConfigDict(strict=True) # hash_id: str # record: TaskDefinitionRecord class TaskEmitRecordReturnToken(BaseModel): """ persists a "return token" value that is used to respond to an issued task, keeps the string open for future consumption. """ WIRE_NAME: ClassVar[str] = "task-emit-record-return-token" model_config = ConfigDict(strict=True) hash_id_short: str task_definition_hash: str task_emit_record_hash: str expires_at: datetime.datetime def _jcs_hash(self): jcs_serialization = rfc8785.dumps([ type(self).WIRE_NAME, self.task_definition_hash, self.task_emit_record_hash, ]) return hashlib.sha256(jcs_serialization).hexdigest() def metadata(self) -> MessageMetadata: return MessageMetadata( record_id=self._jcs_hash(), supersedes=[], ) def mark_completed(self, task: TaskDefinitionRecord) -> 'TaskEmitRecordCompleted': return TaskEmitRecordCompleted( parent=self.metadata().record_id, completion_award=task.completion_award, ) class TaskEmitRecordCompleted(BaseModel): """ persists a "return token" value that is used to respond to an issued task, keeps the string open for future consumption. """ WIRE_NAME: ClassVar[str] = "task-emit-record-completed" model_config = ConfigDict(strict=True) parent: str completion_award: int def _jcs_hash(self): jcs_serialization = rfc8785.dumps([ type(self).WIRE_NAME, self.parent, self.completion_award, ]) return hashlib.sha256(jcs_serialization).hexdigest() def metadata(self) -> MessageMetadata: return MessageMetadata( record_id=self._jcs_hash(), supersedes=[self.parent], ) class TaskEmitRecord(BaseModel): WIRE_NAME: ClassVar[str] = "task-emit-record" model_config = ConfigDict(strict=True) task_definition: TaskDefinitionRecord task_definition_hash: str expires_at: datetime.datetime keyfun_value: str subtask_idx: int charge_idx: int text: str def _task_id(self): return ( self.task_definition.id, self.keyfun_value, self.subtask_idx, self.charge_idx, ) def _jcs_hash(self): jcs_serialization = rfc8785.dumps([ type(self).WIRE_NAME, self._task_id(), ]) return hashlib.sha256(jcs_serialization).hexdigest() def metadata(self) -> MessageMetadata: return MessageMetadata(record_id=self._jcs_hash(), supersedes=[]) def generate_short(self) -> TaskEmitRecordReturnToken: hash_id = self._jcs_hash() return TaskEmitRecordReturnToken( hash_id_short=hash_id[:10], task_definition_hash=self.task_definition_hash, task_emit_record_hash=hash_id, expires_at=self.expires_at, ) class LogVisitor(ABC): def visit_task_definition_record(self, meta: MessageMetadata, v: TaskDefinitionRecord): pass def visit_emitted_task(self, meta: MessageMetadata, v: TaskEmitRecord): pass def visit_task_emit_record_return_token(self, meta: MessageMetadata, v: TaskEmitRecordReturnToken): pass def visit_task_emit_record_completed(self, meta: MessageMetadata, v: TaskEmitRecordCompleted): pass def _not_used_hmac_payload(secret: str, audience: str, task_id: str) -> str: encoded = secret.encode('utf8') + b"\0" + rfc8785.dumps({ 'audience': audience, 'task_id': task_id }) # 48 bits signature should be enough... hashed_hmac = hashlib.sha256(encoded).hexdigest()[:12] return f"{task_id}-{hashed_hmac}" ## ====== src/quest_engine/persistence.py import json import pathlib from abc import ABC, abstractmethod import rfc8785 from .model import ( TaskEmitRecord, LogVisitor, TaskDefinitionRecord, LogVisitor, TaskEmitRecordReturnToken, MessageMetadata, TaskEmitRecordCompleted, ) from .visitors import ( TaskEmitRecordCompletedPreparerVisitor, ) def iterate_jsonl_with_comments(fh): for line in fh: if not line: break line = line.strip() if line.startswith('#'): continue if not line: continue yield json.loads(line) def serde_style_pydantic_object_inflater(class_map, /, ignore_unknown=False): def inflate(doc): doc_items = doc.items() if len(doc_items) != 1: raise ValueError("discriminator dictionary may only have one key") (key, item) = list(doc_items)[0] pydantic_cls = class_map.get(key, None) if pydantic_cls is None and ignore_unknown: return elif pydantic_cls is None: raise ValueError(f"unknown item type {key!r}") # re-encode because pydantic's parser and validator are heavily intertwined.. return pydantic_cls.model_validate_json(json.dumps(item)) return inflate class StorageEngine(ABC): @abstractmethod def emit_task(self, task: TaskEmitRecord): pass @abstractmethod def visit_logs(self, visitor: LogVisitor): pass class JsonStorageEngine(StorageEngine): def __init__(self, base_path: pathlib.Path): self._base_path = base_path def _activation_record_log(self): return self._base_path / 'activation-records.jsonl' def _write_record(self, fh, record): item_key = type(record).WIRE_NAME fh.write(rfc8785.dumps({ 'metadata': record.metadata().model_dump(mode='json'), 'item': { item_key: record.model_dump(mode='json'), }, }) + b"\n") def emit_task(self, task: TaskEmitRecord): task_metadata = task.metadata() first_use_searching_for = task.task_definition.metadata() class Visitor(LogVisitor): def __init__(self): self.items = [] self.already_emitted = False def visit_emitted_task(self, meta: MessageMetadata, v: TaskEmitRecord): if task_metadata.record_id == meta.record_id: self.already_emitted = True def visit_task_definition_record(self, meta: MessageMetadata, v: TaskDefinitionRecord): if first_use_searching_for.record_id == meta.record_id: self.items.append(v) visitor = Visitor() self.visit_logs(visitor) if visitor.already_emitted: raise ValueError("already emitted") with self._activation_record_log().open('ab') as fh: if not visitor.items: self._write_record(fh, task.task_definition) self._write_record(fh, task) self._write_record(fh, task.generate_short()) def complete_task(self, short_code: str): return_visitor = TaskEmitRecordCompletedPreparerVisitor(short_code) self.visit_logs(return_visitor) completion_record = return_visitor.complete() if completion_record: with self._activation_record_log().open('ab') as fh: self._write_record(fh, completion_record) def visit_logs(self, visitor: LogVisitor): if not self._activation_record_log().exists(): return inflater = serde_style_pydantic_object_inflater({ 'task-emit-record': TaskEmitRecord, 'task-definition-record': TaskDefinitionRecord, 'task-emit-record-return-token': TaskEmitRecordReturnToken, 'task-emit-record-completed': TaskEmitRecordCompleted }, ignore_unknown=True) with self._activation_record_log().open('r') as fh: for doc in iterate_jsonl_with_comments(fh): metadata = MessageMetadata.model_validate_json(json.dumps(doc['metadata'])) inflated_or_none = inflater(doc['item']) if inflated_or_none is None: continue if isinstance(inflated_or_none, TaskEmitRecord): visitor.visit_emitted_task(metadata, inflated_or_none) elif isinstance(inflated_or_none, TaskDefinitionRecord): visitor.visit_task_definition_record(metadata, inflated_or_none) elif isinstance(inflated_or_none, TaskEmitRecordReturnToken): visitor.visit_task_emit_record_return_token(metadata, inflated_or_none) elif isinstance(inflated_or_none, TaskEmitRecordCompleted): visitor.visit_task_emit_record_completed(metadata, inflated_or_none) ## ====== src/quest_engine/visitors.py import datetime from abc import ABC, abstractmethod from .model import ( MessageMetadata, TaskEmitRecord, LogVisitor, TaskDefinitionRecord, TaskEmitRecordReturnToken, TaskEmitRecordCompleted, ) class TaskDefinitionRecordCollector(LogVisitor): @classmethod def apply(cls, storage): v = cls() storage.visit_logs(v) return v.score def __init__(self): self.items = [] def visit_task_definition_record(self, meta: MessageMetadata, v: TaskDefinitionRecord): self.items.append(meta.record_id) class TaskIdSetRemoverVisitor(LogVisitor): @classmethod def apply(cls, storage, candidates): v = cls(candidates) storage.visit_logs(v) return v.candidates def __init__(self, candidates): self.candidates = set(candidates) def visit_emitted_task(self, meta: MessageMetadata, v: TaskEmitRecord): self.candidates.discard(meta.record_id) class TaskEmitRecordKeepAllVisitor(LogVisitor): @classmethod def apply(cls, storage): v = cls() storage.visit_logs(v) return v.items def __init__(self): self.items = [] def visit_emitted_task(self, meta: MessageMetadata, v: TaskEmitRecord): self.items.append(v) class TaskEmitRecordCompletedPreparerVisitor(LogVisitor): def __init__( self, short_code: str, /, now: datetime.datetime | None = None ): self._return_token = None self._task_definitions = dict() self._is_completed = False self._short_code = short_code if now is None: now = datetime.datetime.now() self._now = now def visit_task_definition_record(self, meta: MessageMetadata, v: TaskDefinitionRecord): self._task_definitions[meta.record_id] = v def visit_task_emit_record_return_token(self, meta: MessageMetadata, v: TaskEmitRecordReturnToken): if v.expires_at < self._now: return if v.hash_id_short == self._short_code: self._return_token = v def visit_task_emit_record_completed(self, meta: MessageMetadata, v: TaskEmitRecordCompleted): if self._return_token is None: return if self._return_token.metadata().record_id == v.parent: self._is_completed = True def complete(self): if self._return_token is None: raise ValueError("TaskEmitRecordReturnToken not found") task_definition = self._task_definitions.get(self._return_token.task_definition_hash, None) if task_definition is None: raise ValueError("TaskEmitRecordReturnToken orphaned: no associated TaskDefinitionRecord") if self._is_completed: return return self._return_token.mark_completed(task_definition) class TaskEmitRecordFinder(LogVisitor): @classmethod def apply(cls, storage, short_code: str): v = cls(short_code) storage.visit_logs(v) return v.items def __init__(self, short_code: str): self.items = [] self._short_code = short_code def visit_task_emit_record_return_token(self, meta: MessageMetadata, v: TaskEmitRecordReturnToken): if v.hash_id_short == self._short_code: self.items.append(v) class ScoreAccumulatorVisitor(LogVisitor): @classmethod def apply(cls, storage): v = cls() storage.visit_logs(v) return v.score def __init__(self): self.score = 0 def visit_task_emit_record_completed(self, meta: MessageMetadata, v: TaskEmitRecordCompleted): self.score += v.completion_award ## ====== src/quest_engine/test_general.py import unittest import pathlib import random import tempfile import json from quest_engine import ( JsonStorageEngine, TaskDefinitionRecordCollector, TaskDefinitionRecord, TaskEmitRecordKeepAllVisitor, ScoreAccumulatorVisitor, TaskIdSetRemoverVisitor, ) class TestSingleTaskEmit(unittest.TestCase): def _emit_tasks(self, storage): local_rng = random.Random() doc_json = json.dumps({ "id": "example", "keyfun": "daily", "tier": 1, "completion_award": 1000, "charge_count": 1, "texts": ["take 1 breath"] }) docs = [ TaskDefinitionRecord.model_validate_json(doc_json) ] doc = local_rng.sample(docs, 1)[0] subtasks = list(doc.subtasks()) remaining_candidates = TaskIdSetRemoverVisitor.apply(storage, [ s.metadata().record_id for s in subtasks ]) out = [] for subtask in subtasks: if subtask.metadata().record_id in remaining_candidates: storage.emit_task(subtask) out.append(subtask) return out def test_find_emit_record(self): with tempfile.TemporaryDirectory() as td: tdp = pathlib.Path(td) se = JsonStorageEngine(tdp) self._emit_tasks(se) visitor = TaskEmitRecordKeepAllVisitor() se.visit_logs(visitor) items = list(visitor.items) self.assertEqual(len(items), 1) def test_find_task_definition_first_use(self): local_rng = random.Random() with tempfile.TemporaryDirectory() as td: tdp = pathlib.Path(td) se = JsonStorageEngine(tdp) self._emit_tasks(se) visitor = TaskDefinitionRecordCollector() se.visit_logs(visitor) items = list(visitor.items) self.assertEqual(len(items), 1) def test_emit_and_complete(self): local_rng = random.Random() with tempfile.TemporaryDirectory() as td: tdp = pathlib.Path(td) se = JsonStorageEngine(tdp) # get some tasks to do. subtasks = self._emit_tasks(se) first_subtask = subtasks[0] # this represents the barcode that the user holds while doing the # task short_hash = first_subtask.generate_short().hash_id_short # we scan it se.complete_task(short_hash) # and then we validate our score is accounted for self.assertEqual(ScoreAccumulatorVisitor.apply(se), 1000) # ensure double spend does not increment score, as if the user # resubmitted a task completion code se.complete_task(short_hash) # it should remain the same value self.assertEqual(ScoreAccumulatorVisitor.apply(se), 1000) def test_task_emit_only_once(self): with tempfile.TemporaryDirectory() as td: tdp = pathlib.Path(td) se = JsonStorageEngine(tdp) # get some tasks to do, for this test-set, there is only one subtasks = self._emit_tasks(se) self.assertEqual(len(subtasks), 1) # so we should run out if we try again. subtasks = self._emit_tasks(se) self.assertEqual(len(subtasks), 0) if __name__ == '__main__': unittest.main() ```