194 lines
8.3 KiB
Python
194 lines
8.3 KiB
Python
# demo.py
|
||
# ------------------------------------------------------------
|
||
# Demonstration script that creates a few services, loads the
|
||
# classifications, and then schedules members for each position
|
||
# using the new SchedulingService.
|
||
# ------------------------------------------------------------
|
||
|
||
from __future__ import annotations
|
||
|
||
from pathlib import Path
|
||
from datetime import date, timedelta
|
||
from typing import Dict, List, Tuple, Any
|
||
|
||
# Import the concrete repository classes that talk to SQLite (or any DB you use)
|
||
from backend.repositories import (
|
||
ClassificationRepository,
|
||
MemberRepository,
|
||
ServiceRepository,
|
||
ServiceAvailabilityRepository,
|
||
ScheduleRepository
|
||
)
|
||
|
||
# The service we just wrote
|
||
from backend.services.scheduling_service import SchedulingService
|
||
|
||
|
||
# ----------------------------------------------------------------------
|
||
# Helper – return the next *n* Sundays starting from today.
|
||
# ----------------------------------------------------------------------
|
||
def next_n_sundays(n: int) -> List[date]:
|
||
"""Return a list of the next *n* Sundays (including today if today is Sunday)."""
|
||
today = date.today()
|
||
# weekday(): Monday == 0 … Sunday == 6
|
||
days_until_sunday = (6 - today.weekday()) % 7
|
||
first_sunday = today + timedelta(days=days_until_sunday)
|
||
return [first_sunday + timedelta(weeks=i) for i in range(n)]
|
||
|
||
|
||
# ----------------------------------------------------------------------
|
||
# Demo entry point (updated for multi‑classification support)
|
||
# ----------------------------------------------------------------------
|
||
def demo(
|
||
classification_repo: ClassificationRepository,
|
||
member_repo: MemberRepository,
|
||
service_repo: ServiceRepository,
|
||
availability_repo: ServiceAvailabilityRepository,
|
||
schedule_repo: ScheduleRepository,
|
||
) -> None:
|
||
"""
|
||
Populate a handful of services for the coming Sunday and run the
|
||
round‑robin scheduler for each choir‑position.
|
||
|
||
The function prints what it does so you can see the flow in the console.
|
||
"""
|
||
# ------------------------------------------------------------------
|
||
# 0️⃣ Define the members we want to skip.
|
||
# ------------------------------------------------------------------
|
||
EXCLUDED_MEMBER_IDS = {20, 8, 3, 12, 4, 1, 44, 46, 28, 13, 11, 5, 16, 26, 35}
|
||
|
||
# ------------------------------------------------------------------
|
||
# 1️⃣ Build the high‑level SchedulingService from the repos.
|
||
# ------------------------------------------------------------------
|
||
scheduler = SchedulingService(
|
||
classification_repo=classification_repo,
|
||
member_repo=member_repo,
|
||
service_repo=service_repo,
|
||
availability_repo=availability_repo,
|
||
schedule_repo=schedule_repo,
|
||
)
|
||
|
||
# ------------------------------------------------------------------
|
||
# 2️⃣ Create a single Sunday of services (9 AM, 11 AM, 6 PM).
|
||
# ------------------------------------------------------------------
|
||
# We only need one Sunday for the demo – the second element of the
|
||
# list returned by ``next_n_sundays(6)`` matches the original code.
|
||
target_sunday = next_n_sundays(6)[4] # same as original slice [1:2]
|
||
print(f"🗓️ Target Sunday: {target_sunday}")
|
||
input()
|
||
|
||
# Create the three service slots for that day.
|
||
service_ids_by_type: Dict[int, int] = {} # ServiceTypeId → ServiceId
|
||
for service_type_id in (1, 2, 3):
|
||
service = service_repo.create(service_type_id, target_sunday)
|
||
service_ids_by_type[service_type_id] = service.ServiceId
|
||
type_name = {1: "9AM", 2: "11AM", 3: "6PM"}[service_type_id]
|
||
print(
|
||
f"✅ Created Service → ServiceId={service.ServiceId}, "
|
||
f"ServiceType={type_name}, Date={service.ServiceDate}"
|
||
)
|
||
input()
|
||
|
||
# ------------------------------------------------------------------
|
||
# 3️⃣ Load the classification IDs we’ll need later.
|
||
# ------------------------------------------------------------------
|
||
classifications = classification_repo.list_all()
|
||
def _cid(name: str) -> int:
|
||
return next(c.ClassificationId for c in classifications if c.ClassificationName == name)
|
||
|
||
baritone_id = _cid("Baritone")
|
||
tenor_id = _cid("Tenor")
|
||
mezzo_alto_id = _cid("Alto / Mezzo")
|
||
soprano_id = _cid("Soprano")
|
||
|
||
# ------------------------------------------------------------------
|
||
# 4️⃣ Define the choir‑positions and which classifications are acceptable.
|
||
# ------------------------------------------------------------------
|
||
# The mapping mirrors the comment block in the original script.
|
||
positions_to_classifications: Dict[int, List[int]] = {
|
||
1: [baritone_id, tenor_id], # 1 Baritone or Tenor
|
||
2: [tenor_id], # 2 Tenor
|
||
3: [tenor_id, mezzo_alto_id], # 3 Tenor
|
||
4: [mezzo_alto_id], # 4 Mezzo
|
||
5: [mezzo_alto_id, soprano_id], # 5 Mezzo or Soprano
|
||
6: [mezzo_alto_id, soprano_id], # 6 Mezzo or Soprano
|
||
7: [soprano_id], # 7 Soprano
|
||
8: [soprano_id], # 8 Soprano
|
||
}
|
||
|
||
# ------------------------------------------------------------------
|
||
# 5️⃣ Run the scheduler for each position on each service slot.
|
||
# ------------------------------------------------------------------
|
||
# We keep a dict so the final printout resembles the original script.
|
||
full_schedule: Dict[int, List[Tuple[int, str, str, int]]] = {}
|
||
|
||
for service_type_id, service_id in service_ids_by_type.items():
|
||
service_type_name = {1: "9AM", 2: "11AM", 3: "6PM"}[service_type_id]
|
||
print(f"\n=== Sunday {target_sunday} @ {service_type_name} ===")
|
||
full_schedule[service_id] = []
|
||
input()
|
||
|
||
for position, allowed_cids in positions_to_classifications.items():
|
||
# --------------------------------------------------------------
|
||
# New round‑robin path: give the whole list of allowed
|
||
# classifications to the scheduler at once.
|
||
# --------------------------------------------------------------
|
||
result = scheduler.schedule_next_member(
|
||
classification_ids=allowed_cids,
|
||
service_id=service_id,
|
||
only_active=True,
|
||
exclude_member_ids=EXCLUDED_MEMBER_IDS,
|
||
)
|
||
|
||
# --------------------------------------------------------------
|
||
# Store the outcome – either a valid schedule tuple or a placeholder.
|
||
# --------------------------------------------------------------
|
||
if result:
|
||
full_schedule[service_id].append(result)
|
||
print(f"#{position}: {result[1]} {result[2]}")
|
||
input()
|
||
else:
|
||
placeholder = (None, "❓", "No eligible member", None)
|
||
full_schedule[service_id].append(placeholder)
|
||
print(f"#{position}: ❓ No eligible member")
|
||
input()
|
||
|
||
# ------------------------------------------------------------------
|
||
# 6️⃣ Final dump – mirrors the original ``print(schedule)``.
|
||
# ------------------------------------------------------------------
|
||
print("\n🗂️ Complete schedule dictionary:")
|
||
print(full_schedule)
|
||
|
||
|
||
# ----------------------------------------------------------------------
|
||
# Example of wiring everything together (you would normally do this in
|
||
# your application start‑up code).
|
||
# ----------------------------------------------------------------------
|
||
if __name__ == "__main__":
|
||
from backend.db import DatabaseConnection
|
||
from backend.repositories import MemberRepository, ScheduleRepository, ServiceRepository, ServiceAvailabilityRepository
|
||
from backend.services.scheduling_service import SchedulingService
|
||
|
||
DB_PATH = Path(__file__).parent / "db" / "sqlite" / "database.db"
|
||
|
||
# Initialise DB connection (adjust DSN as needed)
|
||
db = DatabaseConnection(DB_PATH)
|
||
|
||
# Instantiate each repository with the shared DB connection.
|
||
classification_repo = ClassificationRepository(db)
|
||
member_repo = MemberRepository(db)
|
||
service_repo = ServiceRepository(db)
|
||
availability_repo = ServiceAvailabilityRepository(db)
|
||
schedule_repo = ScheduleRepository(db)
|
||
|
||
# Run the demo.
|
||
demo(
|
||
classification_repo,
|
||
member_repo,
|
||
service_repo,
|
||
availability_repo,
|
||
schedule_repo,
|
||
)
|
||
|
||
|