diff --git a/README.rst b/README.rst index b20fe47..524f9b9 100644 --- a/README.rst +++ b/README.rst @@ -149,6 +149,60 @@ To try out the new UI mode: +Learner journeys with known results +----------------------------------- +The backends above pick each event independently, which is good for volume but +means nobody knows what the reports *should* show. The ``journeys`` command +instead simulates each enrolled learner moving through a properly nested course +(sections > subsections > units > problems and videos) in time-ordered +sessions, and writes the expected engagement results alongside the data, so +report numbers can be checked against a known answer: + +:: + + ❯ xapi-db-load journeys --config_file example_configs/journeys_oracle.yaml --now "2026-10-08 12:00:00" + +The same config, seed and ``--now`` always produce identical files. Example +configs: + +- ``journeys_oracle.yaml``: ~9K events, every edge case turned on (``mailto:`` + actors, deleted units, duplicate events, pause and resume at the same instant). + Use it to check correctness. +- ``journeys_m.yaml`` / ``journeys_l.yaml``: ~15M / ~75M events for benchmarks. + +Output, in ``journeys.output_dir`` or ``--output_dir``: + +- ``courses``, ``blocks``, ``external_ids``, ``user_profiles``: event sink rows, + in the same CSV layouts as the ``csv`` backend. Each course is published + ``course_publishes`` times; deleted units only appear in the earlier publishes. +- ``xapi``: ``xapi_events_all`` rows (``event_id``, ``emission_time``, ``event``). +- ``expected_engagement``: one row per learner and section / subsection with + pages, problems or videos, with ``done``, ``total`` and the Aspects status + label. For videos, ``done`` is what the events can show (each play paired with + the next video event) and ``done_truth`` is what the learner actually watched. +- ``expected_video_seconds``: watched seconds per learner and video, total and + distinct, both observable and actual. +- ``manifest.json``: counts, settings, and the most active learner in each + course, for the learner dashboard. + +Files are gzipped CSV (the expected results have a header row). To load them, +copy the directory under the ClickHouse ``user_files`` path and insert with the +``file()`` table function, for example:: + + insert into xapi.xapi_events_all + select * from file('journeys/xapi.csv.gz', 'CSV', + 'event_id UUID, emission_time DateTime64(6), event String'); + +The materialized views only see one insert block at a time, so to imitate live +traffic, set ``sort_events: true`` (this holds every event in memory) and insert +in small blocks, e.g. ``settings max_block_size = 20000, +min_insert_block_size_rows = 20000, min_insert_block_size_bytes = 0, +max_insert_threads = 1, max_threads = 1``. + +``window_days`` must stay under 365, because Aspects drops xAPI events older than +a year. Behavior probabilities (``behavior`` section) are documented in +``xapi_db_load/journeys/simulate.py``. + Secrets and environment variable overrides ------------------------------------------ Sensitive credentials should not be committed to source control. The following diff --git a/docs/xapi_db_load.journeys.rst b/docs/xapi_db_load.journeys.rst new file mode 100644 index 0000000..09e701a --- /dev/null +++ b/docs/xapi_db_load.journeys.rst @@ -0,0 +1,53 @@ +xapi\_db\_load.journeys package +=============================== + +Submodules +---------- + +xapi\_db\_load.journeys.generate module +--------------------------------------- + +.. automodule:: xapi_db_load.journeys.generate + :members: + :show-inheritance: + :undoc-members: + +xapi\_db\_load.journeys.oracle module +------------------------------------- + +.. automodule:: xapi_db_load.journeys.oracle + :members: + :show-inheritance: + :undoc-members: + +xapi\_db\_load.journeys.simulate module +--------------------------------------- + +.. automodule:: xapi_db_load.journeys.simulate + :members: + :show-inheritance: + :undoc-members: + +xapi\_db\_load.journeys.statements module +----------------------------------------- + +.. automodule:: xapi_db_load.journeys.statements + :members: + :show-inheritance: + :undoc-members: + +xapi\_db\_load.journeys.structure module +---------------------------------------- + +.. automodule:: xapi_db_load.journeys.structure + :members: + :show-inheritance: + :undoc-members: + +Module contents +--------------- + +.. automodule:: xapi_db_load.journeys + :members: + :show-inheritance: + :undoc-members: diff --git a/docs/xapi_db_load.rst b/docs/xapi_db_load.rst index b44dbe7..969c918 100644 --- a/docs/xapi_db_load.rst +++ b/docs/xapi_db_load.rst @@ -9,6 +9,7 @@ Subpackages xapi_db_load.backends xapi_db_load.fixtures + xapi_db_load.journeys xapi_db_load.tests xapi_db_load.ui xapi_db_load.xapi diff --git a/docs/xapi_db_load.tests.rst b/docs/xapi_db_load.tests.rst index f2aae57..46982e0 100644 --- a/docs/xapi_db_load.tests.rst +++ b/docs/xapi_db_load.tests.rst @@ -28,6 +28,14 @@ xapi\_db\_load.tests.test\_event\_generator module :show-inheritance: :undoc-members: +xapi\_db\_load.tests.test\_journeys module +------------------------------------------ + +.. automodule:: xapi_db_load.tests.test_journeys + :members: + :show-inheritance: + :undoc-members: + xapi\_db\_load.tests.test\_ui module ------------------------------------ diff --git a/example_configs/journeys_l.yaml b/example_configs/journeys_l.yaml new file mode 100644 index 0000000..48a4e33 --- /dev/null +++ b/example_configs/journeys_l.yaml @@ -0,0 +1,51 @@ +# Large learner-journey dataset (~75M xAPI events, 100k learners) for benchmarks. +# Generate: xapi-db-load journeys --config_file example_configs/journeys_l.yaml +lms_url: http://localhost:18000 + +journeys: + seed: 2027 + output_dir: logs/journeys_l + now: null + window_days: 330 + running_course_fraction: 0.4 + course_length_days: 120 + num_organizations: 5 + num_actors: 100000 + mbox_actor_fraction: 0.05 + num_actor_profile_changes: 2 + course_publishes: 5 + deleted_unit_course_fraction: 0.1 + sort_events: false + courses: + - {template: small, count: 300, runs: [1, 3], learners: [50, 300]} + - {template: medium, count: 200, runs: [1, 3], learners: [200, 800]} + - {template: large, count: 75, runs: [1, 2], learners: [300, 1100]} + course_templates: + small: + chapters: [3, 5] + sequentials_per_chapter: [2, 4] + verticals_per_sequential: [2, 5] + problems_per_vertical: [0.5, 0.3, 0.2] + videos_per_vertical: [0.5, 0.45, 0.05] + graded_sequential_fraction: 0.3 + video_length: [60, 600] + medium: + chapters: [5, 8] + sequentials_per_chapter: [3, 5] + verticals_per_sequential: [3, 6] + problems_per_vertical: [0.4, 0.35, 0.2, 0.05] + videos_per_vertical: [0.45, 0.5, 0.05] + graded_sequential_fraction: 0.3 + video_length: [60, 900] + large: + chapters: [8, 14] + sequentials_per_chapter: [3, 6] + verticals_per_sequential: [3, 8] + problems_per_vertical: [0.35, 0.35, 0.2, 0.1] + videos_per_vertical: [0.4, 0.55, 0.05] + graded_sequential_fraction: 0.3 + video_length: [60, 1200] + behavior: + completer_fraction: 0.1 + continue_prob: 0.9 + duplicate_event_prob: 0.001 diff --git a/example_configs/journeys_m.yaml b/example_configs/journeys_m.yaml new file mode 100644 index 0000000..03b6d49 --- /dev/null +++ b/example_configs/journeys_m.yaml @@ -0,0 +1,51 @@ +# Medium learner-journey dataset (~15M xAPI events, 25k learners) for benchmarks. +# Generate: xapi-db-load journeys --config_file example_configs/journeys_m.yaml +lms_url: http://localhost:18000 + +journeys: + seed: 2026 + output_dir: logs/journeys_m + now: null + window_days: 330 + running_course_fraction: 0.4 + course_length_days: 120 + num_organizations: 5 + num_actors: 25000 + mbox_actor_fraction: 0.05 + num_actor_profile_changes: 2 + course_publishes: 5 + deleted_unit_course_fraction: 0.1 + sort_events: false + courses: + - {template: small, count: 60, runs: [1, 3], learners: [50, 300]} + - {template: medium, count: 40, runs: [1, 3], learners: [200, 800]} + - {template: large, count: 15, runs: [1, 2], learners: [300, 1100]} + course_templates: + small: + chapters: [3, 5] + sequentials_per_chapter: [2, 4] + verticals_per_sequential: [2, 5] + problems_per_vertical: [0.5, 0.3, 0.2] + videos_per_vertical: [0.5, 0.45, 0.05] + graded_sequential_fraction: 0.3 + video_length: [60, 600] + medium: + chapters: [5, 8] + sequentials_per_chapter: [3, 5] + verticals_per_sequential: [3, 6] + problems_per_vertical: [0.4, 0.35, 0.2, 0.05] + videos_per_vertical: [0.45, 0.5, 0.05] + graded_sequential_fraction: 0.3 + video_length: [60, 900] + large: + chapters: [8, 14] + sequentials_per_chapter: [3, 6] + verticals_per_sequential: [3, 8] + problems_per_vertical: [0.35, 0.35, 0.2, 0.1] + videos_per_vertical: [0.4, 0.55, 0.05] + graded_sequential_fraction: 0.3 + video_length: [60, 1200] + behavior: + completer_fraction: 0.1 + continue_prob: 0.9 + duplicate_event_prob: 0.001 diff --git a/example_configs/journeys_oracle.yaml b/example_configs/journeys_oracle.yaml new file mode 100644 index 0000000..b5ee357 --- /dev/null +++ b/example_configs/journeys_oracle.yaml @@ -0,0 +1,49 @@ +# Small, deterministic learner-journey dataset with every edge case turned on. +# Generate: xapi-db-load journeys --config_file example_configs/journeys_oracle.yaml +# The expected_*.csv.gz files hold the known-correct engagement results. +lms_url: http://localhost:18000 + +journeys: + seed: 1335 + output_dir: logs/journeys_oracle + # Pin "now" so the dataset is identical on every run. Keep it within a day of the time the data + # is loaded if you want some events inside refreshable-view lookback windows. + now: null + window_days: 200 + running_course_fraction: 0.6 + course_length_days: 90 + num_organizations: 2 + num_actors: 60 + mbox_actor_fraction: 0.15 + num_actor_profile_changes: 3 + course_publishes: 3 + deleted_unit_course_fraction: 0.5 + # Order xapi.csv by time across all learners, as live traffic arrives, so watch sessions get + # split across insert batches when loaded in small blocks. + sort_events: true + courses: + - {template: small, count: 2, runs: [1, 2], learners: [15, 30]} + - {template: medium, count: 2, runs: [1, 1], learners: [20, 40]} + course_templates: + small: + chapters: [2, 3] + sequentials_per_chapter: [1, 3] + verticals_per_sequential: [1, 4] + problems_per_vertical: [0.5, 0.3, 0.2] + videos_per_vertical: [0.5, 0.4, 0.1] + graded_sequential_fraction: 0.3 + video_length: [30, 240] + medium: + chapters: [3, 5] + sequentials_per_chapter: [2, 4] + verticals_per_sequential: [2, 5] + problems_per_vertical: [0.4, 0.4, 0.2] + videos_per_vertical: [0.5, 0.4, 0.1] + graded_sequential_fraction: 0.3 + video_length: [30, 300] + behavior: + completer_fraction: 0.25 + continue_prob: 0.85 + same_time_resume_prob: 0.2 + duplicate_event_prob: 0.02 + recent_activity_fraction: 0.15 diff --git a/xapi_db_load/journeys/__init__.py b/xapi_db_load/journeys/__init__.py new file mode 100644 index 0000000..542f5e2 --- /dev/null +++ b/xapi_db_load/journeys/__init__.py @@ -0,0 +1,9 @@ +""" +Learner-journey data generation. + +Unlike the random event mode, which picks every event independently, this mode simulates each +enrolled learner moving through a properly nested course: navigating units in order, attempting +the problems and watching the videos in them, in time-ordered sessions. Because the generator knows +exactly what each learner did, it also writes the expected engagement results, so reports can be +checked against a known answer. +""" diff --git a/xapi_db_load/journeys/generate.py b/xapi_db_load/journeys/generate.py new file mode 100644 index 0000000..44ee1b4 --- /dev/null +++ b/xapi_db_load/journeys/generate.py @@ -0,0 +1,276 @@ +""" +Generate a learner-journey dataset: event-sink and xAPI CSV files plus the expected results. + +Output files (in ``output_dir``), using the same CSV layouts as the csv backend so they can be +loaded the same way: + +- courses.csv.gz, blocks.csv.gz, external_ids.csv.gz, user_profiles.csv.gz: event sink tables +- xapi.csv.gz: xapi_events_all rows (event_id, emission_time, event) +- expected_engagement.csv.gz, expected_video_seconds.csv.gz: known answers (with header row) +- manifest.json: settings, counts, and learners worth benchmarking the learner dashboard with +""" + +import csv +import datetime +import gzip +import json +import logging +import os +import random +from contextlib import ExitStack +from typing import ( + Dict, + List, + Optional, +) + +from xapi_db_load.journeys import oracle +from xapi_db_load.journeys.simulate import ( + Behavior, + JourneyActor, + Simulator, +) +from xapi_db_load.journeys.structure import ( + CourseTemplate, + JourneyCourse, + seeded_uuid, +) + +TIME_FORMAT = "%Y-%m-%d %H:%M:%S.%f" +FILES = ( + "courses", + "blocks", + "external_ids", + "user_profiles", + "xapi", + "expected_engagement", + "expected_video_seconds", +) + + +def _fmt(t: datetime.datetime) -> str: + return t.astimezone(datetime.UTC).strftime(TIME_FORMAT) + + +class JourneyGenerator: + """Builds a journey dataset from the ``journeys`` section of a config file.""" + + def __init__(self, conf: Dict, lms_url: str, logger: logging.Logger): + self.conf = conf + self.lms_url = lms_url + self.log = logger + self.rng = random.Random(conf.get("seed", 0)) + now = conf.get("now") + self.now = ( + datetime.datetime.fromisoformat(str(now)).replace(tzinfo=datetime.UTC) + if now + else datetime.datetime.now(datetime.UTC).replace(microsecond=0) + ) + window_days = conf.get("window_days", 300) + if window_days >= 365: + # Aspects' default TTL drops events older than a year. + raise ValueError("window_days must be under 365 to survive the xAPI data TTL.") + if conf.get("course_length_days", 120) > window_days: + raise ValueError("course_length_days must not be more than window_days.") + self.behavior = Behavior.from_config(conf.get("behavior")) + self.templates = { + name: CourseTemplate.from_config(t) + for name, t in conf["course_templates"].items() + } + self.actors: List[JourneyActor] = [] + self.courses: List[JourneyCourse] = [] + self.course_windows: Dict[str, tuple] = {} + + # ---- setup ------------------------------------------------------------------------------ + + def setup(self) -> None: + """Create actors and courses.""" + mbox_fraction = self.conf.get("mbox_actor_fraction", 0.0) + for i in range(self.conf["num_actors"]): + username = f"actor_{i}" + self.actors.append( + JourneyActor( + user_id=i + 1, + external_id=seeded_uuid(self.rng), + username=username, + name=f"Actor {i}", + email=f"{username}@aspects.invalid", + uses_mbox=self.rng.random() < mbox_fraction, + ) + ) + + orgs = [f"Org{i}" for i in range(self.conf.get("num_organizations", 3))] + window = datetime.timedelta(days=self.conf.get("window_days", 300)) + length = datetime.timedelta(days=self.conf.get("course_length_days", 120)) + running_fraction = self.conf.get("running_course_fraction", 0.5) + deleted_fraction = self.conf.get("deleted_unit_course_fraction", 0.0) + + for entry in self.conf["courses"]: + template = self.templates[entry["template"]] + for _ in range(entry["count"]): + org = self.rng.choice(orgs) + code = seeded_uuid(self.rng)[:6] + for run in range(self.rng.randint(*entry.get("runs", [1, 1]))): + course = JourneyCourse( + self.rng, org, code, f"run{run}", entry["template"], template, + self.lms_url, + ) + if self.rng.random() < deleted_fraction: + course.add_deleted_vertical(self.rng) + if self.rng.random() < running_fraction: + end = self.now + else: + end = self.now - self.rng.random() * (window - length) + self.course_windows[course.course_key] = (end - length, end) + course.learner_count = self.rng.randint(*entry["learners"]) + self.courses.append(course) + + self.log.info(f"{len(self.actors)} actors, {len(self.courses)} course runs") + + # ---- writing ---------------------------------------------------------------------------- + + def run(self, output_dir: str) -> Dict: + """Generate everything into ``output_dir`` and return the manifest.""" + self.setup() + os.makedirs(output_dir, exist_ok=True) + with ExitStack() as stack: + handles = { + name: stack.enter_context( + gzip.open(os.path.join(output_dir, f"{name}.csv.gz"), "wt", newline="") + ) + for name in FILES + } + writers = {name: csv.writer(h) for name, h in handles.items()} + writers["expected_engagement"] = csv.DictWriter( + handles["expected_engagement"], oracle.ENGAGEMENT_COLUMNS + ) + writers["expected_video_seconds"] = csv.DictWriter( + handles["expected_video_seconds"], oracle.VIDEO_SECONDS_COLUMNS + ) + writers["expected_engagement"].writeheader() + writers["expected_video_seconds"].writeheader() + + self._write_actors(writers) + self._write_courses(writers) + manifest = self._write_journeys(writers) + + manifest.update( + { + "seed": self.conf.get("seed", 0), + "now": _fmt(self.now), + "lms_url": self.lms_url, + "num_actors": len(self.actors), + "num_course_runs": len(self.courses), + "mbox_actors": sum(a.uses_mbox for a in self.actors), + "courses_with_deleted_unit": sum(bool(c.deleted_blocks) for c in self.courses), + "config": self.conf, + } + ) + with open(os.path.join(output_dir, "manifest.json"), "w") as f: + json.dump(manifest, f, indent=2, default=str) + return manifest + + def _write_actors(self, writers) -> None: + """Write external ids and user profiles, dumped before any course starts.""" + dump_time = self.now - datetime.timedelta(days=400) + for a in self.actors: + writers["external_ids"].writerow( + (a.external_id, "xapi", a.username, a.user_id, seeded_uuid(self.rng), + _fmt(dump_time)) + ) + for change in range(self.conf.get("num_actor_profile_changes", 1)): + t = dump_time + datetime.timedelta(days=change) + for a in self.actors: + writers["user_profiles"].writerow( + (a.user_id, a.user_id, a.name, a.username, a.email, "{}", "", "", "", + 1990, "", "", "", "", "", "", "", "", "", "", + seeded_uuid(self.rng), _fmt(t)) + ) + + def _write_courses(self, writers) -> None: + """ + Write every publish of every course; deleted units are only in the earlier publishes. + """ + publishes = self.conf.get("course_publishes", 1) + for course in self.courses: + start, end = self.course_windows[course.course_key] + for p in range(publishes): + final = p == publishes - 1 + dump_id = seeded_uuid(self.rng) + dump_time = _fmt(start - datetime.timedelta(days=publishes - p)) + writers["courses"].writerow( + (course.org, course.course_key, course.name, _fmt(start), _fmt(end), + _fmt(start), _fmt(end), False, "{}", _fmt(start), dump_time, dump_id, + dump_time) + ) + for b in course.block_rows(include_deleted=not final): + xblock_data = json.dumps( + { + "block_type": b["block_type"], + "section": b["section"], + "subsection": b["subsection"], + "unit": b["unit"], + "graded": b["graded"], + } + ) + writers["blocks"].writerow( + (course.org, course.course_key, b["location"], b["display_name"], + xblock_data, b["order"], dump_time, dump_id, dump_time) + ) + + def _write_journeys(self, writers) -> Dict: # pylint: disable=too-many-locals + """Simulate every enrollment, writing its events and expected results.""" + sim = Simulator(self.rng, self.behavior, self.now) + sort_events = self.conf.get("sort_events", False) + buffered: List[Dict] = [] + n_events = 0 + n_enrollments = 0 + heavy: List[Dict] = [] + + for ci, course in enumerate(self.courses): + start, end = self.course_windows[course.course_key] + lengths = { + b.location: b.length for b in course.all_blocks() if b.block_type == "video" + } + total_pages = sum(len(seq.children) for seq in course.sequentials()) + learners = self.rng.sample(self.actors, min(course.learner_count, len(self.actors))) + best_actor: Optional[JourneyActor] = None + best_pages = 0 + for actor in learners: + enroll_time = start + (end - start) * self.rng.random() * 0.6 + events, truth = sim.run_enrollment(course, actor, enroll_time, running=end == self.now) + n_enrollments += 1 + n_events += len(events) + if sort_events: + buffered.extend(events) + else: + for e in events: + writers["xapi"].writerow( + (e["event_id"], _fmt(e["emission_time"]), e["event"]) + ) + for row in oracle.engagement_rows(course, actor.actor_id, truth): + writers["expected_engagement"].writerow(row) + for row in oracle.video_seconds_rows(course, actor.actor_id, truth, lengths): + writers["expected_video_seconds"].writerow(row) + if total_pages and (best_actor is None or len(truth.pages_viewed) > best_pages): + best_actor, best_pages = actor, len(truth.pages_viewed) + + if best_actor: + heavy.append( + { + "course_key": course.course_key, + "actor_id": best_actor.actor_id, + "username": best_actor.username, + "pages_viewed_fraction": round(best_pages / total_pages, 3), + } + ) + if ci % 50 == 0: + self.log.info(f"course {ci + 1}/{len(self.courses)}: {n_events} events so far") + + if sort_events: + buffered.sort(key=lambda e: (e["emission_time"], e["event_id"])) + for e in buffered: + writers["xapi"].writerow((e["event_id"], _fmt(e["emission_time"]), e["event"])) + + return {"num_enrollments": n_enrollments, "num_xapi_events": n_events, + "heavy_learners": heavy} diff --git a/xapi_db_load/journeys/oracle.py b/xapi_db_load/journeys/oracle.py new file mode 100644 index 0000000..048ec8c --- /dev/null +++ b/xapi_db_load/journeys/oracle.py @@ -0,0 +1,155 @@ +""" +Expected engagement results for simulated learners. + +These are computed from what the simulator recorded each learner doing, independently of any SQL, +and follow the definitions behind the Aspects engagement charts: + +- pages: units (verticals) in the subsection / section with a navigation event for them +- problems: problems with at least one evaluated (server-side check) event +- videos: videos with any watched seconds. "observable" (``done``) is what the events can show: + each play paired with the next video event for the same learner and video, counted when the + position moved forward. "truth" (``done_truth``) is the playback that actually happened, + including stretches cut off by closing the page. + +Section rows add up every subsection in the section, including ones the learner never opened. +Only rows where the section / subsection has at least one item of that type are written. +""" + +from typing import ( + Dict, + Iterator, + List, +) + +from xapi_db_load.journeys.simulate import EnrollmentTruth +from xapi_db_load.journeys.structure import ( + Block, + JourneyCourse, +) + +STATUS_LABELS = { + "pages": ("No pages viewed yet", "At least one page viewed", "All pages viewed"), + "problems": ( + "No problems attempted yet", + "At least one problem attempted", + "All problems attempted", + ), + "videos": ("No videos viewed yet", "At least one video viewed", "All videos viewed"), +} + +ENGAGEMENT_COLUMNS = [ + "org", + "course_key", + "actor_id", + "content_level", + "block_id", + "metric", + "done", + "total", + "status", + "done_truth", + "status_truth", +] + +VIDEO_SECONDS_COLUMNS = [ + "org", + "course_key", + "actor_id", + "video_block_id", + "video_duration", + "truth_seconds", + "truth_distinct_seconds", + "observable_seconds", + "observable_distinct_seconds", +] + + +def status(metric: str, done: int, total: int) -> str: + """Return the Aspects label for done-out-of-total.""" + none_label, some_label, all_label = STATUS_LABELS[metric] + if done == 0: + return none_label + if done == total: + return all_label + return some_label + + +def _items(seq: Block, block_type: str) -> List[Block]: + if block_type == "vertical": + return list(seq.children) + return [c for v in seq.children for c in v.children if c.block_type == block_type] + + +def _counts(seq: Block, truth: EnrollmentTruth) -> Dict[str, tuple]: + """Return {metric: (done, total, done_truth)} for one subsection.""" + pages = _items(seq, "vertical") + problems = _items(seq, "problem") + videos = _items(seq, "video") + + def watched(v: Block, kind: str) -> bool: + vt = truth.videos.get(v.location) + if not vt: + return False + segments = vt.truth if kind == "truth" else vt.observable_segments() + return any(b > a for a, b in segments) + + pages_done = sum(p.location in truth.pages_viewed for p in pages) + problems_done = sum(p.location in truth.problems_attempted for p in problems) + return { + "pages": (pages_done, len(pages), pages_done), + "problems": (problems_done, len(problems), problems_done), + "videos": ( + sum(watched(v, "observable") for v in videos), + len(videos), + sum(watched(v, "truth") for v in videos), + ), + } + + +def engagement_rows( + course: JourneyCourse, actor_id: str, truth: EnrollmentTruth +) -> Iterator[Dict]: + """Yield the expected engagement rows for one enrollment.""" + for chapter in course.chapters: + section_sums = {m: [0, 0, 0] for m in STATUS_LABELS} + for seq in chapter.children: + for metric, (done, total, done_truth) in _counts(seq, truth).items(): + section_sums[metric][0] += done + section_sums[metric][1] += total + section_sums[metric][2] += done_truth + if total: + yield _row(course, actor_id, "subsection", seq, metric, done, total, done_truth) + for metric, (done, total, done_truth) in section_sums.items(): + if total: + yield _row(course, actor_id, "section", chapter, metric, done, total, done_truth) + + +def _row(course, actor_id, level, block, metric, done, total, done_truth) -> Dict: # pylint: disable=too-many-positional-arguments + return { + "org": course.org, + "course_key": course.course_key, + "actor_id": actor_id, + "content_level": level, + "block_id": block.location, + "metric": metric, + "done": done, + "total": total, + "status": status(metric, done, total), + "done_truth": done_truth, + "status_truth": status(metric, done_truth, total), + } + + +def video_seconds_rows( + course: JourneyCourse, actor_id: str, truth: EnrollmentTruth, lengths: Dict[str, int] +) -> Iterator[Dict]: + """Yield expected watched seconds per video for one enrollment.""" + for location, vt in sorted(truth.videos.items()): + yield { + "org": course.org, + "course_key": course.course_key, + "actor_id": actor_id, + "video_block_id": location, + "video_duration": lengths[location], + **vt.summary(), + } diff --git a/xapi_db_load/journeys/simulate.py b/xapi_db_load/journeys/simulate.py new file mode 100644 index 0000000..392e74e --- /dev/null +++ b/xapi_db_load/journeys/simulate.py @@ -0,0 +1,424 @@ +""" +Simulate learners moving through journey courses, emitting xAPI events and recording what happened. +""" + +import datetime +import json +import math +import random +from dataclasses import ( + dataclass, + field, +) +from typing import ( + Dict, + List, + Optional, + Set, + Tuple, +) + +from xapi_db_load.journeys import statements as st +from xapi_db_load.journeys.structure import ( + Block, + JourneyCourse, + seeded_uuid, +) + + +@dataclass +class JourneyActor: + """A learner and the identifiers Aspects sees for them.""" + + user_id: int + external_id: str + username: str + name: str + email: str + # When True, events identify the learner by mailto: email instead of account name. + uses_mbox: bool = False + + @property + def actor_id(self) -> str: + """Return the actor id as Aspects parses it from events.""" + return f"mailto:{self.email}" if self.uses_mbox else self.external_id + + def xapi_actor(self, lms_url: str) -> Dict: + """Return the xAPI actor object for this learner.""" + return st.actor_json( + lms_url, self.external_id, f"mailto:{self.email}" if self.uses_mbox else None + ) + + +@dataclass +class Behavior: + """Probabilities that shape learner behavior; see example_configs/journeys_*.yaml.""" + + start_prob: float = 0.8 + completer_fraction: float = 0.15 + continue_prob: float = 0.9 + unit_skip_prob: float = 0.08 + revisit_prob: float = 0.05 + exit_without_nav_prob: float = 0.7 + problem_attempt_prob: float = 0.8 + max_attempts: int = 3 + video_watch_prob: float = 0.75 + second_session_prob: float = 0.15 + days_between_subsections: float = 2.0 + # Weights for what ends each stretch of video playback. + video_actions: Dict[str, float] = field( + default_factory=lambda: { + "to_end": 0.35, + "pause": 0.25, + "seek": 0.15, + "abandon": 0.1, + "terminate": 0.15, + } + ) + resume_after_pause_prob: float = 0.6 + # Share of learners in still-running courses whose activity is moved so it ends within the + # last ``recent_hours``, giving refreshable views something inside their lookback window. + recent_activity_fraction: float = 0.05 + recent_hours: float = 12.0 + # Edge cases, mainly for the known-answer scenario. + same_time_resume_prob: float = 0.0 + duplicate_event_prob: float = 0.0 + + @classmethod + def from_config(cls, conf: Optional[Dict]) -> "Behavior": + """Build from the YAML ``behavior`` section; missing keys keep their defaults.""" + b = cls() + for k, v in (conf or {}).items(): + if not hasattr(b, k): + raise ValueError(f"Unknown behavior setting: {k}") + setattr(b, k, v) + return b + + +@dataclass +class VideoTruth: + """ + One learner's viewing of one video. + + ``truth`` is the playback that actually happened, as (from, to) whole-second positions. + ``stream`` is the video events emitted for it, from which ``observable_segments`` derives what + the events can show. Second n covers playback from n-1 to n. + """ + + truth: List[Tuple[int, int]] = field(default_factory=list) + # (emission_time, event_id, verb, position) for every emitted non-initialized video event. + stream: List[Tuple[datetime.datetime, str, str, float]] = field(default_factory=list) + + def observable_segments(self) -> List[Tuple[int, int]]: + """ + Pair each play with the next video event and keep the pairs that moved forward. + + Events are ordered by time; on equal times any other event goes before a play, so a pause + and a resume at the same instant stay in the order they happened. The next event's + position is ``time`` (or ``time-from`` for a seek); ``completed`` always reports the full + length, as event-routing-backends sends it. + """ + ordered = sorted(self.stream, key=lambda e: (e[0], e[2] == st.VERB_PLAYED, e[1])) + segments = [] + for (_, _, verb, start), (_, _, _, end) in zip(ordered, ordered[1:]): + if verb == st.VERB_PLAYED and end > start: + segments.append((int(start), int(end))) + return segments + + @staticmethod + def _total(segments) -> int: + """Return the seconds covered by ``segments``, counting overlaps again.""" + return sum(b - a for a, b in segments) + + @staticmethod + def _distinct(segments) -> int: + """Return the seconds covered by ``segments``, counting each second once.""" + seconds: Set[int] = set() + for a, b in segments: + seconds.update(range(a + 1, b + 1)) + return len(seconds) + + def summary(self) -> Dict[str, int]: + """Return total and distinct watched seconds, actual and observable from events.""" + observable = self.observable_segments() + return { + "truth_seconds": self._total(self.truth), + "truth_distinct_seconds": self._distinct(self.truth), + "observable_seconds": self._total(observable), + "observable_distinct_seconds": self._distinct(observable), + } + + +@dataclass +class EnrollmentTruth: + """Everything a learner did in one course that the engagement reports should reflect.""" + + pages_viewed: Set[str] = field(default_factory=set) # vertical locations with a nav event + problems_attempted: Set[str] = field(default_factory=set) # with an evaluated event + videos: Dict[str, VideoTruth] = field(default_factory=dict) # by video location + + +def _retime(raw: str, when: datetime.datetime) -> str: + """Return the statement JSON with its timestamp set to ``when``.""" + statement = json.loads(raw) + statement["timestamp"] = when.isoformat() + return json.dumps(statement) + + +class Simulator: + """Generates the events for one enrollment at a time.""" + + def __init__(self, rng: random.Random, behavior: Behavior, now: datetime.datetime): + self.rng = rng + self.b = behavior + self.now = now + self.events: List[Dict] = [] + self._actor: Dict = {} + self._course: Optional[JourneyCourse] = None + + # ---- event plumbing --------------------------------------------------------------------- + + def _emit(self, when: datetime.datetime, statement_fn, *args, **kwargs) -> Optional[str]: + """Emit an event unless it would be in the future; return its id if emitted.""" + if when > self.now: + return None + event_id = seeded_uuid(self.rng) + row = { + "event_id": event_id, + "emission_time": when, + "event": statement_fn(event_id, self._actor, *args, when=when, **kwargs), + } + self.events.append(row) + if self.rng.random() < self.b.duplicate_event_prob: + self.events.append(dict(row)) + return event_id + + def _is_past(self, when: datetime.datetime) -> bool: + """Return True when ``when`` is not after "now", i.e. activity can happen then.""" + return when <= self.now + + # ---- journeys --------------------------------------------------------------------------- + + def run_enrollment( + self, + course: JourneyCourse, + actor: JourneyActor, + enroll_time: datetime.datetime, + running: bool = False, + ) -> Tuple[List[Dict], EnrollmentTruth]: + """ + Simulate one learner in one course; return its events and what it did. + + In a ``running`` course some learners are still active: their whole journey is moved + later so the last event lands within the last ``recent_hours``. + """ + recent = running and self.rng.random() < self.b.recent_activity_fraction + if not recent: + return self._journey(course, actor, enroll_time) + # Simulate as if "now" were far away so nothing is cut off, then shift into place. + real_now = self.now + self.now = datetime.datetime.max.replace(tzinfo=datetime.UTC) + try: + events, truth = self._journey(course, actor, enroll_time) + finally: + self.now = real_now + last = max(e["emission_time"] for e in events) + target = real_now - datetime.timedelta(hours=self.rng.random() * self.b.recent_hours) + shift = target - last + for e in events: + e["emission_time"] += shift + e["event"] = _retime(e["event"], e["emission_time"]) + return events, truth + + def _journey( + self, + course: JourneyCourse, + actor: JourneyActor, + enroll_time: datetime.datetime, + ) -> Tuple[List[Dict], EnrollmentTruth]: + """Enroll, then work through the subsections in order until the learner stops.""" + self.events = [] + self._actor = actor.xapi_actor(course.lms_url) + self._course = course + truth = EnrollmentTruth() + + mode = self.rng.choice(("audit", "honor", "verified")) + self._emit(enroll_time, st.registered, course.url, mode) + + if self.rng.random() >= self.b.start_prob: + return self.events, truth + + completer = self.rng.random() < self.b.completer_fraction + t = enroll_time + datetime.timedelta(minutes=self.rng.randint(1, 600)) + sequentials = list(course.sequentials()) + + for i, seq in enumerate(sequentials): + if not self._is_past(t): + break + last = i == len(sequentials) - 1 + stopping = last or (not completer and self.rng.random() >= self.b.continue_prob) + t = self._run_subsection(seq, t, truth, stopping) + if stopping: + break + gap_days = self.rng.expovariate(1 / self.b.days_between_subsections) + t += datetime.timedelta(days=gap_days) + + return self.events, truth + + def _run_subsection( + self, seq: Block, t: datetime.datetime, truth: EnrollmentTruth, stopping: bool + ) -> datetime.datetime: + """Work through one subsection's units in order, starting at ``t``.""" + units = seq.children + u = 0 + while u < len(units): + unit = units[u] + t = self._work_unit(unit, t, truth) + final_unit = u == len(units) - 1 + + # Leaving the unit emits a navigation event for it, except when the learner just + # closes the page at the end of their study session. + if final_unit and stopping and self.rng.random() < self.b.exit_without_nav_prob: + break + + t += datetime.timedelta(seconds=self.rng.randint(2, 20)) + if not final_unit and self.rng.random() < self.b.revisit_prob and u > 0: + # Go back one unit and come forward again: two more nav events, one of them + # for the previous unit. + self._nav(unit, units, u, "previous unit", t, truth) + t += datetime.timedelta(seconds=self.rng.randint(10, 120)) + self._nav(units[u - 1], units, u - 1, "next unit", t, truth) + t += datetime.timedelta(seconds=self.rng.randint(2, 20)) + if not final_unit and self.rng.random() < self.b.unit_skip_prob and u + 2 < len(units): + # Jump ahead with the unit tabs, skipping the next unit entirely. + self._nav(unit, units, u, str(u + 3), t, truth) + u += 2 + else: + self._nav(unit, units, u, "next unit", t, truth) + u += 1 + return t + + def _nav( # pylint: disable=too-many-positional-arguments + self, unit: Block, units: List[Block], index: int, ending: str, t, truth: EnrollmentTruth + ) -> None: + """Navigate away from ``unit``, which counts it as viewed.""" + assert self._course + if not self._is_past(t): + return + self._emit(t, st.navigated, self._course.url, unit.url, len(units), index + 1, ending) + truth.pages_viewed.add(unit.location) + + def _work_unit(self, unit: Block, t: datetime.datetime, truth: EnrollmentTruth): + """Spend time on a unit: read, watch its videos, attempt its problems.""" + t += datetime.timedelta(seconds=self.rng.randint(15, 240)) + for child in unit.children: + if not self._is_past(t): + break + if child.block_type == "video" and self.rng.random() < self.b.video_watch_prob: + t = self._watch_video(child, t, truth) + if self.rng.random() < self.b.second_session_prob: + t = self._watch_video(child, t + datetime.timedelta(minutes=5), truth) + elif child.block_type == "problem" and self.rng.random() < self.b.problem_attempt_prob: + t = self._attempt_problem(child, t, truth) + return t + + def _attempt_problem(self, problem: Block, t, truth: EnrollmentTruth): + """Submit the problem until it's correct or the learner runs out of attempts.""" + assert self._course + max_score = self.rng.randint(1, 10) + for attempt in range(1, self.rng.randint(1, self.b.max_attempts) + 1): + t += datetime.timedelta(seconds=self.rng.randint(20, 300)) + if not self._is_past(t): + break + success = self.rng.random() < 0.4 + 0.2 * attempt + raw = max_score if success else self.rng.randint(0, max_score - 1) + self._emit(t, st.problem_attempted, self._course.url, problem.url) + t += datetime.timedelta(seconds=1) + self._emit( + t, st.problem_evaluated, self._course.url, problem.url, + attempt, success, raw, max_score, + ) + if self._is_past(t): + truth.problems_attempted.add(problem.location) + if success: + break + return t + + def _watch_video(self, vid: Block, t, truth: EnrollmentTruth): # pylint: disable=too-many-statements + """ + One viewing session: load, play, then a series of pauses / seeks until it ends. + + Positions are whole seconds. As in the edX player, ``completed`` fires once per session, + when playback first reaches 95%, and reports the full length. + """ + assert self._course + course_url = self._course.url + length = vid.length + threshold = math.ceil(length * 0.95) + vt = truth.videos.setdefault(vid.location, VideoTruth()) + actions = list(self.b.video_actions) + weights = list(self.b.video_actions.values()) + completed_sent = False + + def emit(verb, when, position=None, **kw): + event_id = self._emit( + when, st.video, course_url, vid.url, float(length), verb, **kw + ) + if event_id and verb != st.VERB_INITIALIZED: + vt.stream.append((when, event_id, verb, position)) + + if not self._is_past(t): + return t + emit(st.VERB_INITIALIZED, t) + pos = 0 + t += datetime.timedelta(seconds=self.rng.randint(1, 10)) + emit(st.VERB_PLAYED, t, pos, time=float(pos)) + + while True: + action = self.rng.choices(actions, weights)[0] + if action == "to_end" or length - pos <= 5: + stop = length + else: + stop = self.rng.randint(pos + 1, length - 1) + start_t = t + t = start_t + datetime.timedelta(seconds=stop - pos) + # Playback that would run past "now" hasn't happened yet: only count up to now. + played = min(stop - pos, int((self.now - start_t).total_seconds())) + if played > 0: + vt.truth.append((pos, pos + played)) + if not completed_sent and stop >= threshold: + # Strictly before the event that ends this stretch: two closing events at the + # same instant would have no defined order. + done_t = min( + start_t + datetime.timedelta(seconds=max(threshold - pos, 0) + 0.25), + t - datetime.timedelta(milliseconds=1), + ) + emit(st.VERB_COMPLETED, done_t, length, time=float(length)) + completed_sent = True + if not self._is_past(t): + break # Still playing right now; nothing marks the end yet. + + if stop == length: + emit(st.VERB_TERMINATED, t, length, time=float(length)) + break + if action == "abandon": + break + if action == "terminate": + emit(st.VERB_TERMINATED, t, stop, time=float(stop)) + break + if action == "pause": + emit(st.VERB_PAUSED, t, stop, time=float(stop)) + if self.rng.random() >= self.b.resume_after_pause_prob: + break + if self.rng.random() >= self.b.same_time_resume_prob: + t += datetime.timedelta(seconds=self.rng.randint(2, 120)) + pos = stop + emit(st.VERB_PLAYED, t, pos, time=float(pos)) + continue + # Seek while playing: the player logs the seek, then a play from the new position. + target = self.rng.randint(0, length - 1) + emit(st.VERB_SEEKED, t, stop, time_from=float(stop), time_to=float(target)) + t += datetime.timedelta(seconds=1) + pos = target + emit(st.VERB_PLAYED, t, pos, time=float(pos)) + return t diff --git a/xapi_db_load/journeys/statements.py b/xapi_db_load/journeys/statements.py new file mode 100644 index 0000000..eb7a810 --- /dev/null +++ b/xapi_db_load/journeys/statements.py @@ -0,0 +1,226 @@ +""" +Builders for the xAPI statements emitted by learner journeys. + +The shapes follow what event-routing-backends emits for the matching Open edX events, and the +statements built by the random event classes in ``xapi_db_load.xapi``. +""" + +import datetime +import json +from typing import ( + Any, + Dict, + Optional, +) + +from xapi_db_load.constants import ( + SESSION_ID_PLACEHOLDER, + TRANSFORMER_VERSION, + XAPI_VERSION, +) + +VERB_REGISTERED = "http://adlnet.gov/expapi/verbs/registered" +VERB_NAVIGATED = "https://w3id.org/xapi/dod-isd/verbs/navigated" +VERB_ATTEMPTED = "http://adlnet.gov/expapi/verbs/attempted" +VERB_EVALUATED = "https://w3id.org/xapi/acrossx/verbs/evaluated" +VERB_INITIALIZED = "http://adlnet.gov/expapi/verbs/initialized" +VERB_PLAYED = "https://w3id.org/xapi/video/verbs/played" +VERB_PAUSED = "https://w3id.org/xapi/video/verbs/paused" +VERB_SEEKED = "https://w3id.org/xapi/video/verbs/seeked" +VERB_COMPLETED = "http://adlnet.gov/expapi/verbs/completed" +VERB_TERMINATED = "http://adlnet.gov/expapi/verbs/terminated" + +VIDEO_VERB_DISPLAY = { + VERB_INITIALIZED: "initialized", + VERB_PLAYED: "played", + VERB_PAUSED: "paused", + VERB_SEEKED: "seeked", + VERB_COMPLETED: "completed", + VERB_TERMINATED: "terminated", +} + +EXT_VIDEO_TIME = "https://w3id.org/xapi/video/extensions/time" +EXT_VIDEO_TIME_FROM = "https://w3id.org/xapi/video/extensions/time-from" +EXT_VIDEO_TIME_TO = "https://w3id.org/xapi/video/extensions/time-to" +EXT_VIDEO_LENGTH = "https://w3id.org/xapi/video/extensions/length" + + +def actor_json(lms_url: str, account_name: str, mbox: Optional[str]) -> Dict[str, Any]: + """ + Return the xAPI actor. + + event-routing-backends identifies learners by an anonymous account name by default, or by + ``mailto:`` email when configured with ``XAPI_AGENT_IFI_TYPE = "mbox"``. + """ + if mbox: + return {"objectType": "Agent", "mbox": mbox} + return {"objectType": "Agent", "account": {"homePage": lms_url, "name": account_name}} + + +def _course_parent(course_url: str) -> Dict[str, Any]: + return { + "contextActivities": { + "parent": [ + { + "id": course_url, + "objectType": "Activity", + "definition": { + "name": {"en-US": "Demonstration Course"}, + "type": "http://adlnet.gov/expapi/activities/course", + }, + } + ] + } + } + + +def _base(event_id: str, actor: Dict, verb: str, display: str, when: datetime.datetime) -> Dict: + return { + "id": event_id, + "actor": actor, + "timestamp": when.isoformat(), + "verb": {"display": {"en": display}, "id": verb}, + "version": XAPI_VERSION, + } + + +def registered(event_id: str, actor: Dict, course_url: str, mode: str, when) -> str: + """Return an enrollment statement.""" + e = _base(event_id, actor, VERB_REGISTERED, "registered", when) + e["context"] = { + "extensions": { + "https://w3id.org/xapi/openedx/extension/transformer-version": TRANSFORMER_VERSION, + "https://w3id.org/xapi/openedx/extensions/session-id": SESSION_ID_PLACEHOLDER, + } + } + e["object"] = { + "definition": { + "extensions": {"https://w3id.org/xapi/acrossx/extensions/type": mode}, + "name": {"en": "Demonstration Course"}, + "type": "http://adlnet.gov/expapi/activities/course", + }, + "id": course_url, + "objectType": "Activity", + } + return json.dumps(e) + + +def navigated( # pylint: disable=too-many-positional-arguments + event_id: str, + actor: Dict, + course_url: str, + unit_url: str, + tab_count: int, + current_tab: int, + ending_point: str, + when, +) -> str: + """ + Return a sequence navigation statement (next / previous / tab selected). + + As in the Learning MFE, the object is the unit the learner is navigating away from. + """ + e = _base(event_id, actor, VERB_NAVIGATED, "navigated", when) + e["context"] = _course_parent(course_url) + e["context"]["extensions"] = { + "https://w3id.org/xapi/openedx/extension/transformer-version": TRANSFORMER_VERSION, + "https://w3id.org/xapi/openedx/extensions/session-id": SESSION_ID_PLACEHOLDER, + "http://id.tincanapi.com/extension/starting-position": current_tab, + "http://id.tincanapi.com/extension/ending-point": ending_point, + } + e["object"] = { + "definition": { + "extensions": {"https://w3id.org/xapi/acrossx/extensions/total-items": tab_count}, + "type": "http://id.tincanapi.com/activitytype/resource", + }, + "id": unit_url, + "objectType": "Activity", + } + return json.dumps(e) + + +def problem_attempted(event_id: str, actor: Dict, course_url: str, problem_url: str, when) -> str: + """Return the browser-side problem check statement.""" + e = _base(event_id, actor, VERB_ATTEMPTED, "attempted", when) + e["context"] = _course_parent(course_url) + e["object"] = { + "definition": {"type": "http://adlnet.gov/expapi/activities/cmi.interaction"}, + "id": problem_url, + "objectType": "Activity", + } + return json.dumps(e) + + +def problem_evaluated( # pylint: disable=too-many-positional-arguments + event_id: str, + actor: Dict, + course_url: str, + problem_url: str, + attempt: int, + success: bool, + raw_score: int, + max_score: int, + when, +) -> str: + """Return the server-side problem check statement.""" + e = _base(event_id, actor, VERB_EVALUATED, "evaluated", when) + e["context"] = _course_parent(course_url) + e["context"]["extensions"] = { + "https://github.com/openedx/event-routing-backends/blob/master/docs/xapi-extensions/eventVersion.rst": "1.0" + } + e["object"] = { + "definition": { + "extensions": {"http://id.tincanapi.com/extension/attempt-id": attempt}, + "description": {"en-US": "Add the question text, or prompt, here."}, + "interactionType": "other", + "type": "http://adlnet.gov/expapi/activities/cmi.interaction", + }, + "id": problem_url, + "objectType": "Activity", + } + e["result"] = { + "response": "A correct answer" if success else "An incorrect answer", + "score": { + "scaled": raw_score / max_score, + "raw": raw_score, + "min": 0.0, + "max": max_score, + }, + "success": success, + } + return json.dumps(e) + + +def video( # pylint: disable=too-many-positional-arguments + event_id: str, + actor: Dict, + course_url: str, + video_url: str, + length: float, + verb: str, + when, + time: Optional[float] = None, + time_from: Optional[float] = None, + time_to: Optional[float] = None, +) -> str: + """Return a video statement; which result extensions are set depends on the verb.""" + e = _base(event_id, actor, verb, VIDEO_VERB_DISPLAY[verb], when) + e["context"] = _course_parent(course_url) + e["context"]["extensions"] = { + "https://github.com/openedx/event-routing-backends/blob/master/docs/xapi-extensions/eventVersion.rst": "1.0", + EXT_VIDEO_LENGTH: length, + } + e["object"] = { + "definition": {"type": "https://w3id.org/xapi/video/activity-type/video"}, + "id": video_url, + "objectType": "Activity", + } + extensions: Dict[str, float] = {} + if time is not None: + extensions[EXT_VIDEO_TIME] = time + if time_from is not None: + extensions[EXT_VIDEO_TIME_FROM] = time_from + if time_to is not None: + extensions[EXT_VIDEO_TIME_TO] = time_to + e["result"] = {"extensions": extensions} + return json.dumps(e) diff --git a/xapi_db_load/journeys/structure.py b/xapi_db_load/journeys/structure.py new file mode 100644 index 0000000..b32aa16 --- /dev/null +++ b/xapi_db_load/journeys/structure.py @@ -0,0 +1,239 @@ +""" +Properly nested course structures for learner journeys. + +Each course is chapters (sections) > sequentials (subsections) > verticals (units) > problems and +videos. The structure is built once and reused for every publish, so the generator and the +expected results agree on which subsection every block belongs to. +""" + +import random +import uuid +from dataclasses import ( + dataclass, + field, +) +from typing import ( + Dict, + Iterator, + List, + Optional, + Tuple, +) + + +def seeded_uuid(rng: random.Random) -> str: + """Return a UUID4 string drawn from ``rng`` so runs with the same seed are reproducible.""" + return str(uuid.UUID(int=rng.getrandbits(128), version=4)) + + +@dataclass +class Block: + """One course block, with its position in the course outline.""" + + block_type: str # chapter, sequential, vertical, problem, video + location: str # block-v1:... usage key + url: str # LMS xblock URL, as used for xAPI object ids + display_name: str + section: int + subsection: int + unit: int + order: int = 0 + graded: bool = False + length: int = 0 # Video length in seconds, 0 for other blocks + children: List["Block"] = field(default_factory=list) + + +@dataclass +class CourseTemplate: + """Ranges used to randomize the size of a course.""" + + chapters: Tuple[int, int] + sequentials_per_chapter: Tuple[int, int] + verticals_per_sequential: Tuple[int, int] + # Each vertical gets a random count of these, drawn from the given weights + # (index = count), e.g. [0.4, 0.4, 0.2] -> 40% none, 40% one, 20% two. + problems_per_vertical: List[float] + videos_per_vertical: List[float] + graded_sequential_fraction: float + video_length: Tuple[int, int] + + @classmethod + def from_config(cls, conf: Dict) -> "CourseTemplate": + """Build a template from a YAML ``course_templates`` entry.""" + return cls( + chapters=tuple(conf["chapters"]), + sequentials_per_chapter=tuple(conf["sequentials_per_chapter"]), + verticals_per_sequential=tuple(conf["verticals_per_sequential"]), + problems_per_vertical=list(conf["problems_per_vertical"]), + videos_per_vertical=list(conf["videos_per_vertical"]), + graded_sequential_fraction=conf.get("graded_sequential_fraction", 0.3), + video_length=tuple(conf.get("video_length", [60, 900])), + ) + + +class JourneyCourse: + """A course run with a fixed, nested structure.""" + + def __init__( # pylint: disable=too-many-positional-arguments + self, + rng: random.Random, + org: str, + course_code: str, + run: str, + template_name: str, + template: CourseTemplate, + lms_url: str, + ): + self.org = org + self.course_key = f"course-v1:{org}+{course_code}+{run}" + self.url = f"{lms_url}/course/{self.course_key}" + self.name = f"{course_code} ({template_name})" + self.lms_url = lms_url + self.learner_count = 0 + self._order = 0 + self.chapters: List[Block] = [] + # Blocks that only exist in earlier publishes, see add_deleted_vertical. + self.deleted_blocks: List[Block] = [] + self._build(rng, template) + + def _new_block( # pylint: disable=too-many-positional-arguments + self, + rng: random.Random, + block_type: str, + name: str, + section: int, + subsection: int, + unit: int, + graded: bool = False, + length: int = 0, + ) -> Block: + """Create a block with the next position in course order.""" + self._order += 1 + key = self.course_key.replace("course-v1:", "") + block_id = seeded_uuid(rng).replace("-", "")[:16] + location = f"block-v1:{key}+type@{block_type}+block@{block_id}" + return Block( + block_type=block_type, + location=location, + url=f"{self.lms_url}/xblock/{location}", + display_name=name, + section=section, + subsection=subsection, + unit=unit, + order=self._order, + graded=graded, + length=length, + ) + + def _build(self, rng: random.Random, t: CourseTemplate) -> None: + """Build the outline, drawing each level's size from the template's ranges.""" + for s in range(1, rng.randint(*t.chapters) + 1): + chapter = self._new_block(rng, "chapter", f"Section {s}", s, 0, 0) + self.chapters.append(chapter) + for ss in range(1, rng.randint(*t.sequentials_per_chapter) + 1): + graded = rng.random() < t.graded_sequential_fraction + seq = self._new_block( + rng, "sequential", f"Subsection {s}.{ss}", s, ss, 0, graded + ) + chapter.children.append(seq) + for u in range(1, rng.randint(*t.verticals_per_sequential) + 1): + vert = self._new_block( + rng, "vertical", f"Unit {s}.{ss}.{u}", s, ss, u, graded + ) + seq.children.append(vert) + n_videos = rng.choices( + range(len(t.videos_per_vertical)), t.videos_per_vertical + )[0] + n_problems = rng.choices( + range(len(t.problems_per_vertical)), t.problems_per_vertical + )[0] + for v in range(n_videos): + vert.children.append( + self._new_block( + rng, + "video", + f"Video {s}.{ss}.{u}.{v + 1}", + s, + ss, + u, + graded, + length=rng.randint(*t.video_length), + ) + ) + for p in range(n_problems): + vert.children.append( + self._new_block( + rng, + "problem", + f"Problem {s}.{ss}.{u}.{p + 1}", + s, + ss, + u, + graded, + ) + ) + + def add_deleted_vertical(self, rng: random.Random) -> Optional[Block]: + """ + Record a unit that existed in earlier publishes but was deleted before the final one. + + It's placed in the first subsection, so it would inflate that subsection's page count if a + report kept counting it. + """ + if not self.chapters or not self.chapters[0].children: + return None + seq = self.chapters[0].children[0] + unit = len(seq.children) + 1 + block = self._new_block( + rng, "vertical", f"Unit {seq.section}.{seq.subsection}.{unit} (deleted)", + seq.section, seq.subsection, unit, seq.graded, + ) + self.deleted_blocks.append(block) + return block + + def sequentials(self) -> Iterator[Block]: + """Yield the subsections in course order.""" + for chapter in self.chapters: + yield from chapter.children + + def all_blocks(self) -> Iterator[Block]: + """Yield every block of the final publish, parents before children.""" + for chapter in self.chapters: + yield chapter + for seq in chapter.children: + yield seq + for vert in seq.children: + yield vert + yield from vert.children + + def block_rows(self, include_deleted: bool) -> Iterator[Dict]: + """ + Yield course_blocks rows as event-sink-clickhouse writes them. + + ``include_deleted`` adds the blocks that only exist in earlier publishes. + """ + key = self.course_key.replace("course-v1:", "") + yield { + "location": f"block-v1:{key}+type@course+block@course", + "display_name": self.name, + "block_type": "course", + "section": 0, + "subsection": 0, + "unit": 0, + "graded": False, + "order": 0, + } + blocks = list(self.all_blocks()) + if include_deleted: + blocks += self.deleted_blocks + for b in blocks: + yield { + "location": b.location, + "display_name": b.display_name, + "block_type": b.block_type, + "section": b.section, + "subsection": b.subsection, + "unit": b.unit, + "graded": b.graded, + "order": b.order, + } diff --git a/xapi_db_load/main.py b/xapi_db_load/main.py index 75068cb..65af77b 100644 --- a/xapi_db_load/main.py +++ b/xapi_db_load/main.py @@ -4,6 +4,7 @@ import asyncio import datetime +import logging import os import sys @@ -12,6 +13,7 @@ import yaml from xapi_db_load.constants import DEFAULT_LMS_URL +from xapi_db_load.journeys.generate import JourneyGenerator from xapi_db_load.ui.text_ui import TextUI _ENV_VAR_OVERRIDES = { @@ -111,7 +113,55 @@ def load_db(config_file: str, load_db_only: bool): sys.exit(0) +@click.command() +@click.option( + "--config_file", + help="Configuration file with a 'journeys' section.", + required=True, + type=click.Path(exists=True, dir_okay=False, file_okay=True, writable=False), +) +@click.option( + "--output_dir", + help="Where to write the dataset. Defaults to journeys.output_dir from the config.", + default=None, +) +@click.option("--seed", help="Override journeys.seed.", type=int, default=None) +@click.option( + "--now", + help="Override journeys.now (UTC, e.g. '2026-10-08 12:00:00'); pin it for identical output.", + default=None, +) +def journeys(config_file: str, output_dir: str | None, seed: int | None, now: str | None): + """ + Generate a learner-journey dataset with known expected results, as CSV files. + """ + logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s") + config = get_config(config_file) + conf = config.get("journeys") + if not conf: + raise click.UsageError(f"{config_file} has no 'journeys' section.") + if seed is not None: + conf["seed"] = seed + if now is not None: + conf["now"] = now + out = output_dir or conf.get("output_dir") + if not out: + raise click.UsageError("Set --output_dir or journeys.output_dir in the config.") + try: + generator = JourneyGenerator(conf, config["lms_url"], logging.getLogger("journeys")) + except ValueError as e: + raise click.UsageError(str(e)) from e + start = datetime.datetime.now() + manifest = generator.run(out) + click.echo( + f"Wrote {manifest['num_xapi_events']} xAPI events for {manifest['num_enrollments']} " + f"enrollments in {manifest['num_course_runs']} course runs to {out} " + f"in {datetime.datetime.now() - start}" + ) + + cli.add_command(load_db) +cli.add_command(journeys) cli.add_command(ui) if __name__ == "__main__": diff --git a/xapi_db_load/tests/test_journeys.py b/xapi_db_load/tests/test_journeys.py new file mode 100644 index 0000000..ac1143c --- /dev/null +++ b/xapi_db_load/tests/test_journeys.py @@ -0,0 +1,349 @@ +""" +Tests for learner-journey generation. + +The key check re-derives the expected engagement results from the generated xAPI events alone, so +the simulator's bookkeeping and the events it emits can't drift apart. +""" + +import collections +import csv +import datetime +import gzip +import json +import logging +import pathlib +import random + +import pytest +import yaml +from click.testing import CliRunner + +from xapi_db_load.journeys import statements as st +from xapi_db_load.journeys.generate import JourneyGenerator +from xapi_db_load.journeys.oracle import status +from xapi_db_load.journeys.simulate import ( + Behavior, + EnrollmentTruth, + Simulator, + VideoTruth, +) +from xapi_db_load.journeys.structure import ( + CourseTemplate, + JourneyCourse, +) +from xapi_db_load.main import journeys + +CONFIG = pathlib.Path(__file__).parents[2] / "example_configs" / "journeys_oracle.yaml" +NOW = "2026-10-08 12:00:00" + + +def _generate(out: pathlib.Path, **overrides) -> dict: + conf = yaml.safe_load(CONFIG.read_text())["journeys"] + conf["now"] = NOW + conf.update(overrides) + return JourneyGenerator(conf, "http://localhost:18000", logging.getLogger()).run(str(out)) + + +def _rows(path: pathlib.Path): + with gzip.open(path, "rt", newline="") as f: + yield from csv.reader(f) + + +def _dict_rows(path: pathlib.Path): + with gzip.open(path, "rt", newline="") as f: + yield from csv.DictReader(f) + + +@pytest.fixture(name="dataset", scope="module") +def fixture_dataset(tmp_path_factory): + out = tmp_path_factory.mktemp("journeys") + manifest = _generate(out) + return out, manifest + + +def _actor_id(event: dict) -> str: + actor = event["actor"] + return actor["mbox"] if "mbox" in actor else actor["account"]["name"] + + +def _location(url: str) -> str: + return url.split("/xblock/")[-1] + + +def _events(out): + """ + Unique events (duplicates removed by id), in time order. + + On equal timestamps any other event sorts before a video play, so a pause and the resume at + the same instant pair up in the order they happened. + """ + seen = {} + for event_id, emission_time, raw in _rows(out / "xapi.csv.gz"): + seen[event_id] = (emission_time, json.loads(raw)) + return sorted( + seen.values(), + key=lambda e: (e[0], e[1]["verb"]["id"] == st.VERB_PLAYED, e[1]["id"]), + ) + + +def _structure(out): + """Return {course_key: {block location: (type, section, subsection, unit)}}, final publish.""" + latest = {} + for row in _rows(out / "blocks.csv.gz"): + _org, course_key, location, _name, data, _order, _edited, _dump, dumped = row + if location not in latest or dumped > latest[location][0]: + latest[location] = (dumped, course_key, json.loads(data)) + final_dump = {} + for dumped, course_key, _ in latest.values(): + final_dump[course_key] = max(final_dump.get(course_key, ""), dumped) + courses = collections.defaultdict(dict) + for location, (dumped, course_key, d) in latest.items(): + if dumped == final_dump[course_key]: # blocks missing from the final publish are deleted + courses[course_key][location] = ( + d["block_type"], d["section"], d["subsection"], d["unit"] + ) + return courses + + +def test_reproducible(tmp_path): + _generate(tmp_path / "a") + _generate(tmp_path / "b") + for name in ("xapi", "blocks", "expected_engagement", "expected_video_seconds"): + a = gzip.open(tmp_path / "a" / f"{name}.csv.gz").read() + b = gzip.open(tmp_path / "b" / f"{name}.csv.gz").read() + assert a == b, name + + +def test_ttl_guard(tmp_path): + with pytest.raises(ValueError, match="window_days"): + _generate(tmp_path, window_days=400) + + +def test_events_sorted_and_recent(dataset): + out, manifest = dataset + times = [t for _, t, _ in _rows(out / "xapi.csv.gz")] + assert times == sorted(times) + assert max(times) <= manifest["now"] + # Some activity falls inside a 1-day refresh lookback window. + assert any(t >= "2026-10-07 12:00:00" for t in times) + + +def test_structure_is_nested(dataset): + out, _ = dataset + for blocks in _structure(out).values(): + units = {(s, ss, u) for t, s, ss, u in blocks.values() if t == "vertical"} + subsections = {(s, ss) for t, s, ss, _ in blocks.values() if t == "sequential"} + for t, s, ss, u in blocks.values(): + if t in ("problem", "video"): + assert (s, ss, u) in units + if t == "vertical": + assert (s, ss) in subsections + + +def test_edge_cases_present(dataset): + out, manifest = dataset + assert manifest["mbox_actors"] > 0 + assert manifest["courses_with_deleted_unit"] > 0 + ids = collections.Counter(r[0] for r in _rows(out / "xapi.csv.gz")) + assert any(c > 1 for c in ids.values()), "expected duplicated events" + # Pause and resume at the same instant. + by_time = collections.defaultdict(set) + for t, e in _events(out): + by_time[(t, _actor_id(e), e["object"]["id"])].add(e["verb"]["id"]) + assert any({st.VERB_PAUSED, st.VERB_PLAYED} <= v for v in by_time.values()) + + +def test_expected_engagement_matches_events(dataset): # pylint: disable=too-many-locals + """Rebuild every expected subsection/section row from the raw events and structure.""" + out, _ = dataset + structure = _structure(out) + pages = collections.defaultdict(set) + problems = collections.defaultdict(set) + for _, e in _events(out): + verb = e["verb"]["id"] + key = (e["context"]["contextActivities"]["parent"][0]["id"].split("/course/")[-1] + if "contextActivities" in e.get("context", {}) else None, _actor_id(e)) + if verb == st.VERB_NAVIGATED: + pages[key].add(_location(e["object"]["id"])) + elif verb == st.VERB_EVALUATED: + problems[key].add(_location(e["object"]["id"])) + + checked = 0 + for row in _dict_rows(out / "expected_engagement.csv.gz"): + if row["metric"] == "videos": + continue + blocks = structure[row["course_key"]] + _, s, ss, _ = blocks[row["block_id"]] + level = row["content_level"] + item_type = "vertical" if row["metric"] == "pages" else "problem" + items = { + loc for loc, (t, bs, bss, _) in blocks.items() + if t == item_type and bs == s and (level == "section" or bss == ss) + } + seen = (pages if row["metric"] == "pages" else problems)[ + (row["course_key"], row["actor_id"]) + ] + done = len(items & seen) + assert (done, len(items)) == (int(row["done"]), int(row["total"])), row + assert row["status"] == status(row["metric"], done, len(items)) + checked += 1 + assert checked > 1000 + + +def test_observable_video_seconds_match_events(dataset): + """ + Pair each play with the next event for the same learner and video; the forward progress of + those pairs is what the events can show, and must equal the expected observable seconds. + """ + out, _ = dataset + streams = collections.defaultdict(list) + for _, e in _events(out): + if e["verb"]["id"] in st.VIDEO_VERB_DISPLAY and e["verb"]["id"] != st.VERB_INITIALIZED: + course = e["context"]["contextActivities"]["parent"][0]["id"].split("/course/")[-1] + streams[(course, _actor_id(e), _location(e["object"]["id"]))].append(e) + + def position(e): + ext = e["result"]["extensions"] + return ext.get(st.EXT_VIDEO_TIME, ext.get(st.EXT_VIDEO_TIME_FROM)) + + derived = collections.Counter() + for key, events in streams.items(): + for cur, nxt in zip(events, events[1:]): + if cur["verb"]["id"] == st.VERB_PLAYED and position(nxt) > position(cur): + derived[key] += int(position(nxt) - position(cur)) + + expected = { + (r["course_key"], r["actor_id"], r["video_block_id"]): int(r["observable_seconds"]) + for r in _dict_rows(out / "expected_video_seconds.csv.gz") + } + mismatches = {k: (derived.get(k, 0), v) for k, v in expected.items() if derived.get(k, 0) != v} + assert not mismatches, list(mismatches.items())[:5] + + +def test_unsorted_output_has_same_events(dataset, tmp_path): + """sort_events only changes the order of xapi.csv, not what is generated.""" + out, _ = dataset + _generate(tmp_path, sort_events=False) + for name in ("xapi", "expected_engagement", "expected_video_seconds"): + assert sorted(_rows(out / f"{name}.csv.gz")) == sorted(_rows(tmp_path / f"{name}.csv.gz")) + + +def test_status_labels(): + assert status("pages", 0, 3) == "No pages viewed yet" + assert status("problems", 1, 3) == "At least one problem attempted" + assert status("videos", 3, 3) == "All videos viewed" + + +def test_unknown_behavior_setting(): + with pytest.raises(ValueError, match="Unknown behavior setting: typo"): + Behavior.from_config({"typo": 1}) + + +def _stream(*events): + """Build a video event stream from (second, verb, position) tuples.""" + t0 = datetime.datetime(2026, 1, 1, tzinfo=datetime.UTC) + return [ + (t0 + datetime.timedelta(seconds=sec), f"id{i}", verb, pos) + for i, (sec, verb, pos) in enumerate(events) + ] + + +def test_observable_segments_pause_and_resume_at_same_time(): + """A pause and a resume with the same timestamp still pair in the order they happened.""" + vt = VideoTruth( + truth=[(0, 40), (40, 60)], + # The resume is listed first to show that ordering doesn't depend on input order. + stream=_stream( + (40, st.VERB_PLAYED, 40), + (0, st.VERB_PLAYED, 0), + (40, st.VERB_PAUSED, 40), + (60, st.VERB_TERMINATED, 60), + ), + ) + assert vt.observable_segments() == [(0, 40), (40, 60)] + assert vt.summary() == { + "truth_seconds": 60, + "truth_distinct_seconds": 60, + "observable_seconds": 60, + "observable_distinct_seconds": 60, + } + + +def test_observable_segments_seek_back_and_close(): + """Seeking back re-watches seconds; closing the tab leaves the last play unpaired.""" + vt = VideoTruth( + truth=[(0, 30), (10, 25)], + stream=_stream( + (0, st.VERB_PLAYED, 0), + (30, st.VERB_SEEKED, 30), # from 30 back to 10 + (31, st.VERB_PLAYED, 10), + ), + ) + assert vt.observable_segments() == [(0, 30)] + assert vt.summary() == { + "truth_seconds": 45, + "truth_distinct_seconds": 30, + "observable_seconds": 30, + "observable_distinct_seconds": 30, + } + + +def test_video_truth_stops_at_now(): + """A video still playing at "now" only counts the seconds played so far, and has no end event.""" + rng = random.Random(1) + template = CourseTemplate.from_config({ + "chapters": [1, 1], "sequentials_per_chapter": [1, 1], "verticals_per_sequential": [1, 1], + "problems_per_vertical": [1], "videos_per_vertical": [0, 1], "video_length": [600, 600], + }) + course = JourneyCourse(rng, "Org", "C", "run", "t", template, "http://lms") + video = next(b for b in course.all_blocks() if b.block_type == "video") + start = datetime.datetime(2026, 1, 1, tzinfo=datetime.UTC) + sim = Simulator(rng, Behavior(video_actions={"to_end": 1}), start + datetime.timedelta(seconds=100)) + sim._course = course # pylint: disable=protected-access + truth = EnrollmentTruth() + sim._watch_video(video, start, truth) # pylint: disable=protected-access + + vt = truth.videos[video.location] + (begin, end), = vt.truth + assert begin == 0 and 80 <= end < 100 # play starts 1-10 s after loading + assert [verb for _, _, verb, _ in vt.stream] == [st.VERB_PLAYED] + assert vt.observable_segments() == [] + + +def test_cli(tmp_path): + out = tmp_path / "out" + result = CliRunner().invoke( + journeys, + ["--config_file", str(CONFIG), "--output_dir", str(out), "--now", NOW, "--seed", "7"], + catch_exceptions=False, + ) + assert result.exit_code == 0, result.output + assert "xAPI events for" in result.output + manifest = json.loads((out / "manifest.json").read_text()) + assert manifest["seed"] == 7 + assert manifest["now"].startswith(NOW) + for name in ("courses", "blocks", "external_ids", "user_profiles", "xapi", + "expected_engagement", "expected_video_seconds"): + assert (out / f"{name}.csv.gz").exists() + + +@pytest.mark.parametrize( + "journeys_conf, message", + [ + (None, "has no 'journeys' section"), + ({"window_days": 400}, "window_days must be under 365"), + ({"window_days": 30, "course_length_days": 60}, "course_length_days"), + ({"output_dir": None}, "Set --output_dir"), + ], +) +def test_cli_config_errors(tmp_path, journeys_conf, message): + conf = yaml.safe_load(CONFIG.read_text()) + if journeys_conf is None: + del conf["journeys"] + else: + conf["journeys"].update(journeys_conf) + config_file = tmp_path / "config.yaml" + config_file.write_text(yaml.safe_dump(conf)) + result = CliRunner().invoke(journeys, ["--config_file", str(config_file)]) + assert result.exit_code == 2 + assert message in result.output