#!/usr/bin/env python3 """HALO Student 2 — single-file matching demo (stdlib only). Skeleton aligned with docs/human/matching-demo-structure.html. Does NOT import HALO backend packages. No DB / HTTP / Event generation. Usage: python matching_demo.py --list python matching_demo.py --scenario A """ from __future__ import annotations import argparse from dataclasses import dataclass, field from datetime import datetime, timedelta, timezone from itertools import combinations from typing import Iterable, Sequence # --------------------------------------------------------------------------- # 1. CONFIG (mirrors app/student2/config.py weights / thresholds) # --------------------------------------------------------------------------- PREP_MINUTES = 15 MIN_EVENT_MINUTES = 30 MIN_GROUP_SIZE = 2 MAX_GROUP_SIZE = 5 MIN_BEACON_CAPACITY = 5 PAIR_SCORE_THRESHOLD = 0.60 GROUP_SCORE_THRESHOLD = 0.70 GROUP_SIZE_BONUS_PER_EXTRA = 0.02 WEIGHT_INTENT = 0.30 WEIGHT_TIME = 0.25 WEIGHT_LOCATION = 0.15 WEIGHT_INTEREST = 0.10 WEIGHT_DEPARTMENT = 0.05 WEIGHT_FRIEND = 0.05 WEIGHT_CAPACITY = 0.10 REASON_INACTIVE_STATUS = "inactive_status" REASON_NO_AVAILABILITY = "NO_AVAILABILITY" REASON_INSUFFICIENT_REMAINING_TIME = "insufficient_remaining_time" REASON_NO_SUITABLE_BEACON = "no_suitable_beacon" REASON_BEACON_PARALLEL_FULL = "beacon_parallel_full" REASON_DUPLICATE_CANDIDATE_EVENT = "duplicate_candidate_event" # --------------------------------------------------------------------------- # 2. TYPES # --------------------------------------------------------------------------- @dataclass(frozen=True) class TimeInterval: start: datetime end: datetime @dataclass(frozen=True) class Candidate: student_id: str intents: tuple[str, ...] intervals: tuple[TimeInterval, ...] expires_at: datetime current_location_id: str | None = None @dataclass(frozen=True) class Beacon: id: str capacity: int max_parallel_events: int supports_micro: bool = True building_id: str | None = None campus_id: str | None = None active_parallel_count: int = 0 @dataclass(frozen=True) class Profile: student_id: str department_id: str | None = None friend_ids: frozenset[str] = field(default_factory=frozenset) @dataclass(frozen=True) class FactorScores: intent_match: float time_match: float location_match: float interest_match: float department_match: float friend_match: float capacity_suitability: float def weighted_total(self) -> float: total = ( self.intent_match * WEIGHT_INTENT + self.time_match * WEIGHT_TIME + self.location_match * WEIGHT_LOCATION + self.interest_match * WEIGHT_INTEREST + self.department_match * WEIGHT_DEPARTMENT + self.friend_match * WEIGHT_FRIEND + self.capacity_suitability * WEIGHT_CAPACITY ) return round(min(max(total, 0.0), 1.0), 4) def as_dict(self) -> dict[str, float]: return { "intent_match": self.intent_match, "time_match": self.time_match, "location_match": self.location_match, "interest_match": self.interest_match, "department_match": self.department_match, "friend_match": self.friend_match, "capacity_suitability": self.capacity_suitability, } @dataclass(frozen=True) class PairScore: student_a: str student_b: str matching_score: float factors: FactorScores common_minutes: int beacon_id: str | None @property def pair_key(self) -> tuple[str, str]: return tuple(sorted((self.student_a, self.student_b))) # type: ignore[return-value] @dataclass(frozen=True) class Exclusion: student_id: str reason_code: str details: dict = field(default_factory=dict) @dataclass(frozen=True) class MatchedGroup: student_ids: tuple[str, ...] matching_score: float selection_score: float beacon_id: str | None common_minutes: int factors: FactorScores @dataclass(frozen=True) class MatchingResult: groups: tuple[MatchedGroup, ...] exclusions: tuple[Exclusion, ...] pair_scores: tuple[PairScore, ...] eligible_student_ids: tuple[str, ...] @dataclass(frozen=True) class Scenario: name: str description: str now: datetime candidates: tuple[Candidate, ...] beacons: tuple[Beacon, ...] profiles: dict[str, Profile] duplicate_student_sets: frozenset[frozenset[str]] = frozenset() # --------------------------------------------------------------------------- # 3. FIXTURES / SCENARIOS # --------------------------------------------------------------------------- UTC = timezone.utc def _dt(hour: int, minute: int = 0) -> datetime: """Fixed demo day 2026-08-21 in UTC (keeps scenarios deterministic).""" return datetime(2026, 8, 21, hour, minute, tzinfo=UTC) def _iv(start_h: int, start_m: int, end_h: int, end_m: int) -> TimeInterval: return TimeInterval(start=_dt(start_h, start_m), end=_dt(end_h, end_m)) BEACONS_BASE: tuple[Beacon, ...] = ( Beacon( id="beacon_library", capacity=10, max_parallel_events=2, building_id="bldg_lib", campus_id="campus_main", ), Beacon( id="beacon_cafe", capacity=8, max_parallel_events=1, building_id="bldg_cafe", campus_id="campus_main", ), Beacon( id="beacon_gym", capacity=20, max_parallel_events=2, building_id="bldg_gym", campus_id="campus_east", ), ) def _profile( student_id: str, department_id: str | None = "dept_cs", friends: Iterable[str] = (), ) -> Profile: return Profile( student_id=student_id, department_id=department_id, friend_ids=frozenset(friends), ) def _candidate( student_id: str, intents: Sequence[str], intervals: Sequence[TimeInterval], location: str | None, expires_at: datetime | None = None, ) -> Candidate: return Candidate( student_id=student_id, intents=tuple(intents), intervals=tuple(intervals), expires_at=expires_at or _dt(23, 59), current_location_id=location, ) def build_scenarios() -> dict[str, Scenario]: now = _dt(13, 0) # 13:00 UTC demo "now" afternoon = (_iv(13, 0, 16, 0),) short_window = (_iv(13, 10, 13, 35),) # after +15 prep → too short # --- A: three aligned students → one group --- a_cands = ( _candidate("u001", ["study"], afternoon, "beacon_library"), _candidate("u002", ["study", "make_friends"], afternoon, "beacon_library"), _candidate("u003", ["study"], afternoon, "beacon_cafe"), ) a_profiles = { "u001": _profile("u001", friends=["u002"]), "u002": _profile("u002", friends=["u001"]), "u003": _profile("u003", department_id="dept_ee"), } # --- B: availability too short after prep → exclusions --- b_cands = ( _candidate("u001", ["study"], short_window, "beacon_library"), _candidate("u002", ["study"], short_window, "beacon_library"), ) b_profiles = {c.student_id: _profile(c.student_id) for c in b_cands} # --- C: six students → greedily two groups + leftover --- c_cands = ( _candidate("u001", ["study"], afternoon, "beacon_library"), _candidate("u002", ["study"], afternoon, "beacon_library"), _candidate("u003", ["study"], afternoon, "beacon_library"), _candidate("u004", ["lunch"], afternoon, "beacon_cafe"), _candidate("u005", ["lunch"], afternoon, "beacon_cafe"), _candidate("u006", ["wellness"], afternoon, "beacon_gym"), ) c_profiles = {c.student_id: _profile(c.student_id) for c in c_cands} # --- D: duplicate ACTIVE sets block every viable combo of the trio --- d_cands = a_cands d_profiles = a_profiles d_dup = frozenset( { frozenset({"u001", "u002"}), frozenset({"u001", "u003"}), frozenset({"u002", "u003"}), frozenset({"u001", "u002", "u003"}), } ) # --- E: friend / department weight visibility --- e_cands = ( _candidate("u001", ["social"], afternoon, "beacon_library"), _candidate("u002", ["social"], afternoon, "beacon_library"), _candidate("u003", ["social"], afternoon, "beacon_library"), ) e_profiles = { "u001": _profile("u001", "dept_cs", friends=["u002"]), "u002": _profile("u002", "dept_cs", friends=["u001"]), "u003": _profile("u003", "dept_ee"), } return { "A": Scenario( name="A", description="3 students align on study/time/place → 1 group", now=now, candidates=a_cands, beacons=BEACONS_BASE, profiles=a_profiles, ), "B": Scenario( name="B", description="Remaining time after prep < 30m → gate exclusions", now=now, candidates=b_cands, beacons=BEACONS_BASE, profiles=b_profiles, ), "C": Scenario( name="C", description="6 students → multiple groups via greedy non-reuse", now=now, candidates=c_cands, beacons=BEACONS_BASE, profiles=c_profiles, ), "D": Scenario( name="D", description="Same trio as A; duplicate sets block all pair/trio groups", now=now, candidates=d_cands, beacons=BEACONS_BASE, profiles=d_profiles, duplicate_student_sets=d_dup, ), "E": Scenario( name="E", description="Friend/department factors differ across pairs", now=now, candidates=e_cands, beacons=BEACONS_BASE, profiles=e_profiles, ), } SCENARIOS = build_scenarios() # --------------------------------------------------------------------------- # 4. TIME UTILS # --------------------------------------------------------------------------- def merge_intervals(intervals: Sequence[TimeInterval]) -> list[TimeInterval]: if not intervals: return [] ordered = sorted(intervals, key=lambda i: (i.start, i.end)) merged = [ordered[0]] for item in ordered[1:]: last = merged[-1] if item.start <= last.end: merged[-1] = TimeInterval(start=last.start, end=max(last.end, item.end)) else: merged.append(item) return merged def intersect_intervals( left: Sequence[TimeInterval], right: Sequence[TimeInterval], ) -> list[TimeInterval]: overlaps: list[TimeInterval] = [] for a in left: for b in right: start = max(a.start, b.start) end = min(a.end, b.end) if end > start: overlaps.append(TimeInterval(start=start, end=end)) return merge_intervals(overlaps) def clip_intervals_after( intervals: Sequence[TimeInterval], earliest: datetime, ) -> list[TimeInterval]: clipped: list[TimeInterval] = [] for interval in intervals: start = max(interval.start, earliest) if interval.end > start: clipped.append(TimeInterval(start=start, end=interval.end)) return clipped def total_minutes(intervals: Sequence[TimeInterval]) -> int: total = timedelta(0) for interval in intervals: total += interval.end - interval.start return int(total.total_seconds() // 60) def common_minutes_after_prep( left: Sequence[TimeInterval], right: Sequence[TimeInterval], *, now: datetime, ) -> int: earliest = now.astimezone(UTC) + timedelta(minutes=PREP_MINUTES) return total_minutes(clip_intervals_after(intersect_intervals(left, right), earliest)) def remaining_minutes_after_prep( intervals: Sequence[TimeInterval], *, now: datetime, ) -> int: earliest = now.astimezone(UTC) + timedelta(minutes=PREP_MINUTES) return total_minutes(clip_intervals_after(intervals, earliest)) # --------------------------------------------------------------------------- # 5. SCORING # --------------------------------------------------------------------------- def overlap_coefficient(left: Iterable[str], right: Iterable[str]) -> float: a = {x for x in left if x} b = {x for x in right if x} if not a or not b: return 0.0 return len(a & b) / min(len(a), len(b)) def time_match_score(common_minutes: int) -> float: if common_minutes < 30: return 0.0 if common_minutes < 60: return 0.50 if common_minutes < 90: return 0.75 return 1.00 def location_match_score( location_a: str | None, location_b: str | None, beacons: dict[str, Beacon], ) -> float: if not location_a or not location_b: return 0.50 if location_a == location_b: return 1.00 ba = beacons.get(location_a) bb = beacons.get(location_b) if ba is None or bb is None: return 0.50 if ba.building_id and bb.building_id and ba.building_id == bb.building_id: return 0.70 if ba.campus_id and bb.campus_id and ba.campus_id == bb.campus_id: return 0.30 if ba.campus_id and bb.campus_id and ba.campus_id != bb.campus_id: return 0.00 return 0.50 def department_match_score(dept_a: str | None, dept_b: str | None) -> float: if not dept_a or not dept_b: return 0.50 return 1.00 if dept_a == dept_b else 0.00 def friend_match_score(profile_a: Profile, profile_b: Profile) -> float: if profile_b.student_id in profile_a.friend_ids or profile_a.student_id in profile_b.friend_ids: return 1.00 return 0.00 def capacity_suitability_score(capacity: int) -> float | None: if capacity < MIN_BEACON_CAPACITY: return None if capacity <= 10: return 1.00 if capacity <= 15: return 0.70 return 0.40 def interest_match_score( interests_a: frozenset[str] | None, interests_b: frozenset[str] | None, ) -> float: # Provisional production policy: unknown → fixed 0.50 if interests_a is None or interests_b is None: return 0.50 return overlap_coefficient(interests_a, interests_b) def group_size_bonus(size: int) -> float: if size < MIN_GROUP_SIZE: return 0.0 capped = min(size, MAX_GROUP_SIZE) return round((capped - MIN_GROUP_SIZE) * GROUP_SIZE_BONUS_PER_EXTRA, 4) def is_beacon_suitable(beacon: Beacon, *, group_size: int) -> tuple[bool, str | None]: if not beacon.supports_micro: return False, REASON_NO_SUITABLE_BEACON if beacon.capacity < MIN_BEACON_CAPACITY or beacon.capacity < group_size: return False, REASON_NO_SUITABLE_BEACON if capacity_suitability_score(beacon.capacity) is None: return False, REASON_NO_SUITABLE_BEACON if beacon.active_parallel_count >= beacon.max_parallel_events: return False, REASON_BEACON_PARALLEL_FULL return True, None def select_beacon_for_members( members: Sequence[Candidate], beacons: Sequence[Beacon], ) -> Beacon | None: """Place score = proximity 0.70 + capacity 0.30.""" group_size = len(members) best: Beacon | None = None best_score = -1.0 for beacon in beacons: ok, _ = is_beacon_suitable(beacon, group_size=group_size) if not ok: continue cap = capacity_suitability_score(beacon.capacity) if cap is None: continue proximity_scores: list[float] = [] for member in members: loc = member.current_location_id if not loc: proximity_scores.append(0.50) elif loc == beacon.id: proximity_scores.append(1.00) else: other = next((b for b in beacons if b.id == loc), None) if other is None: proximity_scores.append(0.50) elif other.building_id and other.building_id == beacon.building_id: proximity_scores.append(0.70) elif other.campus_id and other.campus_id == beacon.campus_id: proximity_scores.append(0.30) else: proximity_scores.append(0.00) proximity = sum(proximity_scores) / len(proximity_scores) place_score = proximity * 0.70 + cap * 0.30 if place_score > best_score or ( place_score == best_score and best is not None and beacon.id < best.id ) or (place_score == best_score and best is None): best_score = place_score best = beacon return best def mean_factors(pairs: Sequence[FactorScores]) -> FactorScores: n = len(pairs) return FactorScores( intent_match=sum(p.intent_match for p in pairs) / n, time_match=sum(p.time_match for p in pairs) / n, location_match=sum(p.location_match for p in pairs) / n, interest_match=sum(p.interest_match for p in pairs) / n, department_match=sum(p.department_match for p in pairs) / n, friend_match=sum(p.friend_match for p in pairs) / n, capacity_suitability=sum(p.capacity_suitability for p in pairs) / n, ) # --------------------------------------------------------------------------- # 6. PIPELINE # --------------------------------------------------------------------------- def filter_eligible( candidates: Sequence[Candidate], *, now: datetime, beacons: Sequence[Beacon], duplicate_student_sets: frozenset[frozenset[str]], ) -> tuple[list[Candidate], list[Exclusion]]: now_utc = now.astimezone(UTC) exclusions: list[Exclusion] = [] eligible: list[Candidate] = [] suitable_exists = any( is_beacon_suitable(b, group_size=MIN_GROUP_SIZE)[0] for b in beacons ) blocked = duplicate_student_sets for candidate in sorted(candidates, key=lambda c: c.student_id): if candidate.expires_at <= now_utc: exclusions.append(Exclusion(candidate.student_id, REASON_INACTIVE_STATUS)) continue if not candidate.intervals: exclusions.append(Exclusion(candidate.student_id, REASON_NO_AVAILABILITY)) continue remaining = remaining_minutes_after_prep(candidate.intervals, now=now_utc) if remaining < MIN_EVENT_MINUTES: exclusions.append( Exclusion( candidate.student_id, REASON_INSUFFICIENT_REMAINING_TIME, {"remaining_minutes": remaining}, ) ) continue if not suitable_exists: exclusions.append(Exclusion(candidate.student_id, REASON_NO_SUITABLE_BEACON)) continue if frozenset({candidate.student_id}) in blocked: exclusions.append(Exclusion(candidate.student_id, REASON_DUPLICATE_CANDIDATE_EVENT)) continue eligible.append(candidate) return eligible, exclusions def score_pair( left: Candidate, right: Candidate, *, profile_a: Profile, profile_b: Profile, beacons: Sequence[Beacon], now: datetime, ) -> PairScore | None: common_minutes = common_minutes_after_prep(left.intervals, right.intervals, now=now) if common_minutes < MIN_EVENT_MINUTES: return None beacon = select_beacon_for_members((left, right), beacons) if beacon is None: return None cap = capacity_suitability_score(beacon.capacity) if cap is None: return None beacon_map = {b.id: b for b in beacons} factors = FactorScores( intent_match=overlap_coefficient(left.intents, right.intents), time_match=time_match_score(common_minutes), location_match=location_match_score( left.current_location_id, right.current_location_id, beacon_map, ), interest_match=interest_match_score(None, None), department_match=department_match_score(profile_a.department_id, profile_b.department_id), friend_match=friend_match_score(profile_a, profile_b), capacity_suitability=cap, ) return PairScore( student_a=left.student_id, student_b=right.student_id, matching_score=factors.weighted_total(), factors=factors, common_minutes=common_minutes, beacon_id=beacon.id, ) def form_groups( candidates: Sequence[Candidate], *, pair_scores: dict[tuple[str, str], PairScore], beacons: Sequence[Beacon], now: datetime, duplicate_student_sets: frozenset[frozenset[str]], ) -> list[MatchedGroup]: blocked = duplicate_student_sets by_id = {c.student_id: c for c in candidates} ids = sorted(by_id.keys()) proposals: list[MatchedGroup] = [] for size in range(MIN_GROUP_SIZE, MAX_GROUP_SIZE + 1): for combo in combinations(ids, size): member_ids = tuple(sorted(combo)) if frozenset(member_ids) in blocked: continue scored_pairs: list[PairScore] = [] ok = True for a, b in combinations(member_ids, 2): key = (a, b) if a < b else (b, a) pair = pair_scores.get(key) if pair is None or pair.matching_score < PAIR_SCORE_THRESHOLD: ok = False break scored_pairs.append(pair) if not ok: continue members = [by_id[sid] for sid in member_ids] common_intervals = list(members[0].intervals) for member in members[1:]: common_intervals = intersect_intervals(common_intervals, member.intervals) earliest = now.astimezone(UTC) + timedelta(minutes=PREP_MINUTES) common_minutes = total_minutes(clip_intervals_after(common_intervals, earliest)) if common_minutes < MIN_EVENT_MINUTES: continue beacon = select_beacon_for_members(members, beacons) if beacon is None: continue cap = capacity_suitability_score(beacon.capacity) if cap is None: continue averaged = mean_factors([p.factors for p in scored_pairs]) factors = FactorScores( intent_match=averaged.intent_match, time_match=time_match_score(common_minutes), location_match=averaged.location_match, interest_match=averaged.interest_match, department_match=averaged.department_match, friend_match=averaged.friend_match, capacity_suitability=cap, ) matching_score = factors.weighted_total() if matching_score < GROUP_SCORE_THRESHOLD: continue selection = round(matching_score + group_size_bonus(size), 4) proposals.append( MatchedGroup( student_ids=member_ids, matching_score=matching_score, selection_score=selection, beacon_id=beacon.id, common_minutes=common_minutes, factors=factors, ) ) proposals.sort(key=lambda g: (-g.selection_score, -g.matching_score, g.student_ids)) selected: list[MatchedGroup] = [] used: set[str] = set() for group in proposals: if any(sid in used for sid in group.student_ids): continue selected.append(group) used.update(group.student_ids) return selected def run_matching( candidates: Sequence[Candidate], *, profiles: dict[str, Profile], beacons: Sequence[Beacon], now: datetime, duplicate_student_sets: frozenset[frozenset[str]] | None = None, ) -> MatchingResult: """Public entry (same name as production matching_service.run_matching).""" blocked = duplicate_student_sets or frozenset() eligible, exclusions = filter_eligible( candidates, now=now, beacons=beacons, duplicate_student_sets=blocked, ) pair_score_map: dict[tuple[str, str], PairScore] = {} for left, right in combinations(eligible, 2): profile_a = profiles.get(left.student_id) or Profile(student_id=left.student_id) profile_b = profiles.get(right.student_id) or Profile(student_id=right.student_id) pair = score_pair( left, right, profile_a=profile_a, profile_b=profile_b, beacons=beacons, now=now, ) if pair is None: continue pair_score_map[pair.pair_key] = pair groups = form_groups( eligible, pair_scores=pair_score_map, beacons=beacons, now=now, duplicate_student_sets=blocked, ) return MatchingResult( groups=tuple(groups), exclusions=tuple(exclusions), pair_scores=tuple(sorted(pair_score_map.values(), key=lambda p: p.pair_key)), eligible_student_ids=tuple(sorted(c.student_id for c in eligible)), ) # --------------------------------------------------------------------------- # 7. REPORT (stdout — participants + result sections) # --------------------------------------------------------------------------- def _fmt_intervals(intervals: Sequence[TimeInterval]) -> str: if not intervals: return "(none)" return ", ".join( f"{item.start.strftime('%H:%M')}-{item.end.strftime('%H:%M')}" for item in intervals ) def _fmt_weights() -> str: return ( f"intent={WEIGHT_INTENT:.2f}, time={WEIGHT_TIME:.2f}, " f"location={WEIGHT_LOCATION:.2f}, interest={WEIGHT_INTEREST:.2f}, " f"department={WEIGHT_DEPARTMENT:.2f}, friend={WEIGHT_FRIEND:.2f}, " f"capacity={WEIGHT_CAPACITY:.2f}" ) def _candidate_by_id(scenario: Scenario) -> dict[str, Candidate]: return {c.student_id: c for c in scenario.candidates} def print_participants(scenario: Scenario, result: MatchingResult) -> None: """Show each participant's attributes used for gates and scoring.""" print("\n[1] Participants (availability / place / profile → scoring inputs)") print("-" * 72) print(f" score weights: {_fmt_weights()}") print(f" prep={PREP_MINUTES}m min_event={MIN_EVENT_MINUTES}m") if scenario.duplicate_student_sets: sets = [ ",".join(sorted(s)) for s in sorted(scenario.duplicate_student_sets, key=lambda s: sorted(s)) ] print(f" duplicate_student_sets: {{{'; '.join('{' + s + '}' for s in sets)}}}") print() exclusion_by_id = {item.student_id: item for item in result.exclusions} eligible = set(result.eligible_student_ids) beacons = {b.id: b for b in scenario.beacons} for candidate in sorted(scenario.candidates, key=lambda c: c.student_id): profile = scenario.profiles.get(candidate.student_id) or Profile( student_id=candidate.student_id ) remaining = remaining_minutes_after_prep( candidate.intervals, now=scenario.now, ) loc = candidate.current_location_id or "(none)" beacon = beacons.get(candidate.current_location_id or "") loc_detail = loc if beacon is not None: loc_detail = ( f"{beacon.id} (building={beacon.building_id}, " f"campus={beacon.campus_id}, capacity={beacon.capacity})" ) if candidate.student_id in eligible: status = "ELIGIBLE" else: ex = exclusion_by_id.get(candidate.student_id) status = f"EXCLUDED ({ex.reason_code})" if ex else "EXCLUDED" friends = ", ".join(sorted(profile.friend_ids)) if profile.friend_ids else "(none)" intents = ", ".join(candidate.intents) if candidate.intents else "(none)" print(f" {candidate.student_id} [{status}]") print(f" intents : {intents}") print( f" availability: {_fmt_intervals(candidate.intervals)} " f"(remaining after prep: {remaining}m)" ) print(f" location : {loc_detail}") print(f" expires_at : {candidate.expires_at.isoformat()}") print(f" department : {profile.department_id or '(none)'}") print(f" friends : {friends}") print(" interest : (unknown → interest_match fixed 0.50)") print() def print_pair_inputs(scenario: Scenario, pair: PairScore) -> None: """Show the raw attribute comparison behind each factor.""" by_id = _candidate_by_id(scenario) left = by_id[pair.student_a] right = by_id[pair.student_b] pa = scenario.profiles.get(left.student_id) or Profile(student_id=left.student_id) pb = scenario.profiles.get(right.student_id) or Profile(student_id=right.student_id) beacons = {b.id: b for b in scenario.beacons} print(" inputs:") print( f" intents {left.student_id}={list(left.intents)} " f"/ {right.student_id}={list(right.intents)}" ) print( f" time {left.student_id}={_fmt_intervals(left.intervals)} " f"/ {right.student_id}={_fmt_intervals(right.intervals)} " f"→ common_after_prep={pair.common_minutes}m" ) print( f" location {left.student_id}={left.current_location_id} " f"/ {right.student_id}={right.current_location_id} " f"→ chosen_beacon={pair.beacon_id}" ) print( f" department {left.student_id}={pa.department_id} " f"/ {right.student_id}={pb.department_id}" ) print( f" friends {left.student_id}→{sorted(pa.friend_ids) or []} " f"/ {right.student_id}→{sorted(pb.friend_ids) or []}" ) if pair.beacon_id and pair.beacon_id in beacons: beacon = beacons[pair.beacon_id] print( f" capacity beacon={beacon.id} capacity={beacon.capacity} " f"→ suitability={pair.factors.capacity_suitability:.2f}" ) print(" interest both unknown → 0.50") def print_report(scenario: Scenario, result: MatchingResult) -> None: print("=" * 72) print(f"Scenario {scenario.name}: {scenario.description}") print(f"now = {scenario.now.isoformat()}") print( f"candidates = {len(scenario.candidates)} " f"eligible = {len(result.eligible_student_ids)}" ) print("=" * 72) print_participants(scenario, result) print("[2] Exclusions") print("-" * 72) if not result.exclusions: print(" (none)") else: for item in result.exclusions: extra = f" {item.details}" if item.details else "" print(f" {item.student_id:8} {item.reason_code}{extra}") print("\n[3] Pair scores (threshold >= {:.2f})".format(PAIR_SCORE_THRESHOLD)) print("-" * 72) if not result.pair_scores: print(" (none)") else: for pair in result.pair_scores: mark = "OK" if pair.matching_score >= PAIR_SCORE_THRESHOLD else " " print( f" [{mark}] {pair.student_a}-{pair.student_b} " f"score={pair.matching_score:.4f} " f"common={pair.common_minutes}m beacon={pair.beacon_id}" ) factors = pair.factors.as_dict() parts = ", ".join(f"{k}={v:.2f}" for k, v in factors.items()) print(f" factors: {parts}") contrib = ( f"intent={pair.factors.intent_match * WEIGHT_INTENT:.3f}, " f"time={pair.factors.time_match * WEIGHT_TIME:.3f}, " f"location={pair.factors.location_match * WEIGHT_LOCATION:.3f}, " f"interest={pair.factors.interest_match * WEIGHT_INTEREST:.3f}, " f"department={pair.factors.department_match * WEIGHT_DEPARTMENT:.3f}, " f"friend={pair.factors.friend_match * WEIGHT_FRIEND:.3f}, " f"capacity={pair.factors.capacity_suitability * WEIGHT_CAPACITY:.3f}" ) print(f" weighted: {contrib}") print_pair_inputs(scenario, pair) print() print( "[4] Groups (matching_score >= {:.2f}, greedy non-reuse)".format( GROUP_SCORE_THRESHOLD ) ) print("-" * 72) if not result.groups: print(" (none)") else: for idx, group in enumerate(result.groups, start=1): members = ", ".join(group.student_ids) print( f" G{idx}: [{members}] " f"matching={group.matching_score:.4f} " f"selection={group.selection_score:.4f} " f"beacon={group.beacon_id} common={group.common_minutes}m" ) factors = group.factors.as_dict() parts = ", ".join(f"{k}={v:.2f}" for k, v in factors.items()) print(f" factors: {parts}") print() # --------------------------------------------------------------------------- # 8. MAIN # --------------------------------------------------------------------------- def main(argv: Sequence[str] | None = None) -> int: parser = argparse.ArgumentParser( description="HALO Student 2 matching demo (single-file, stdlib only)", ) parser.add_argument( "--scenario", choices=sorted(SCENARIOS.keys()), help="Scenario id to run (A–E)", ) parser.add_argument( "--list", action="store_true", help="List embedded scenarios and exit", ) args = parser.parse_args(argv) if args.list or args.scenario is None: print("Available scenarios:") for key in sorted(SCENARIOS): sc = SCENARIOS[key] print(f" {key}: {sc.description}") if args.scenario is None and not args.list: print("\nRun with: python matching_demo.py --scenario A") return 0 scenario = SCENARIOS[args.scenario] result = run_matching( scenario.candidates, profiles=scenario.profiles, beacons=scenario.beacons, now=scenario.now, duplicate_student_sets=scenario.duplicate_student_sets, ) print_report(scenario, result) return 0 if __name__ == "__main__": raise SystemExit(main())