diff --git a/README.rst b/README.rst index b20fe47..310a573 100644 --- a/README.rst +++ b/README.rst @@ -149,6 +149,84 @@ 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 (the "oracle", computed in +``xapi_db_load/journeys/oracle.py``): + +:: + + ❯ 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 a +dataset into an Aspects install, use ``load-journeys`` with a config that has +the same ClickHouse settings as ``load-db``, plus the ``journeys`` section or +``--dataset_dir``:: + + ❯ xapi-db-load load-journeys --config_file my_config.yaml --dataset_dir logs/journeys_oracle + +It inserts the event sink rows straight into ClickHouse (``event_sink_db_name``, +default ``event_sink``) and sends the xAPI events through ``backend``: +``clickhouse`` inserts them into ``db_name``.``xapi_events_all``, ``ralph`` POSTs +them to ``lrs_url`` (without the duplicate events, which an LRS rejects), and +``vector`` logs them to stdout for Vector to pick up, pausing +``vector_batch_pause_seconds`` (default 0.1) after every ``vector_batch_size`` +(default 100) events. The expected results go +into the ``expected_db_name`` database (default ``journeys``). Ralph and Vector +insert asynchronously, so the command waits, up to ``--wait_seconds``, until +every event has arrived. To load journeys next to data from ``load-db``, set +``org_prefix`` and ``first_user_id`` in the ``journeys`` section so the +organizations and users don't collide with ``load-db``'s ``Org0``... and user +ids from 0. + +To load the files by hand instead, 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. Every ``behavior`` setting is described on the ``Behavior`` class in +``xapi_db_load/journeys/simulate.py``. Course template sizes are ``[low, high]`` +ranges, except ``problems_per_vertical_weights`` and +``videos_per_vertical_weights``, which weight how many problems and videos each +unit gets: ``[0.4, 0.4, 0.2]`` means 40% none, 40% one and 20% two. + 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..44a095d --- /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_weights: [0.5, 0.3, 0.2] + videos_per_vertical_weights: [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_weights: [0.4, 0.35, 0.2, 0.05] + videos_per_vertical_weights: [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_weights: [0.35, 0.35, 0.2, 0.1] + videos_per_vertical_weights: [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..329f855 --- /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_weights: [0.5, 0.3, 0.2] + videos_per_vertical_weights: [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_weights: [0.4, 0.35, 0.2, 0.05] + videos_per_vertical_weights: [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_weights: [0.35, 0.35, 0.2, 0.1] + videos_per_vertical_weights: [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..7c910af --- /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_weights: [0.5, 0.3, 0.2] + videos_per_vertical_weights: [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_weights: [0.4, 0.4, 0.2] + videos_per_vertical_weights: [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/config.py b/xapi_db_load/journeys/config.py new file mode 100644 index 0000000..8f2cc48 --- /dev/null +++ b/xapi_db_load/journeys/config.py @@ -0,0 +1,81 @@ +""" +Checks for ``journeys`` config values, so a bad value fails before any output is written. + +Each check returns the value (converted where noted) or raises ValueError naming the key. +""" + +from typing import ( + Any, + Dict, + List, + Optional, + Tuple, +) + + +def _is_number(value: Any) -> bool: + return isinstance(value, (int, float)) and not isinstance(value, bool) + + +def require(conf: Dict, key: str, prefix: str = "") -> Any: + """Return ``conf[key]``, or raise ValueError if it's missing.""" + if key not in conf: + raise ValueError(f"Missing required setting: {prefix}{key}") + return conf[key] + + +def check_int(name: str, value: Any, minimum: int = 0) -> int: + """Check ``value`` is an integer of at least ``minimum``.""" + if not isinstance(value, int) or isinstance(value, bool) or value < minimum: + raise ValueError(f"{name} must be an integer of at least {minimum}, got {value!r}.") + return value + + +def check_number( + name: str, value: Any, minimum: float = 0.0, maximum: Optional[float] = None +) -> float: + """Check ``value`` is a number in [``minimum``, ``maximum``] and return it as a float.""" + if not _is_number(value) or value < minimum or (maximum is not None and value > maximum): + bounds = f"between {minimum} and {maximum}" if maximum is not None else f">= {minimum}" + raise ValueError(f"{name} must be a number {bounds}, got {value!r}.") + return float(value) + + +def check_fraction(name: str, value: Any) -> float: + """Check ``value`` is a probability or fraction between 0 and 1.""" + return check_number(name, value, 0.0, 1.0) + + +def check_bool(name: str, value: Any) -> bool: + """Check ``value`` is true or false.""" + if not isinstance(value, bool): + raise ValueError(f"{name} must be true or false, got {value!r}.") + return value + + +def check_range(name: str, value: Any, minimum: int = 0) -> Tuple[int, int]: + """Check ``value`` is ``[low, high]`` integers with ``minimum <= low <= high``.""" + if ( + not isinstance(value, (list, tuple)) + or len(value) != 2 + or not all(isinstance(v, int) and not isinstance(v, bool) for v in value) + or not minimum <= value[0] <= value[1] + ): + raise ValueError( + f"{name} must be [low, high] integers with {minimum} <= low <= high, got {value!r}." + ) + return value[0], value[1] + + +def check_weights(name: str, value: Any) -> List[float]: + """Check ``value`` is a non-empty list of non-negative numbers that don't all equal 0.""" + if ( + not isinstance(value, (list, tuple)) + or not value + or not all(_is_number(v) and v >= 0 for v in value) + or not sum(value) + ): + raise ValueError( + f"{name} must be a list of non-negative weights, not all 0, got {value!r}." + ) + return [float(v) for v in value] diff --git a/xapi_db_load/journeys/generate.py b/xapi_db_load/journeys/generate.py new file mode 100644 index 0000000..5e23751 --- /dev/null +++ b/xapi_db_load/journeys/generate.py @@ -0,0 +1,324 @@ +""" +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 config as cfg +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._check_config() + self.rng = random.Random(conf.get("seed", 0)) + now = conf.get("now") + try: + self.now = ( + datetime.datetime.fromisoformat(str(now)).replace(tzinfo=datetime.UTC) + if now + else datetime.datetime.now(datetime.UTC).replace(microsecond=0) + ) + except ValueError as e: + raise ValueError(f"now must be a UTC time like '2026-10-08 12:00:00': {e}") from e + self.behavior = Behavior.from_config(conf.get("behavior")) + self.templates = { + name: CourseTemplate.from_config(t, name) + for name, t in conf["course_templates"].items() + } + self.actors: List[JourneyActor] = [] + self.courses: List[JourneyCourse] = [] + self.course_windows: Dict[str, tuple] = {} + + def _check_config(self) -> None: + """Check the top-level settings, so a bad value fails before any file is written.""" + conf = self.conf + cfg.check_int("seed", conf.get("seed", 0)) + cfg.check_int("num_actors", cfg.require(conf, "num_actors"), 1) + cfg.check_int("first_user_id", conf.get("first_user_id", 1), 1) + cfg.check_int("num_organizations", conf.get("num_organizations", 3), 1) + cfg.check_int("num_actor_profile_changes", conf.get("num_actor_profile_changes", 1)) + cfg.check_int("course_publishes", conf.get("course_publishes", 1), 1) + for key in ("mbox_actor_fraction", "running_course_fraction", + "deleted_unit_course_fraction"): + if key in conf: + cfg.check_fraction(key, conf[key]) + cfg.check_bool("sort_events", conf.get("sort_events", False)) + prefix = conf.get("org_prefix", "Org") + if not isinstance(prefix, str) or not prefix.isalnum(): + raise ValueError(f"org_prefix must be letters and digits, got {prefix!r}.") + window_days = cfg.check_number("window_days", conf.get("window_days", 300), 1) + 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.") + length_days = cfg.check_number( + "course_length_days", conf.get("course_length_days", 120), 1 + ) + if length_days > window_days: + raise ValueError("course_length_days must not be more than window_days.") + + templates = cfg.require(conf, "course_templates") + if not isinstance(templates, dict) or not templates: + raise ValueError("course_templates must map template names to templates.") + courses = cfg.require(conf, "courses") + if not isinstance(courses, list) or not courses: + raise ValueError("courses must be a list of {template, count, learners} entries.") + for i, entry in enumerate(courses): + prefix = f"courses[{i}]." + if not isinstance(entry, dict): + raise ValueError(f"courses[{i}] must be a mapping, got {entry!r}.") + if cfg.require(entry, "template", prefix) not in templates: + raise ValueError(f"{prefix}template {entry['template']!r} is not in course_templates.") + cfg.check_int(prefix + "count", cfg.require(entry, "count", prefix)) + cfg.check_range(prefix + "learners", cfg.require(entry, "learners", prefix)) + cfg.check_range(prefix + "runs", entry.get("runs", [1, 1]), 1) + + # ---- setup ------------------------------------------------------------------------------ + + def setup(self) -> None: + """Create actors and courses.""" + mbox_fraction = self.conf.get("mbox_actor_fraction", 0.0) + first_user_id = self.conf.get("first_user_id", 1) + for i in range(self.conf["num_actors"]): + username = f"actor_{first_user_id - 1 + i}" + self.actors.append( + JourneyActor( + user_id=first_user_id + i, + external_id=seeded_uuid(self.rng), + username=username, + name=f"Actor {first_user_id - 1 + i}", + email=f"{username}@aspects.invalid", + uses_mbox=self.rng.random() < mbox_fraction, + ) + ) + + prefix = self.conf.get("org_prefix", "Org") + orgs = [f"{prefix}{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) + # A run that stops partway must not leave an earlier run's manifest describing it. + manifest_path = os.path.join(output_dir, "manifest.json") + if os.path.exists(manifest_path): + os.remove(manifest_path) + 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(manifest_path, "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 + most_active: 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 and best_pages: + most_active.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, + "most_active_learner_per_course": most_active} diff --git a/xapi_db_load/journeys/load.py b/xapi_db_load/journeys/load.py new file mode 100644 index 0000000..b042190 --- /dev/null +++ b/xapi_db_load/journeys/load.py @@ -0,0 +1,243 @@ +""" +Load a generated journeys dataset into an Aspects install. + +Event sink rows and the expected results are inserted straight into ClickHouse. The xAPI events +go through the configured ``backend``, so a test can cover the same path real events take: + +- ``clickhouse``: inserted straight into ``xapi_events_all``. +- ``ralph``: POSTed to the LRS, without the duplicate events, which an LRS rejects. +- ``vector``: logged to stdout in the format Vector reads from the LMS logs. + +The expected results go into ``expected_db_name`` (default ``journeys``) as +``expected_engagement`` and ``expected_video_seconds``, with ``event_ids`` listing every event in +the dataset. Ralph and Vector insert asynchronously, so ``wait`` polls until every event id is in +``xapi_events_all``. +""" + +import csv +import gzip +import json +import logging +import os +import sys +import time +from typing import ( + Any, + Dict, + Iterator, + List, +) + +import clickhouse_connect +import requests + +from xapi_db_load.journeys import oracle + +# Dataset file -> event sink table. Each file has the table's columns, in order. +SINK_FILES = { + "courses": "course_overviews", + "blocks": "course_blocks", + "external_ids": "external_id", + "user_profiles": "user_profile", +} +XAPI_COLUMNS = ["event_id", "emission_time", "event"] +BACKENDS = ("clickhouse", "ralph", "vector") +EXPECTED_TABLES = { + "expected_engagement": ( + oracle.ENGAGEMENT_COLUMNS, + {"done": "UInt32", "total": "UInt32", "done_truth": "UInt32"}, + ), + "expected_video_seconds": ( + oracle.VIDEO_SECONDS_COLUMNS, + {c: "UInt32" for c in oracle.VIDEO_SECONDS_COLUMNS if c.endswith(("seconds", "duration", "views"))}, + ), +} +RALPH_BATCH_SIZE = 500 +VECTOR_BATCH_SIZE = 100 +VECTOR_BATCH_PAUSE_SECONDS = 0.1 + + +def _read_gz(path: str) -> bytes: + with gzip.open(path, "rb") as f: + return f.read() + + +def _xapi_rows(path: str) -> Iterator[List[str]]: + with gzip.open(path, "rt", newline="") as f: + yield from csv.reader(f) + + +class JourneyLoader: + """Loads one dataset directory, as written by the ``journeys`` command.""" + + def __init__(self, config: Dict, dataset_dir: str, logger: logging.Logger, client: Any = None): + self.config = config + self.dataset_dir = dataset_dir + self.log = logger + self.backend = config.get("backend", "clickhouse") + if self.backend not in BACKENDS: + raise ValueError(f"backend must be one of {', '.join(BACKENDS)}, got {self.backend!r}.") + try: + with open(self._path("manifest.json")) as f: + self.manifest = json.load(f) + except FileNotFoundError as e: + raise ValueError( + f"{dataset_dir} has no manifest.json; generate it with the journeys command first." + ) from e + self.xapi_db = config.get("db_name", "xapi") + self.sink_db = config.get("event_sink_db_name", "event_sink") + self.expected_db = config.get("expected_db_name", "journeys") + self.client = client or self._connect() + + def _connect(self): + port = self.config.get("db_port", 18123) + return clickhouse_connect.get_client( + host=self.config.get("db_host", "localhost"), + port=port, + username=self.config.get("db_username", "default"), + password=self.config.get("db_password", ""), + secure=str(port).endswith(("443", "440")), + settings={"date_time_input_format": "best_effort"}, + ) + + def _path(self, name: str) -> str: + return os.path.join(self.dataset_dir, name) + + def run(self, wait_seconds: float = 0) -> str: + """Load everything and return a one-line summary.""" + self.load_event_sink() + n_events = self.load_expected() + self.load_xapi() + self.wait(n_events, wait_seconds) + return ( + f"Loaded {n_events} xAPI events for {self.manifest['num_enrollments']} enrollments " + f"in {self.manifest['num_course_runs']} course runs via {self.backend}; expected " + f"results are in {self.expected_db}." + ) + + def load_event_sink(self) -> None: + """Insert the course, block, user and profile rows, and reload the dictionaries on them.""" + for name, table in SINK_FILES.items(): + self.log.info(f"Inserting {name} into {self.sink_db}.{table}") + self.client.raw_insert( + f"{self.sink_db}.{table}", insert_block=_read_gz(self._path(f"{name}.csv.gz")), + fmt="CSV", + ) + # Aspects looks up course names and users through dictionaries that otherwise only + # refresh every few minutes. + self.client.command("system reload dictionaries") + + def load_expected(self) -> int: + """Replace the expected-results tables; return the number of distinct events.""" + self.client.command(f"create database if not exists {self.expected_db}") + for table, (columns, types) in EXPECTED_TABLES.items(): + schema = ", ".join(f"{c} {types.get(c, 'String')}" for c in columns) + self.client.command(f"drop table if exists {self.expected_db}.{table}") + self.client.command( + f"create table {self.expected_db}.{table} ({schema}) " + f"engine = MergeTree order by tuple()" + ) + self.client.raw_insert( + f"{self.expected_db}.{table}", insert_block=_read_gz(self._path(f"{table}.csv.gz")), + fmt="CSVWithNames", + ) + ids = sorted({row[0] for row in _xapi_rows(self._path("xapi.csv.gz"))}) + self.client.command(f"drop table if exists {self.expected_db}.event_ids") + self.client.command( + f"create table {self.expected_db}.event_ids (event_id UUID) " + f"engine = MergeTree order by event_id" + ) + self.client.raw_insert( + f"{self.expected_db}.event_ids", insert_block="\n".join(ids).encode(), fmt="CSV" + ) + return len(ids) + + def load_xapi(self) -> None: + """Send the xAPI events through the configured backend.""" + path = self._path("xapi.csv.gz") + self.log.info(f"Sending xAPI events via {self.backend}") + if self.backend == "clickhouse": + self.client.raw_insert( + f"{self.xapi_db}.xapi_events_all", column_names=XAPI_COLUMNS, + insert_block=_read_gz(path), fmt="CSV", + ) + elif self.backend == "ralph": + self._post_to_ralph(path) + else: + self._log_for_vector(path) + + def _post_to_ralph(self, path: str) -> None: + """POST the events to the LRS in batches, skipping repeated event ids.""" + seen = set() + batch: List[Dict] = [] + for event_id, _, event in _xapi_rows(path): + if event_id in seen: + continue + seen.add(event_id) + batch.append(json.loads(event)) + if len(batch) == RALPH_BATCH_SIZE: + self._post_batch(batch) + batch = [] + if batch: + self._post_batch(batch) + + def _post_batch(self, batch: List[Dict]) -> None: + """POST one batch of statements, failing on any error response.""" + resp = requests.post( + self.config["lrs_url"], + auth=(self.config["lrs_username"], self.config["lrs_password"]), + json=batch, + timeout=self.config.get("lrs_request_timeout", 120), + ) + resp.raise_for_status() + + def _log_for_vector(self, path: str) -> None: + """ + Log the events in the line format Vector reads from the LMS. + + Pauses between batches: printing the whole dataset in a few seconds is far faster than + an LMS logs, and a log line has been lost in CI at that rate. + """ + batch_size = self.config.get("vector_batch_size", VECTOR_BATCH_SIZE) + pause = self.config.get("vector_batch_pause_seconds", VECTOR_BATCH_PAUSE_SECONDS) + for i, (_, _, event) in enumerate(_xapi_rows(path), start=1): + # The smallest line format that passes Vector's LMS tracking log regex. + sys.stdout.write(f" [xapi_tracking] [] {event}\n") + if i % batch_size == 0: + sys.stdout.flush() + time.sleep(pause) + sys.stdout.flush() + + def _log_missing(self, limit: int = 10) -> None: + """Log the first events that never arrived, as sent, to help find where they were lost.""" + missing = self.client.query( + f"select toString(event_id) from {self.expected_db}.event_ids where event_id not in " + f"(select event_id from {self.xapi_db}.xapi_events_all) order by event_id limit {limit}" + ).result_rows + missing_ids = {row[0] for row in missing} + for event_id, emission_time, event in _xapi_rows(self._path("xapi.csv.gz")): + if event_id in missing_ids: + missing_ids.discard(event_id) + self.log.error( + f"Missing event {event_id} ({emission_time}, {len(event)} characters): {event}" + ) + + def wait(self, n_events: int, wait_seconds: float, poll_seconds: float = 5) -> None: + """Wait until every event in the dataset is in xapi_events_all.""" + deadline = time.monotonic() + wait_seconds + while True: + arrived = self.client.command( + f"select count() from {self.expected_db}.event_ids where event_id in " + f"(select event_id from {self.xapi_db}.xapi_events_all)" + ) + if int(arrived) >= n_events: + self.log.info(f"All {n_events} xAPI events are in {self.xapi_db}.xapi_events_all") + return + if time.monotonic() >= deadline: + self._log_missing() + raise TimeoutError( + f"Only {arrived} of {n_events} xAPI events reached " + f"{self.xapi_db}.xapi_events_all within {wait_seconds} s." + ) + self.log.info(f"{arrived} of {n_events} xAPI events have arrived, waiting") + time.sleep(poll_seconds) diff --git a/xapi_db_load/journeys/oracle.py b/xapi_db_load/journeys/oracle.py new file mode 100644 index 0000000..e974cfc --- /dev/null +++ b/xapi_db_load/journeys/oracle.py @@ -0,0 +1,156 @@ +""" +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", + "observable_max_views", +] + + +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..b1cabeb --- /dev/null +++ b/xapi_db_load/journeys/simulate.py @@ -0,0 +1,480 @@ +""" +Simulate learners moving through journey courses, emitting xAPI events and recording what happened. +""" + +import datetime +import json +import math +import random +from collections import Counter +from dataclasses import ( + dataclass, + field, +) +from typing import ( + Dict, + List, + Optional, + Set, + Tuple, +) + +from xapi_db_load.journeys import config as cfg +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 + ) + + +VIDEO_ACTIONS = ("to_end", "pause", "seek", "abandon", "terminate") + + +@dataclass +class Behavior: + """ + Settings that shape learner behavior: the ``behavior`` section of a journeys config. + + Every field can be set there; missing ones keep these defaults. ``*_prob`` and + ``*_fraction`` values are between 0 and 1. + """ + + # Chance an enrolled learner opens the course at all; the rest only register. + start_prob: float = 0.8 + # Share of learners who work through every subsection, ignoring continue_prob. + completer_fraction: float = 0.15 + # Chance a non-completer goes on to the next subsection after finishing one. + continue_prob: float = 0.9 + # Chance of jumping over the next unit with the unit tabs instead of "next unit". + unit_skip_prob: float = 0.08 + # Chance of going back one unit and forward again before moving on. + revisit_prob: float = 0.05 + # Chance the learner closes the page on the last unit of a session instead of navigating + # away, so that unit gets no navigation event and doesn't count as viewed. + exit_without_nav_prob: float = 0.7 + # Chance of attempting each problem in a unit. + problem_attempt_prob: float = 0.8 + # Most submissions per problem; each learner stops at a random count up to this, or when + # they get it right. + max_attempts: int = 3 + # Chance of watching each video in a unit. + video_watch_prob: float = 0.75 + # Chance of watching the same video again 5 minutes later. + second_session_prob: float = 0.15 + # Mean days between subsections (exponentially distributed). + days_between_subsections: float = 2.0 + # Weights for what ends each stretch of video playback: play to the end, pause, seek, + # close the tab with no event (abandon), or stop the player (terminate). + 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, + } + ) + # Chance of pressing play again after a pause. + 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 (oracle) scenario: + # chance a resume after a pause happens at the same instant as the pause, + same_time_resume_prob: float = 0.0 + # and chance each event is emitted twice with the same event id. + 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.""" + behavior = cls() + if conf is not None and not isinstance(conf, dict): + raise ValueError(f"behavior must be a mapping, got {conf!r}.") + for key, value in (conf or {}).items(): + if not hasattr(behavior, key): + raise ValueError(f"Unknown behavior setting: {key}") + name = f"behavior.{key}" + if key == "video_actions": + if not isinstance(value, dict) or not set(value) <= set(VIDEO_ACTIONS): + raise ValueError( + f"{name} must map some of {', '.join(VIDEO_ACTIONS)} to weights, " + f"got {value!r}." + ) + weights = cfg.check_weights(name, list(value.values())) + value = dict(zip(value, weights)) + elif key == "max_attempts": + value = cfg.check_int(name, value, 1) + elif key.endswith(("_prob", "_fraction")): + value = cfg.check_fraction(name, value) + elif key == "days_between_subsections": + value = cfg.check_number(name, value, 0.001) + else: + value = cfg.check_number(name, value) + setattr(behavior, key, value) + return behavior + + +@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) + + @staticmethod + def _max_views(segments) -> int: + """Return how many times the most-watched second was watched, 0 if none was.""" + views: Counter = Counter() + for a, b in segments: + views.update(range(a + 1, b + 1)) + return max(views.values(), default=0) + + def summary(self) -> Dict[str, int]: + """Return watched seconds, actual and observable from events, and the most views.""" + 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), + "observable_max_views": self._max_views(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.behavior = 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.behavior.duplicate_event_prob: + self.events.append(dict(row)) + return event_id + + def _has_happened(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.behavior.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.behavior.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.behavior.start_prob: + return self.events, truth + + completer = self.rng.random() < self.behavior.completer_fraction + t = enroll_time + datetime.timedelta(minutes=self.rng.randint(1, 600)) + sequentials = list(course.sequentials()) + + # The last subsection always ends the loop with a break. + for i, seq in enumerate(sequentials): # pragma: no branch + if not self._has_happened(t): + break + last = i == len(sequentials) - 1 + stopping = last or (not completer and self.rng.random() >= self.behavior.continue_prob) + t = self._run_subsection(seq, t, truth, stopping) + if stopping: + break + gap_days = self.rng.expovariate(1 / self.behavior.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.behavior.exit_without_nav_prob: + break + + t += datetime.timedelta(seconds=self.rng.randint(2, 20)) + if not final_unit and self.rng.random() < self.behavior.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.behavior.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._has_happened(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._has_happened(t): + break + if child.block_type == "video" and self.rng.random() < self.behavior.video_watch_prob: + t = self._watch_video(child, t, truth) + if self.rng.random() < self.behavior.second_session_prob: + t = self._watch_video(child, t + datetime.timedelta(minutes=5), truth) + elif child.block_type == "problem" and self.rng.random() < self.behavior.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.behavior.max_attempts) + 1): + t += datetime.timedelta(seconds=self.rng.randint(20, 300)) + if not self._has_happened(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._has_happened(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.behavior.video_actions) + weights = list(self.behavior.video_actions.values()) + completed_sent = False + + def emit(verb, when, position=None, **kw): + event_id = self._emit( + when, st.video_statement, 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._has_happened(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._has_happened(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.behavior.resume_after_pause_prob: + break + if self.rng.random() >= self.behavior.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..ae7f9fe --- /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_statement( # 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..2327c2a --- /dev/null +++ b/xapi_db_load/journeys/structure.py @@ -0,0 +1,251 @@ +""" +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, + Tuple, +) + +from xapi_db_load.journeys import config as cfg + + +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_weights: List[float] + videos_per_vertical_weights: List[float] + graded_sequential_fraction: float + video_length: Tuple[int, int] + + @classmethod + def from_config(cls, conf: Dict, name: str = "") -> "CourseTemplate": + """Build a template from a YAML ``course_templates`` entry, checking every value.""" + prefix = f"course_templates.{name}." if name else "" + + def required_range(key: str, minimum: int) -> Tuple[int, int]: + return cfg.check_range(prefix + key, cfg.require(conf, key, prefix), minimum) + + def required_weights(key: str) -> List[float]: + return cfg.check_weights(prefix + key, cfg.require(conf, key, prefix)) + + return cls( + chapters=required_range("chapters", 1), + sequentials_per_chapter=required_range("sequentials_per_chapter", 1), + verticals_per_sequential=required_range("verticals_per_sequential", 1), + problems_per_vertical_weights=required_weights("problems_per_vertical_weights"), + videos_per_vertical_weights=required_weights("videos_per_vertical_weights"), + graded_sequential_fraction=cfg.check_fraction( + prefix + "graded_sequential_fraction", conf.get("graded_sequential_fraction", 0.3) + ), + video_length=cfg.check_range( + prefix + "video_length", conf.get("video_length", [60, 900]), 1 + ), + ) + + +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_weights)), t.videos_per_vertical_weights + )[0] + n_problems = rng.choices( + range(len(t.problems_per_vertical_weights)), t.problems_per_vertical_weights + )[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) -> 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. + """ + # Templates have at least one chapter and one sequential each (see CourseTemplate). + 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..14e9441 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,8 @@ import yaml from xapi_db_load.constants import DEFAULT_LMS_URL +from xapi_db_load.journeys.generate import JourneyGenerator +from xapi_db_load.journeys.load import JourneyLoader from xapi_db_load.ui.text_ui import TextUI _ENV_VAR_OVERRIDES = { @@ -111,7 +114,94 @@ 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}" + ) + + +@click.command() +@click.option( + "--config_file", + help=( + "Configuration file with the ClickHouse settings and backend (clickhouse, ralph or " + "vector) to load with, as for load-db." + ), + required=True, + type=click.Path(exists=True, dir_okay=False, file_okay=True, writable=False), +) +@click.option( + "--dataset_dir", + help="Dataset written by the journeys command. Defaults to journeys.output_dir.", + default=None, +) +@click.option( + "--wait_seconds", + help="How long to wait for Ralph or Vector to insert every event before failing.", + type=float, + default=600, + show_default=True, +) +def load_journeys(config_file: str, dataset_dir: str | None, wait_seconds: float): + """ + Load a journeys dataset and its expected results into ClickHouse. + """ + logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s") + config = get_config(config_file) + dataset_dir = dataset_dir or (config.get("journeys") or {}).get("output_dir") + if not dataset_dir: + raise click.UsageError("Set --dataset_dir or journeys.output_dir in the config.") + try: + loader = JourneyLoader(config, dataset_dir, logging.getLogger("journeys")) + except ValueError as e: + raise click.UsageError(str(e)) from e + click.echo(loader.run(wait_seconds)) + + cli.add_command(load_db) +cli.add_command(journeys) +cli.add_command(load_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..d7e344e --- /dev/null +++ b/xapi_db_load/tests/test_journeys.py @@ -0,0 +1,549 @@ +""" +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, + JourneyActor, + 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() + views = collections.defaultdict(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)) + views[key].update(range(int(position(cur)) + 1, int(position(nxt)) + 1)) + + rows = list(_dict_rows(out / "expected_video_seconds.csv.gz")) + expected = { + (r["course_key"], r["actor_id"], r["video_block_id"]): int(r["observable_seconds"]) + for r in rows + } + max_views = { + (r["course_key"], r["actor_id"], r["video_block_id"]): int(r["observable_max_views"]) + for r in rows + } + assert max_views == {k: max(views[k].values(), default=0) for k in max_views} + assert any(v > 1 for v in max_views.values()), "some learner should rewatch a second" + 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 _set_template(key, value): + def change(conf): + conf["course_templates"]["small"][key] = value + return change + + +def _set_behavior(key, value): + def change(conf): + conf.setdefault("behavior", {})[key] = value + return change + + +def _set_course(key, value): + def change(conf): + conf["courses"][0][key] = value + return change + + +def _drop(key): + def change(conf): + del conf[key] + return change + + +@pytest.mark.parametrize( + "change, message", + [ + (lambda c: c.update(window_days="300"), "window_days must be a number"), + (lambda c: c.update(num_actors=0), "num_actors must be an integer of at least 1"), + (lambda c: c.update(sort_events="yes"), "sort_events must be true or false"), + (lambda c: c.update(mbox_actor_fraction=1.5), "mbox_actor_fraction must be a number"), + (lambda c: c.update(now="tomorrow"), "now must be a UTC time"), + (lambda c: c.update(org_prefix="Org-"), "org_prefix must be letters and digits"), + (lambda c: c.update(first_user_id=0), "first_user_id must be an integer of at least 1"), + (_drop("num_actors"), "Missing required setting: num_actors"), + (_set_behavior("days_between_subsections", "2.0"), + "behavior.days_between_subsections must be a number"), + (_set_behavior("max_attempts", 0), "behavior.max_attempts must be an integer"), + (_set_behavior("video_actions", {"rewind": 1}), "behavior.video_actions must map"), + (_set_template("problems_per_vertical_weights", [0, 0]), + "course_templates.small.problems_per_vertical_weights must be a list"), + (_set_template("chapters", [3, 1]), "course_templates.small.chapters must be [low, high]"), + (_set_course("template", "huge"), "courses[0].template 'huge' is not in course_templates"), + (_set_course("learners", [5]), "courses[0].learners must be [low, high]"), + (lambda c: c.update(behavior="fast"), "behavior must be a mapping"), + (lambda c: c.update(course_templates={}), "course_templates must map template names"), + (lambda c: c.update(courses=[]), "courses must be a list"), + (lambda c: c.update(courses=["small"]), "courses[0] must be a mapping"), + ], +) +def test_config_errors(change, message): + conf = yaml.safe_load(CONFIG.read_text())["journeys"] + change(conf) + with pytest.raises(ValueError) as e: + JourneyGenerator(conf, "http://localhost:18000", logging.getLogger()) + assert message in str(e.value) + + +def test_behavior_settings_are_converted(): + behavior = Behavior.from_config({ + "video_actions": {"to_end": 1, "pause": 3}, "recent_hours": 6, "max_attempts": 2, + "days_between_subsections": 1, "continue_prob": 1, + }) + assert behavior.video_actions == {"to_end": 1.0, "pause": 3.0} + assert (behavior.recent_hours, behavior.max_attempts) == (6.0, 2) + assert (behavior.days_between_subsections, behavior.continue_prob) == (1.0, 1.0) + + +def test_optional_settings_have_defaults(tmp_path): + conf = yaml.safe_load(CONFIG.read_text())["journeys"] + for key in ("mbox_actor_fraction", "running_course_fraction", "deleted_unit_course_fraction", + "behavior"): + del conf[key] + conf["now"] = NOW + manifest = JourneyGenerator(conf, "http://localhost:18000", logging.getLogger()).run( + str(tmp_path) + ) + assert manifest["mbox_actors"] == 0 and manifest["courses_with_deleted_unit"] == 0 + + +def test_activity_stops_at_now(): + """ + Learners whose journeys run past "now" stop there, at every step: no event is dated after + now, and the expected results only count what those events show. + """ + rng = random.Random(7) + template = CourseTemplate.from_config({ + "chapters": [2, 2], "sequentials_per_chapter": [2, 2], "verticals_per_sequential": [3, 3], + "problems_per_vertical_weights": [0, 1], "videos_per_vertical_weights": [0, 1], + "video_length": [60, 120], + }) + course = JourneyCourse(rng, "Org", "C", "run", "t", template, "http://lms") + actor = JourneyActor(1, "e", "u", "U", "u@example.com") + enrolled = datetime.datetime(2026, 1, 1, tzinfo=datetime.UTC) + pages_by_now = [] + for minutes in range(0, 2 * 24 * 60, 4): + now = enrolled + datetime.timedelta(minutes=minutes, seconds=minutes % 59) + sim = Simulator(rng, Behavior(start_prob=1, completer_fraction=1), now) + events, truth = sim.run_enrollment(course, actor, enrolled) + assert all(e["emission_time"] <= now for e in events) + statements = [json.loads(e["event"]) for e in events] + navigated = {_location(s["object"]["id"]) for s in statements if s["verb"]["id"] == st.VERB_NAVIGATED} + evaluated = {_location(s["object"]["id"]) for s in statements + if s["verb"]["id"] == st.VERB_EVALUATED} + assert truth.pages_viewed <= navigated + assert truth.problems_attempted <= evaluated + pages_by_now.append(len(truth.pages_viewed)) + # From nothing yet to the whole course, as "now" moves later. + assert pages_by_now[0] == 0 and max(pages_by_now) >= 10 + + +def test_org_prefix_and_first_user_id(tmp_path): + _generate(tmp_path, org_prefix="Journey", first_user_id=100001) + orgs = {row[0] for row in _rows(tmp_path / "courses.csv.gz")} + assert orgs and all(o.startswith("Journey") for o in orgs) + users = [(int(r[3]), r[2]) for r in _rows(tmp_path / "external_ids.csv.gz")] + assert users[0] == (100001, "actor_100000") + assert min(u for u, _ in users) == 100001 + + +def test_failed_run_removes_old_manifest(tmp_path, monkeypatch): + _generate(tmp_path) + assert (tmp_path / "manifest.json").exists() + + def fail(self, writers): + raise RuntimeError("stopped partway") + + monkeypatch.setattr(JourneyGenerator, "_write_journeys", fail) + with pytest.raises(RuntimeError): + _generate(tmp_path) + assert not (tmp_path / "manifest.json").exists() + + +def test_most_active_learner_skips_courses_without_page_views(dataset, tmp_path): + _, manifest = dataset + assert manifest["most_active_learner_per_course"] + assert all(m["pages_viewed_fraction"] > 0 for m in manifest["most_active_learner_per_course"]) + assert not _generate(tmp_path, behavior={"start_prob": 0.0})["most_active_learner_per_course"] + + +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, + "observable_max_views": 1, + } + + +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, + "observable_max_views": 1, + } + + +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_weights": [1], "videos_per_vertical_weights": [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() == [] + + +class LowestRandom(random.Random): + """Always picks the lowest integer, so the simulator's timings are known.""" + + def randint(self, a, b): + return a + + +def _one_block_course(kind): + """Return a one-unit course with a single problem or video, and that block.""" + template = CourseTemplate.from_config({ + "chapters": [1, 1], "sequentials_per_chapter": [1, 1], "verticals_per_sequential": [1, 1], + "problems_per_vertical_weights": [0, 1] if kind == "problem" else [1], + "videos_per_vertical_weights": [0, 1] if kind == "video" else [1], + "video_length": [600, 600], + }) + course = JourneyCourse(random.Random(1), "Org", "C", "run", "t", template, "http://lms") + return course, next(b for b in course.all_blocks() if b.block_type == kind) + + +def test_problem_submitted_just_before_now_is_not_counted(): + """The attempt is logged at 20 s and graded at 21 s; with now in between, it isn't counted.""" + course, problem = _one_block_course("problem") + start = datetime.datetime(2026, 1, 1, tzinfo=datetime.UTC) + sim = Simulator(LowestRandom(1), Behavior(), start + datetime.timedelta(seconds=20.5)) + sim._course = course # pylint: disable=protected-access + sim.events = [] + truth = EnrollmentTruth() + sim._attempt_problem(problem, start, truth) # pylint: disable=protected-access + assert [json.loads(e["event"])["verb"]["id"] for e in sim.events] == [st.VERB_ATTEMPTED] + assert not truth.problems_attempted + + +def test_video_paused_at_now_counts_only_the_seconds_before(): + """Play at 1 s, pause at 2 s, resume at 4 s: with now at 3 s, one second was watched.""" + course, video = _one_block_course("video") + start = datetime.datetime(2026, 1, 1, tzinfo=datetime.UTC) + behavior = Behavior(video_actions={"pause": 1}, resume_after_pause_prob=1) + sim = Simulator(LowestRandom(1), behavior, start + datetime.timedelta(seconds=3)) + sim._course = course # pylint: disable=protected-access + sim.events = [] + truth = EnrollmentTruth() + sim._watch_video(video, start, truth) # pylint: disable=protected-access + vt = truth.videos[video.location] + assert vt.truth == [(0, 1)] + assert [verb for _, _, verb, _ in vt.stream] == [st.VERB_PLAYED, st.VERB_PAUSED] + + +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": "300"}, "window_days must be a number"), + ({"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 diff --git a/xapi_db_load/tests/test_journeys_load.py b/xapi_db_load/tests/test_journeys_load.py new file mode 100644 index 0000000..820d6f6 --- /dev/null +++ b/xapi_db_load/tests/test_journeys_load.py @@ -0,0 +1,206 @@ +""" +Tests for loading a journeys dataset, with a fake ClickHouse client. +""" + +import gzip +import json +import logging +import pathlib + +import pytest +import yaml +from click.testing import CliRunner + +from xapi_db_load.journeys import load as journeys_load +from xapi_db_load.journeys.generate import JourneyGenerator +from xapi_db_load.journeys.load import JourneyLoader +from xapi_db_load.main import load_journeys + +CONFIG = pathlib.Path(__file__).parents[2] / "example_configs" / "journeys_oracle.yaml" +NOW = "2026-10-08 12:00:00" + + +class FakeClient: + """Records inserts and commands; reports ``arrived`` events as present.""" + + def __init__(self, arrived=None, missing=()): + self.inserts = {} + self.commands = [] + self.arrived = arrived + self.missing = missing + + def query(self, sql): + """Return the ``missing`` event ids.""" + assert "not in" in sql + return type("Result", (), {"result_rows": [(m,) for m in self.missing]}) + + def raw_insert(self, table, column_names=None, insert_block=None, fmt=None): + self.inserts[table] = (column_names, insert_block, fmt) + + def command(self, sql): + self.commands.append(sql) + if sql.startswith("select count()"): + return self.arrived + return None + + +@pytest.fixture(name="dataset", scope="module") +def fixture_dataset(tmp_path_factory): + """Generate the oracle dataset; return its dir, event row count and distinct event count.""" + out = tmp_path_factory.mktemp("journeys") + conf = yaml.safe_load(CONFIG.read_text())["journeys"] + conf["now"] = NOW + JourneyGenerator(conf, "http://localhost:18000", logging.getLogger()).run(str(out)) + with gzip.open(out / "xapi.csv.gz", "rt") as f: + lines = f.read().splitlines() + unique_ids = {line.split(",", 1)[0] for line in lines} + return out, len(lines), len(unique_ids) + + +def _loader(dataset, backend="clickhouse", arrived=None, **config): + """Return a loader for the dataset with a fake client reporting ``arrived`` events.""" + out, _, n_unique = dataset + client = FakeClient(n_unique if arrived is None else arrived) + config = {"backend": backend, "lrs_url": "http://ralph/xAPI/statements/", + "lrs_username": "u", "lrs_password": "p", **config} + return JourneyLoader(config, str(out), logging.getLogger(), client=client), client + + +def test_clickhouse_backend_loads_everything(dataset): + loader, client = _loader(dataset) + summary = loader.run() + assert set(client.inserts) == { + "event_sink.course_overviews", "event_sink.course_blocks", "event_sink.external_id", + "event_sink.user_profile", "journeys.expected_engagement", + "journeys.expected_video_seconds", "journeys.event_ids", "xapi.xapi_events_all", + } + columns, block, fmt = client.inserts["xapi.xapi_events_all"] + assert columns == ["event_id", "emission_time", "event"] and fmt == "CSV" + assert block == gzip.open(dataset[0] / "xapi.csv.gz").read() + assert client.inserts["journeys.expected_engagement"][2] == "CSVWithNames" + assert len(client.inserts["journeys.event_ids"][1].decode().splitlines()) == dataset[2] + assert f"Loaded {dataset[2]} xAPI events" in summary + assert "system reload dictionaries" in client.commands + + +def test_database_names_come_from_config(dataset): + loader, client = _loader( + dataset, db_name="openedx", event_sink_db_name="sink", expected_db_name="answers" + ) + loader.run() + assert "openedx.xapi_events_all" in client.inserts + assert "sink.course_blocks" in client.inserts + assert "answers.expected_video_seconds" in client.inserts + + +def test_ralph_backend_posts_each_event_once(dataset, monkeypatch): + posted = [] + + class Response: + def raise_for_status(self): + pass + + def post(url, auth, json, timeout): # pylint: disable=redefined-outer-name,unused-argument + assert url == "http://ralph/xAPI/statements/" and auth == ("u", "p") + assert len(json) <= journeys_load.RALPH_BATCH_SIZE + posted.extend(json) + return Response() + + monkeypatch.setattr(journeys_load.requests, "post", post) + loader, client = _loader(dataset, backend="ralph") + loader.run() + assert "xapi.xapi_events_all" not in client.inserts + _, n_rows, n_unique = dataset + assert n_rows > n_unique, "the oracle dataset should include duplicate events" + assert len(posted) == n_unique == len({s["id"] for s in posted}) + + +def test_ralph_batches_that_end_exactly(dataset, monkeypatch): + posted = [] + + class Response: + def raise_for_status(self): + pass + + monkeypatch.setattr(journeys_load, "RALPH_BATCH_SIZE", dataset[2]) + monkeypatch.setattr( + journeys_load.requests, "post", lambda url, **kw: posted.append(kw["json"]) or Response() + ) + loader, _ = _loader(dataset, backend="ralph") + loader.load_xapi() + assert [len(batch) for batch in posted] == [dataset[2]] + + +def test_connects_with_the_config_settings(dataset, monkeypatch): + calls = [] + monkeypatch.setattr( + journeys_load.clickhouse_connect, "get_client", lambda **kw: calls.append(kw) or FakeClient() + ) + config = {"db_host": "ch", "db_port": 8443, "db_username": "u", "db_password": "p"} + JourneyLoader(config, str(dataset[0]), logging.getLogger()) + assert calls == [{ + "host": "ch", "port": 8443, "username": "u", "password": "p", "secure": True, + "settings": {"date_time_input_format": "best_effort"}, + }] + + +def test_vector_backend_logs_every_event(dataset, capsys, monkeypatch): + pauses = [] + monkeypatch.setattr(journeys_load.time, "sleep", pauses.append) + loader, client = _loader(dataset, backend="vector", vector_batch_pause_seconds=0.5) + loader.run() + assert pauses == [0.5] * (dataset[1] // journeys_load.VECTOR_BATCH_SIZE) + lines = capsys.readouterr().out.splitlines() + assert "xapi.xapi_events_all" not in client.inserts + assert len(lines) == dataset[1] + assert lines[0].startswith(" [xapi_tracking] [] {") + json.loads(lines[0].split("[] ", 1)[1]) + + +def test_wait_times_out_when_events_are_missing(dataset, caplog): + loader, client = _loader(dataset, arrived=3) + with gzip.open(dataset[0] / "xapi.csv.gz", "rt") as f: + missing_id = f.readline().split(",", 1)[0] + client.missing = [missing_id] + with pytest.raises(TimeoutError, match=f"Only 3 of {dataset[2]}"): + loader.wait(dataset[2], wait_seconds=0) + assert f"Missing event {missing_id}" in caplog.text + + +def test_wait_polls_until_events_arrive(dataset, monkeypatch): + loader, client = _loader(dataset, arrived=0) + counts = iter([0, dataset[2]]) + monkeypatch.setattr(client, "command", lambda sql: next(counts)) + monkeypatch.setattr(journeys_load.time, "sleep", lambda s: None) + loader.wait(dataset[2], wait_seconds=60) + + +def test_bad_backend(dataset): + with pytest.raises(ValueError, match="backend must be one of"): + _loader(dataset, backend="kafka") + + +def test_missing_dataset(tmp_path): + with pytest.raises(ValueError, match="no manifest.json"): + JourneyLoader({}, str(tmp_path), logging.getLogger(), client=FakeClient()) + + +def test_cli(dataset, tmp_path, monkeypatch): + out, _, n_unique = dataset + monkeypatch.setattr(JourneyLoader, "_connect", lambda self: FakeClient(n_unique)) + config_file = tmp_path / "config.yaml" + config_file.write_text(yaml.safe_dump({"backend": "clickhouse"})) + result = CliRunner().invoke( + load_journeys, ["--config_file", str(config_file), "--dataset_dir", str(out)], + catch_exceptions=False, + ) + assert result.exit_code == 0, result.output + assert f"Loaded {n_unique} xAPI events" in result.output + + result = CliRunner().invoke(load_journeys, ["--config_file", str(config_file)]) + assert result.exit_code == 2 and "Set --dataset_dir" in result.output + + result = CliRunner().invoke( + load_journeys, ["--config_file", str(config_file), "--dataset_dir", str(tmp_path)] + ) + assert result.exit_code == 2 and "no manifest.json" in result.output