[요구사항 설명]
•
모든 부원들은 각자의 팀이 있다.
•
합주실은 각각 이용가능한 시간이 있다.
•
현 알고리즘은 각 팀의 합주 일정을 자동으로 정해주는 알고리즘이며, 각 팀마다 합주는 월요일을 한 주의 시작으로 했을때 그 주의 하루만 배정해주면 된다. 그렇게 매주마다 매칭되는 시간이 있는 각 팀들이 최소 한번 합주를 진행해야 한다.
1.
모든 부원들의 시간을 받는다
2.
각 팀의 가능 시간을 파악한다 → 매칭되는 시간이 아예 없는 팀은 따로 뺀다.
3.
합주실의 이용 시간을 파악한다 → 합주 가능 시간을 파악한다.(여러 합주실중 가장 이른 시간 ~ 가장 빠른 시간)
4.
합주 가능 시간에, 각 팀이 가능한 시간대를 하나씩 배정한다.
휴리스틱한 부분
•
아무리 합주실 이용 가능 시간이더라도, 너무 이른 시간과 늦은 시간은 되도록 피한다.
•
최대한 모든 팀이 연속된 시간이 되도록 한다.
◦
팀 두개가 연속된 시간으로 잡을 수 있는 경우, 30분씩 할당한다. 이유는 합주실은 한시간 단위로만 예약을 할 수 있는데, 두 팀을 한시간에 넣으면 돈을 절약할 수 있기 때문이다.
◦
팀 두개를 연속된 시간으로 잡을 수 없는 경우 1시간을 할당한다.
[알고리즘 구현 설명]
step 1. 합주팀별 멤버들의 되는시간 파악하기.
1.
performance_id를 받아, 해당 공연의 ‘합주팀’ 엔티티들을 ‘합주팀들’로 받는다.
2.
‘합주팀들’의 멤버 목록을 ‘멤버’ 엔티티들인 ‘멤버들’로 받는다.
3.
‘멤버들’로 ‘멤버되는시간슬롯’객체들을 만들고(멤버되는시간슬롯들에 저장), 이를 ‘합주팀들’로 ‘팀되는시간슬롯’객체들을 만든다.(팀되는시간슬롯들에 저장)
step 2. 날짜별 합주가능시간들 파악하기
1.
performance_id를 받아, 해당 공연의 동아리의 ‘합주실’ 엔티티들을 ‘합주실들’로 받는다.
2.
합주실들의 되는 시간 중, 합주실을 통틀어 날짜별 가장 이른 시간과, 가장 늦은 시간을 저장한다. 합주실이 적어도 한개는 가능한 시간대에 합주를 배정해주기 위해서다.
3.
오전 11시부터 오후 11시까지의 시간대를 위에서 구한 시간과 교집합한다. 너무 이른시간이나 너무 늦은 시간에 합주를 배정해주지 않기 위해서다.
4.
각 날짜별 합주 가능 시간(가장 이른 시간 ~ 가장 늦은 시간)을 파악하여 저장한다. (합주가능시간들)
step 3. 매주 팀별 1회 합주를 배정해준다.
1.
각 주마다 배정을 수행해준다. 이때 주의 시작은 월요일로 본다.
a.
각 주마다 매칭되는 시간이 있는 팀, 매칭되는 시간이 아예 없는 팀을 먼저 분류한다. (’합주가능팀들’에는 매칭되는 시간이 있는 팀들의 팀되는시간슬롯 객체가 들어가고, ‘합주불가능팀들’에는 매칭되는 시간이 없는 팀들의 팀되는시간슬롯 객체가 들어간다.)
b.
합주가능팀들로 요구사항에 맞는 기존 코드의 알고리즘을 수행한다.
여기서 각 날짜별 탐색 시간 범위는 당연히 step2에서 저장한 날짜별 합주가능시간들로 한다.
•
주간 캐싱으로 성능 최적화
•
2팀 페어 최대 조합 찾기 (캐시 사용)
•
최대 페어 조합 계산
•
최적 조합으로 배정 생성
현재 알고리즘 코드
_find_weekly_allocations_fast
•
_allocate_single_week
◦
_find_week_continuous_blocks
◦
_greedy_fill_continuous_block
◦
_complete_week_allocation
◦
_fallback_week_allocation
현재 11~23시 하드코딩 + 합주실 교집합으로 자동배정을 해주고 있음
합주실이 24시간 운영하더라도, 새벽으로 자동배정되는건 막기위해..
코드 요약
시간구간: 15분 단위로 내림처리되어있다. 시작시간(datetype), 종료시간(datetype)으로 이루어져있다.
•
시작시간
•
종료시간
되는시간Type
•
가능시간: list[시간구간]
•
불가능시간: list[시간구간]
•
가능하지만비선호시간: list[시간구간]
되는시간Type(
가능시간=[시간구간(시작시간=datetime(2024, 1, 15, 19, 0), 종료시간=datetime(2024, 1, 15, 21, 0)), # 월 19:00-21:00
시간구간(시작시간=datetime(2024, 1, 16, 18, 0), 종료시간=datetime(2024, 1, 16, 20, 0)), # 화 18:00-20:00],
불가능시간=[시간구간(시작시간=datetime(2024, 1, 17, 19, 0), 종료시간=datetime(2024, 1, 17, 21, 0)), # 수 19:00-21:00],
가능하지만비선호시간=[시간구간(시작시간=datetime(2024, 1, 18, 18, 30), 종료시간=datetime(2024, 1,18, 20, 30)), # 목 18:30-20:30]
)
Python
복사
가용상태
•
마스크
•
가능
•
불가능
•
비선호
되는시간슬롯
•
start_date
•
end_date
•
data: list[list[가용상태]]
해야할 것.
1.
team과 team member 작년거로 더미데이터 추가
2.
member들의 가용시간 더미데이터로 추가
현재 클럽에 합주실이 대응되고,
시간 할당에서 합주실을 찾을때는 공연 id로 찾아야한다.
클럽이 합주실에 대응되는게 맞을까?
이상적인건 클럽마다 후보지로 가능한 합주실들이 있고, 공연을 생성할 때마다 합주실을 선택해주어야할 것 같기도 하다… 아 여기가 너무 애매하다
API에서 돌려줘야할 것.
1.
가능한 팀, 불가능한 팀 목록
2.
가능한 팀들 각 팀의 가능한 시간대
1.
가능한 팀들의 자동 매칭 시간대
1.
되는시간Type (인간 친화적 표현)
김현서라는 멤버가 다음과 같이 가능한 시간을 말했다고 가정해봅시다:
김현서의 가용시간:
•
"월요일 오후 7시~9시는 가능해"
•
"화요일 오후 6시~8시도 가능해"
•
"수요일 오후 7시~9시는 불가능해"
•
"목요일 오후 6:30~8:30은 가능하긴 한데 별로 선호하지 않아"
이것을 되는시간Type으로 표현하면:
되는시간Type(
가능시간=[
시간구간(시작시간=datetime(2024, 1, 15, 19, 0), 종료시간=datetime(2024, 1,
15, 21, 0)), # 월 19:00-21:00
시간구간(시작시간=datetime(2024, 1, 16, 18, 0), 종료시간=datetime(2024, 1,
16, 20, 0)), # 화 18:00-20:00
],
불가능시간=[
시간구간(시작시간=datetime(2024, 1, 17, 19, 0), 종료시간=datetime(2024, 1,
17, 21, 0)), # 수 19:00-21:00
],
가능하지만비선호시간=[
시간구간(시작시간=datetime(2024, 1, 18, 18, 30), 종료시간=datetime(2024, 1,
18, 20, 30)), # 목 18:30-20:30
]
)
2. 되는시간슬롯 (컴퓨터 친화적 표현)같은 김현서의 가용시간을 2차원 배열로 표현하면:
되는시간슬롯(
start_date=date(2024, 1, 15), # 월요일
end_date=date(2024, 1, 18), # 목요일 (4일간)
data=[
# 1월 15일 (월요일) - 96개 슬롯 (00:00~23:45, 15분 단위)
[마스크, 마스크, ..., 가능, 가능, 가능, 가능, 가능, 가능, 가능, 가능,
마스크, ...],
# 0~18:45 19:00 19:15 19:30 19:45 20:00 20:15 20:30 20:45 21:00~
# 1월 16일 (화요일)
[마스크, 마스크, ..., 가능, 가능, 가능, 가능, 가능, 가능, 가능, 가능,
Plain Text
복사
마스크, ...],
# 0~17:45 18:00 18:15 18:30 18:45 19:00 19:15 19:30 19:45 20:00~
# 1월 17일 (수요일)
[마스크, 마스크, ..., 불가능, 불가능, 불가능, 불가능, 불가능, 불가능,
Plain Text
복사
불가능, 불가능, 마스크, ...],
# 0~18:45 19:00 19:15 19:30 19:45 20:00 20:15 20:30
20:45 21:00~
# 1월 18일 (목요일)
[마스크, 마스크, ..., 비선호, 비선호, 비선호, 비선호, 비선호, 비선호,
Plain Text
복사
비선호, 비선호, 비선호, 마스크, ...],
# 0~18:15 18:30 18:45 19:00 19:15 19:30 19:45 20:00 20:15
20:30 20:45~
]
)
슬롯 인덱스 계산법
•
1시간 = 4개 슬롯 (15분 × 4 = 60분)
•
하루 = 96개 슬롯 (24시간 × 4 = 96개)
예시:
•
19:00 → 슬롯 인덱스 76 (19 × 4 + 0)
•
19:15 → 슬롯 인덱스 77 (19 × 4 + 1)
•
20:45 → 슬롯 인덱스 83 (20 × 4 + 3)
Update Todos
⎿ ☒ Explain 되는시간Type with concrete examples☒ Explain 되는시간슬롯 with concrete examples
☐ Show conversion process between the two representations
3. 변환 과정 (from_되는시간 메서드)선택하신 코드가 바로 이 변환 과정입니다:
@classmethod
def from_되는시간(cls, 되는시간: 되는시간Type) -> Self:
# 1단계: 모든 시간구간을 수집해서 전체 날짜 범위 파악
모든_시간구간 = []
모든_시간구간.extend(되는시간.가능시간) # [월19-21, 화18-20]
모든_시간구간.extend(되는시간.불가능시간) # [수19-21]
모든_시간구간.extend(되는시간.가능하지만비선호시간) # [목18:30-20:30]
# 2단계: 날짜 범위 계산
start_date = 1월15일 # 가장 빠른 날짜
end_date = 1월18일 # 가장 늦은 날짜
# 3단계: 빈 슬롯 배열 생성 (4일 × 96슬롯, 모두 마스크로 초기화)
slots = [
[마스크] * 96, # 1월15일
[마스크] * 96, # 1월16일
[마스크] * 96, # 1월17일
[마스크] * 96 # 1월18일
]
# 4단계: 각 시간구간을 슬롯에 채워넣기
# 월19:00-21:00 → slots[0][76:84] = 가능
# 화18:00-20:00 → slots[1][72:80] = 가능
# 수19:00-21:00 → slots[2][76:84] = 불가능
# 목18:30-20:30 → slots[3][74:82] = 비선호
Plain Text
복사
왜 이런 이중 표현을 사용하나요?
1.
되는시간Type: 사람이 이해하기 쉬움
- "월요일 7시부터 9시까지 가능해"
- UI에서 입력받기 편함
2.
되는시간슬롯: 컴퓨터 계산에 최적화
- 팀 멤버들의 교집합 계산이 쉬움
- 두 사람의 가용시간을 비교할 때 슬롯 단위로 AND 연산
- 알고리즘에서 빠른 처리 가능
Update Todos
⎿ ☒ Explain 되는시간Type with concrete examples☒ Explain 되는시간슬롯 with concrete examples
☒ Show conversion process between the two representations
이렇게 두 가지 표현 방식을 활용해서 사용자 인터페이스에서는 직관적으로, 백엔드
계산에서는 효율적으로 시간 데이터를 처리할 수 있습니다."""단계별 시간 배정 알고리즘 테스트"""
from datetime import datetime, date, timedelta
from typing import List, Dict, Tuple, Optional
from collections import defaultdict
from hegemony.common.container import Container
from hegemony.domain.vo.availability import 시간구간, 되는시간Type, 되는시간슬롯, 가용상태
from hegemony.domain.domain_service.allocation_algorithm_service import (
팀되는시간슬롯,
멤버되는시간슬롯,
연습실되는시간슬롯,
팀시간배정결과
)
class StepByStepTester:
"""단계별 알고리즘 테스터"""
def __init__(self):
self.container = Container()
self.container.wire(modules=[__name__])
self.performance_repo = self.container.performance_repository()
self.team_repo = self.container.team_repository()
self.member_repo = self.container.member_repository()
self.rehearsal_room_repo = self.container.rehearsal_room_repository()
def run_full_test(self, performance_id: str):
"""전체 단계별 테스트 실행"""
print("=" * 60)
print("🎯 단계별 시간 배정 알고리즘 테스트")
print("=" * 60)
# Step 1: 팀별 사용시간 수집
print("\n📋 Step 1: 부원들의 가용시간 수집")
members_data = self.step1_collect_member_availability(performance_id)
print(members_data)
self.visualize_member_availability(members_data)
# Step 2: 각 팀의 가능 시간을 파악한다
print("\n👥 Step 2: 팀별 가용시간 계산 및 필터링")
teams_data = self.step2_calculate_team_availability(performance_id, members_data)
self.visualize_team_availability(teams_data)
# Step 3: 합주실의 이용 시간을 파악한다
print("\n🏠 Step 3: 합주실 가용시간 분석")
rooms_data = self.step3_analyze_rehearsal_rooms(performance_id)
self.visualize_room_availability(rooms_data)
# Step 4: 팀 슬롯 데이터 준비 (새로운 구조)
print("\n🎲 Step 4: 팀 슬롯 데이터 준비")
team_slot_data = self.step4_prepare_team_slot_data(teams_data)
# Step 5: 최대 연속 페어 배정 알고리즘 적용
print("\n⚡ Step 5: 최대 연속 페어 배정")
final_candidates = self._find_weekly_allocations_fast(team_slot_data, rooms_data)
self.visualize_final_candidates(final_candidates)
return final_candidates
def step1_collect_member_availability(self, performance_id: str) -> Dict:
"""Step 1: 부원들의 가용시간 수집"""
performance = self.performance_repo.find_by_performance_id(performance_id)
if not performance:
raise ValueError(f"공연을 찾을 수 없습니다: {performance_id}")
teams = self.team_repo.find_by_performance_id(performance_id)
members_data = {
'performance_id': performance_id,
'total_teams': len(teams),
'members': {},
'teams': {}
}
# N+1 문제 해결: 모든 멤버를 한 번에 조회
all_member_ids = []
for team in teams:
all_member_ids.extend(team.멤버_목록)
# 중복 제거 후 배치 조회 (1번의 쿼리로 모든 멤버 조회)
unique_member_ids = list(set(all_member_ids))
members_list = self.member_repo.find_by_ids(unique_member_ids)
members_dict = {member.id: member for member in members_list}
total_members = 0
members_with_availability = 0
for team in teams:
team_members = []
for member_id in team.멤버_목록:
member = members_dict.get(member_id) # ← 딕셔너리에서 O(1) 조회
if member:
total_members += 1
has_availability = member.되는시간 is not None
if has_availability:
members_with_availability += 1
available_hours = self._calculate_total_available_hours(member.되는시간)
else:
available_hours = 0
members_data['members'][member_id] = {
'name': member.이름,
'has_availability': has_availability,
'available_hours': available_hours,
'availability_data': member.되는시간
}
team_members.append(member_id)
members_data['teams'][team.id] = {
'team_id': team.id,
'members': team_members,
'member_count': len(team_members)
}
members_data['summary'] = {
'total_members': total_members,
'members_with_availability': members_with_availability,
'coverage_rate': members_with_availability / total_members if total_members > 0 else 0
}
return members_data
def step2_calculate_team_availability(self, performance_id: str, members_data: Dict) -> Dict:
"""Step 2: 팀별 가용시간 계산 및 필터링 (기존 도메인 객체 사용)"""
teams_data = {
'teams_with_availability': {},
'teams_without_availability': {},
'summary': {}
}
for team_id, team_info in members_data['teams'].items():
print(f" 📊 팀 {team_id} 분석 중...")
# 멤버되는시간슬롯 객체들 생성
멤버슬롯들 = []
missing_members = []
for member_id in team_info['members']:
member_data = members_data['members'][member_id]
if member_data['has_availability'] and member_data['availability_data']:
# 되는시간Type을 되는시간슬롯으로 변환
member_slots = 되는시간슬롯.from_되는시간(member_data['availability_data'])
멤버슬롯 = 멤버되는시간슬롯(member_id=member_id, slots=member_slots)
멤버슬롯들.append(멤버슬롯)
print(f" - 멤버 {member_id}: 가용시간 있음")
else:
missing_members.append(member_id)
print(f" - 멤버 {member_id}: 가용시간 없음")
if not 멤버슬롯들:
# 가용시간이 있는 멤버가 아예 없는 팀
teams_data['teams_without_availability'][team_id] = {
'team_id': team_id,
'reason': 'no_member_availability',
'missing_members': missing_members,
'total_members': len(team_info['members'])
}
print(f" ⚠️ 가용시간을 계산할 수 있는 멤버가 없습니다.")
else:
# 팀되는시간슬롯 객체 생성 (자동으로 멤버들의 교집합 계산)
team_slot = 팀되는시간슬롯(team_id=team_id, members=멤버슬롯들)
# 교집합 결과를 되는시간Type으로 변환하여 실제 가용시간 구간 확인
팀_가용시간 = team_slot.slots.to_되는시간()
가능시간_구간_수 = len(팀_가용시간.가능시간)
print(f" → 팀 교집합 결과: {가능시간_구간_수}개 가능시간 구간")
if missing_members:
print(f" → 가용시간 없는 멤버 {len(missing_members)}명: {missing_members}")
if 가능시간_구간_수 > 0: # 실제로 공통 가용시간이 있는지 확인
# 시간 단위로 변환하여 표시
team_available_hours = self._calculate_available_hours_from_slots(team_slot.slots)
teams_data['teams_with_availability'][team_id] = {
'team_id': team_id,
'members': team_info['members'],
'team_slot': team_slot,
'available_hours': team_available_hours,
'available_time_intervals': 가능시간_구간_수,
'member_count': len(team_info['members']),
'members_with_data': len(멤버슬롯들),
'members_without_data': len(missing_members)
}
print(f" ✅ {team_available_hours:.1f}시간 가용 ({가능시간_구간_수}개 구간)")
else:
print(f" ⚠️ 이 팀은 공통 가용시간이 없습니다!")
teams_data['teams_without_availability'][team_id] = {
'team_id': team_id,
'reason': 'no_common_time',
'individual_availabilities': len(멤버슬롯들),
'total_members': len(team_info['members']),
'members_with_data': len(멤버슬롯들),
'members_without_data': len(missing_members)
}
# 요약 통계
teams_data['summary'] = {
'total_teams': len(members_data['teams']),
'teams_with_availability': len(teams_data['teams_with_availability']),
'teams_without_availability': len(teams_data['teams_without_availability']),
'allocation_possible': len(teams_data['teams_with_availability']) > 0
}
return teams_data
def step3_analyze_rehearsal_rooms(self, performance_id: str) -> Dict:
"""Step 3: 합주실 가용시간 분석 (기존 도메인 객체 사용)"""
performance = self.performance_repo.find_by_performance_id(performance_id)
rehearsal_rooms = self.rehearsal_room_repo.find_by_club_id(performance.동아리_id)
rooms_data = {
'rooms': {},
'room_slots': [], # 연습실되는시간슬롯 객체들
'summary': {},
'global_availability_window': None
}
if not rehearsal_rooms:
rooms_data['summary'] = {
'total_rooms': 0,
'rooms_with_availability': 0,
'allocation_possible': False,
'reason': 'no_rehearsal_rooms'
}
return rooms_data
earliest_time = None
latest_time = None
rooms_with_availability = 0
for room in rehearsal_rooms:
room_id = room.id
print(f" 🏠 합주실 {room.합주실명} 분석 중...")
if room.되는시간:
# 연습실되는시간슬롯 객체 생성
room_slots = 되는시간슬롯.from_되는시간(room.되는시간)
room_slot_obj = 연습실되는시간슬롯(room_id=room_id, slots=room_slots)
rooms_data['room_slots'].append(room_slot_obj)
available_hours = self._calculate_available_hours_from_slots(room_slots)
rooms_with_availability += 1
# 전체 가용 시간대 계산
room_earliest, room_latest = self._get_availability_time_range(room.되는시간)
if earliest_time is None or (room_earliest and room_earliest < earliest_time):
earliest_time = room_earliest
if latest_time is None or (room_latest and room_latest > latest_time):
latest_time = room_latest
rooms_data['rooms'][room_id] = {
'room_id': room_id,
'name': room.합주실명,
'location': room.위치,
'room_slot': room_slot_obj,
'available_hours': available_hours,
'earliest_time': room_earliest,
'latest_time': room_latest
}
else:
rooms_data['rooms'][room_id] = {
'room_id': room_id,
'name': room.합주실명,
'location': room.위치,
'room_slot': None,
'available_hours': 0,
'reason': 'no_availability_data'
}
# 전체 합주 가능 시간대 계산
if earliest_time and latest_time:
rooms_data['global_availability_window'] = {
'earliest_time': earliest_time,
'latest_time': latest_time,
'total_hours': (latest_time - earliest_time).total_seconds() / 3600
}
rooms_data['summary'] = {
'total_rooms': len(rehearsal_rooms),
'rooms_with_availability': rooms_with_availability,
'allocation_possible': rooms_with_availability > 0,
'global_window': rooms_data['global_availability_window']
}
return rooms_data
def step4_prepare_team_slot_data(self, teams_data: Dict) -> Dict:
"""Step 4: 팀 슬롯 데이터 준비 (새로운 구조)"""
if not teams_data['summary']['allocation_possible']:
return {
'teams_with_availability': {},
'summary': {
'total_teams': 0,
'teams_with_availability': 0,
'reason': 'no_teams_available'
}
}
print(f" 📋 {len(teams_data['teams_with_availability'])}개 팀의 슬롯 데이터 준비 중...")
return {
'teams_with_availability': teams_data['teams_with_availability'],
'summary': {
'total_teams': len(teams_data['teams_with_availability']),
'teams_with_availability': len(teams_data['teams_with_availability'])
}
}
def _find_weekly_allocations_fast(self, 팀_슬롯_데이터: Dict, 합주실_데이터: Dict) -> List[Dict]:
"""주별 스케줄링 알고리즘 - 매주 팀별 1회 합주 배정"""
print("\n🚀 5단계: 주별 합주 배정 시작")
# 가용 팀들만 추출
가용_팀들 = [team_id for team_id, team_data in 팀_슬롯_데이터['teams_with_availability'].items()]
print(f"가용 팀 수: {len(가용_팀들)}개")
# 전체 기간 계산
max_days = self._get_max_available_days(팀_슬롯_데이터, 합주실_데이터)
total_weeks = (max_days + 6) // 7 # 주 수 계산
print(f"전체 기간: {max_days}일 ({total_weeks}주)")
# 주별 배정 결과 저장
weekly_results = []
# 전역 점유 상태 추적 (팀별로 이미 배정된 시간들)
global_team_allocations = {team_id: [] for team_id in 가용_팀들}
global_time_occupancy = set() # (day, hour) 튜플로 점유된 시간 추적
# 각 주마다 배정 수행
for week_num in range(total_weeks):
print(f"\n📅 {week_num + 1}주차 배정 시작...")
# 이번 주 범위 계산
week_start_day = week_num * 7
week_end_day = min((week_num + 1) * 7, max_days)
print(f" 📅 범위: {week_start_day}일 ~ {week_end_day-1}일")
# 이번 주에 대한 배정 수행
week_result = self._allocate_single_week(
week_num, week_start_day, week_end_day,
팀_슬롯_데이터, 합주실_데이터
)
if week_result:
weekly_results.append(week_result)
print(f" ✅ {week_num + 1}주차: {week_result.get('total_teams_allocated', 0)}팀 배정")
else:
print(f" ❌ {week_num + 1}주차: 배정 실패")
print(f"\n📊 전체 결과: {len(weekly_results)}주 배정 완료")
return weekly_results
def _allocate_single_week(self, week_num: int, week_start_day: int, week_end_day: int,
팀_슬롯_데이터: Dict, 합주실_데이터: Dict) -> Dict:
"""단일 주 배정 수행 (2칸 슬롯 알고리즘 사용)"""
print(f" 🔄 {week_num+1}주차 2칸 슬롯 알고리즘 적용")
# ⚡ 1단계: 주간 캐싱으로 성능 최적화
week_cache = self._cache_week_data(week_start_day, week_end_day, 팀_슬롯_데이터)
# 디버깅: 이번 주 가용한 팀들과 시간 확인
available_teams_this_week = []
for team_id, team_info in 팀_슬롯_데이터['teams_with_availability'].items():
team_slot = team_info['team_slot']
has_availability_this_week = False
for day in range(week_start_day, week_end_day):
if day < len(team_slot.slots.data):
day_slots = team_slot.slots.data[day]
if any(slot == 가용상태.가능 for slot in day_slots):
has_availability_this_week = True
break
if has_availability_this_week:
available_teams_this_week.append(team_id)
print(f" 📊 {week_num+1}주차 가용 팀: {len(available_teams_this_week)}개 (전체 {len(팀_슬롯_데이터['teams_with_availability'])}개 중)")
# 🤝 2단계: 2팀 페어 최대 조합 찾기 (캐시 사용)
all_pairs = self._find_all_possible_pairs_fast_cached(week_cache, 합주실_데이터)
if not all_pairs:
print(f" ❌ 가능한 페어 없음 → 독립 배정만 수행")
return self._fallback_week_allocation(
week_start_day, week_end_day, 팀_슬롯_데이터, 합주실_데이터
)
# 🎯 3단계: 최대 페어 조합 계산
print(f" 📊 발견된 가능 페어: {len(all_pairs)}개")
max_pair_combinations = self._find_maximum_pair_combinations(all_pairs)
if not max_pair_combinations:
print(f" ❌ 유효한 페어 조합 없음 → 독립 배정만 수행")
return self._fallback_week_allocation(
week_start_day, week_end_day, 팀_슬롯_데이터, 합주실_데이터
)
# 🏆 4단계: 최적 조합으로 배정 생성
best_combination = max_pair_combinations[0] # 첫 번째 최적 조합 선택
week_result = self._create_candidate_with_daytime_allocation(
best_combination, 팀_슬롯_데이터, 합주실_데이터, week_start_day, week_end_day
)
if week_result:
week_result['week_number'] = week_num + 1
week_result['week_start_day'] = week_start_day
week_result['week_end_day'] = week_end_day
return week_result
def _find_team_week_slot(self, team_slot, week_start_day: int, week_end_day: int, occupied_slots) -> Dict:
"""팀의 특정 주 범위에서 1시간 슬롯 찾기"""
# 이번 주 범위에서만 탐색
for day in range(week_start_day, week_end_day):
if day < len(team_slot.slots.data):
# 낮 시간대 우선 (10시~18시)
for hour in range(10, 18):
start_slot = hour * 4
if self._check_hour_available(team_slot.slots.data[day], start_slot):
time_slot = {
'day': day,
'hour': hour,
'start_slot': start_slot
}
return time_slot
return None
def _find_team_week_slot_simple(self, team_slot, week_start_day: int, week_end_day: int, occupied_time_slots: set) -> Dict:
"""주 범위에서 팀의 1시간 슬롯 찾기 (합주실 체크 제거)"""
for day in range(week_start_day, week_end_day):
if day >= len(team_slot.slots.data):
continue
for hour in range(9, 22): # 9시~22시
start_slot = hour * 4 # 1시간 = 4슬롯
# 이미 점유된 시간인지 체크
if (day, start_slot) in occupied_time_slots:
continue
# 팀이 이 1시간에 가능한지 체크
if self._check_hour_available(team_slot.slots.data[day], start_slot):
return {
'day': day,
'hour': hour,
'start_slot': start_slot
}
return None
def _find_common_hour_slot_fast(self, team1_slots, team2_slots) -> Dict:
"""두 팀의 공통 1시간 슬롯을 빠르게 찾기"""
# 전체 데이터 기간 체크
days_to_check = min(len(team1_slots.slots.data), len(team2_slots.slots.data))
for day in range(days_to_check):
# 9시~22시만 체크 (합주 가능 시간)
for hour in range(9, 22):
start_slot = hour * 4 # 15분 단위 → 슬롯 인덱스
# 1시간(4슬롯) 연속으로 둘 다 가능한지 체크
if self._check_hour_available(team1_slots.slots.data[day], start_slot) and \
self._check_hour_available(team2_slots.slots.data[day], start_slot):
return {
'day': day,
'hour': hour,
'start_slot': start_slot
}
return None
def _find_team_hour_slot_fast(self, team_slots) -> Dict:
"""팀의 1시간 슬롯을 빠르게 찾기"""
days_to_check = len(team_slots.slots.data) # 전체 기간
for day in range(days_to_check):
for hour in range(9, 22):
start_slot = hour * 4
if self._check_hour_available(team_slots.slots.data[day], start_slot):
return {
'day': day,
'hour': hour,
'start_slot': start_slot
}
return None
def _check_hour_available(self, day_slots, start_slot) -> bool:
"""1시간(4슬롯) 연속 가용성 체크"""
if start_slot + 3 >= len(day_slots): # 96슬롯 체크
return False
for i in range(4): # 1시간 = 4슬롯
if day_slots[start_slot + i] != 가용상태.가능:
return False
return True
def _find_available_room_fast(self, time_slot, 합주실_데이터, occupied_slots) -> str:
"""사용 가능한 합주실을 빠르게 찾기"""
day = time_slot['day']
hour = time_slot['hour']
start_slot = time_slot['start_slot']
for room_id, room_info in 합주실_데이터['rooms'].items():
room_slot = room_info.get('room_slot')
if not room_slot:
continue
# 1. 이미 점유된 시간인지 체크
if room_id in occupied_slots and \
day in occupied_slots[room_id] and \
hour in occupied_slots[room_id][day]:
continue
# 2. 합주실이 해당 시간에 가용한지 체크
if day < len(room_slot.slots.data) and \
self._check_hour_available(room_slot.slots.data[day], start_slot):
return room_id
return None
def _occupy_room_slot(self, occupied_slots, room_id, time_slot):
"""합주실 시간대 점유 표시"""
day = time_slot['day']
hour = time_slot['hour']
if room_id not in occupied_slots:
occupied_slots[room_id] = {}
if day not in occupied_slots[room_id]:
occupied_slots[room_id][day] = {}
occupied_slots[room_id][day][hour] = True
def _find_common_continuous_blocks(self, 팀_슬롯_데이터: Dict, 합주실_데이터: Dict) -> List[Dict]:
"""연속 시간 블록 찾기 (전체 기간 대응)"""
teams = list(팀_슬롯_데이터['teams_with_availability'].keys())
common_blocks = []
print(f" 🔍 {len(teams)}개 팀의 연속 블록 탐색...")
# 실제 데이터 기간 확인
max_days = self._get_max_available_days(팀_슬롯_데이터, 합주실_데이터)
print(f" 📅 전체 데이터 기간: {max_days}일")
# 각 날짜별로 연속 블록 찾기 (전체 기간)
for day in range(max_days):
print(f" 📅 {day}일 차 탐색...")
# 9시~22시 범위에서 30분 단위로 체크
for start_hour in range(9, 22):
block_info = self._analyze_time_block(day, start_hour, teams, 팀_슬롯_데이터, 합주실_데이터)
if block_info and len(block_info['available_teams']) >= 2: # 최소 2팀 이상
common_blocks.append(block_info)
print(f" ✅ {start_hour}시: {len(block_info['available_teams'])}팀 가능")
print(f" 📊 총 {len(common_blocks)}개 블록 발견")
return common_blocks
def _get_max_available_days(self, 팀_슬롯_데이터: Dict, 합주실_데이터: Dict) -> int:
"""실제 데이터에서 사용 가능한 최대 일 수 확인"""
max_days = 0
# 팀 데이터에서 최대 일 수 확인
for team_data in 팀_슬롯_데이터['teams_with_availability'].values():
team_slot = team_data['team_slot']
team_days = len(team_slot.slots.data)
max_days = max(max_days, team_days)
# 합주실 데이터에서도 확인
for room_info in 합주실_데이터['rooms'].values():
room_slot = room_info.get('room_slot')
if room_slot:
room_days = len(room_slot.slots.data)
max_days = max(max_days, room_days)
return max_days
def _cache_team_data_for_day(self, day: int, teams: List[str], 팀_슬롯_데이터: Dict) -> Dict[str, List[int]]:
"""하루치 팀 데이터를 미리 캐싱하여 성능 최적화"""
cached = {}
teams_with_availability = 팀_슬롯_데이터['teams_with_availability']
for team_id in teams:
if team_id not in teams_with_availability:
continue
team_slot = teams_with_availability[team_id]['team_slot']
# 범위 체크 후 정수 배열로 변환 (열거형 → 정수)
if day < len(team_slot.slots.data):
cached[team_id] = [1 if slot == 가용상태.가능 else 0
for slot in team_slot.slots.data[day]]
return cached
def _cache_week_data(self, week_start_day: int, week_end_day: int, 팀_슬롯_데이터: Dict) -> Dict:
"""주간 팀 데이터를 미리 캐싱하여 2칸 슬롯 알고리즘 최적화"""
import time
start_time = time.time()
teams = list(팀_슬롯_데이터['teams_with_availability'].keys())
week_cache = {
'teams': teams,
'daily_cache': {},
'team_combinations': []
}
# 각 날짜별 캐시 생성
for day in range(week_start_day, week_end_day):
week_cache['daily_cache'][day] = self._cache_team_data_for_day(day, teams, 팀_슬롯_데이터)
# 팀 조합 미리 생성 (성능 최적화)
from itertools import combinations
week_cache['team_combinations'] = list(combinations(teams, 2))
cache_time = time.time() - start_time
print(f" ⚡ 주간 캐시 생성: {len(teams)}팀 × {week_end_day-week_start_day}일 = {cache_time:.3f}s")
return week_cache
def _find_all_possible_pairs_fast_cached(self, week_cache: Dict, 합주실_데이터: Dict) -> List[Dict]:
"""캐시를 사용하여 모든 가능한 연속 페어 조합을 찾기 (초고속 버전)"""
import time
start_time = time.time()
print(" 🔍 캐시된 데이터로 팀 페어 조합 검색 중...")
all_pairs = []
team_combinations = week_cache['team_combinations'] # 미리 생성된 조합
daily_cache = week_cache['daily_cache']
# 각 팀 조합에 대해 캐시된 데이터로 빠른 검색
for team1_id, team2_id in team_combinations:
# 각 날짜에서 공통 1시간 슬롯 찾기
for day, day_cache in daily_cache.items():
team1_data = day_cache.get(team1_id)
team2_data = day_cache.get(team2_id)
if not team1_data or not team2_data:
continue
# 캐시된 데이터로 공통 시간 찾기 (초고속)
common_slot = self._find_common_hour_slot_fast_cached(team1_data, team2_data, day)
if common_slot:
# 공통 슬롯 개수 계산 (그리디 우선순위용)
common_count = self._count_common_slots_cached(team1_data, team2_data)
# 합주실 체크는 일단 기본값 사용 (성능 우선)
pair_info = {
'team1_id': team1_id,
'team2_id': team2_id,
'time_slot': common_slot,
'room_id': 'ROOM_DEFAULT', # 합주실 체크 간소화
'common_slots_count': common_count # 그리디 우선순위용
}
all_pairs.append(pair_info)
print(f" ✅ 페어 발견: {team1_id} & {team2_id}, 일 {day}, 시간 {common_slot['hour']}, 공통슬롯 {common_count}")
break # 한 쌍당 하나의 시간만 찾으면 충분
search_time = time.time() - start_time
print(f" ⚡ 페어 검색 완료: {len(all_pairs)}개 페어, {search_time:.3f}s")
return all_pairs
def _find_common_hour_slot_fast_cached(self, team1_data: List[int], team2_data: List[int], day: int) -> Dict:
"""캐시된 데이터로 두 팀의 공통 1시간 슬롯을 초고속 찾기"""
# 9시~22시 범위에서 1시간(4슬롯) 연속 체크
for hour in range(9, 22):
start_slot = hour * 4 # 15분 단위 → 슬롯 인덱스
# 범위 체크
if start_slot + 3 >= len(team1_data) or start_slot + 3 >= len(team2_data):
continue
# 1시간(4슬롯) 연속으로 둘 다 가능한지 체크 (정수 비교로 초고속)
both_available = True
for i in range(4):
if team1_data[start_slot + i] != 1 or team2_data[start_slot + i] != 1:
both_available = False
break
if both_available:
return {
'day': day,
'hour': hour,
'start_slot': start_slot
}
return None
def _count_common_slots_cached(self, team1_data: List[int], team2_data: List[int]) -> int:
"""캐시된 데이터로 두 팀의 총 공통 가능 슬롯 개수 계산"""
count = 0
min_length = min(len(team1_data), len(team2_data))
for i in range(min_length):
if team1_data[i] == 1 and team2_data[i] == 1:
count += 1
return count
def _find_maximum_pair_combinations(self, all_pairs: List[Dict]) -> List[List[Dict]]:
"""최대 연속 페어 조합 찾기 (팀 중복 없이) - 그리디 접근"""
print(f" 🎯 그리디 알고리즘 시작: {len(all_pairs)}개 페어로부터 최대 조합 선택")
# 그리디 접근: 페어를 순차적으로 선택하되 팀 중복과 시간 중복 없이 최대한 많이 선택
used_teams = set()
used_time_slots = set() # (day, hour) 튜플로 시간 중복 방지
selected_pairs = []
# 페어들을 우선순위에 따라 정렬 (예: 더 많은 공통 시간을 가진 페어 우선)
sorted_pairs = sorted(all_pairs, key=lambda p: p.get('common_slots_count', 1), reverse=True)
print(f" 📊 정렬된 페어 상위 5개: {[(p['team1_id'][:8], p['team2_id'][:8], p['common_slots_count']) for p in sorted_pairs[:5]]}")
for pair in sorted_pairs:
# 팀 중복 체크
if pair['team1_id'] in used_teams or pair['team2_id'] in used_teams:
continue
# 시간 중복 체크
time_slot_key = (pair['time_slot']['day'], pair['time_slot']['hour'])
if time_slot_key in used_time_slots:
continue
# 둘 다 통과하면 선택
selected_pairs.append(pair)
used_teams.add(pair['team1_id'])
used_teams.add(pair['team2_id'])
used_time_slots.add(time_slot_key)
# 결과를 리스트로 감싸서 기존 인터페이스와 호환
return [selected_pairs] if selected_pairs else []
def _create_candidate_with_daytime_allocation(self, pair_combination: List[Dict],
팀_슬롯_데이터: Dict, 합주실_데이터: Dict,
week_start_day: int, week_end_day: int) -> Dict:
"""페어 조합과 나머지 팀들의 낮 시간 배정으로 후보 생성"""
# 페어링된 팀들
paired_teams = set()
for pair in pair_combination:
paired_teams.add(pair['team1_id'])
paired_teams.add(pair['team2_id'])
# 나머지 팀들
all_teams = set(팀_슬롯_데이터['teams_with_availability'].keys())
remaining_teams = all_teams - paired_teams
# 페어들의 점유 시간 수집 (시간 충돌 방지용)
occupied_time_slots = set()
for pair in pair_combination:
day = pair['time_slot']['day']
hour = pair['time_slot']['hour']
# 1시간 전체를 점유로 표시 (30분씩 2팀이므로)
occupied_time_slots.add((day, hour))
# 나머지 팀들을 낮 시간대(10시-18시)에 배정
daytime_allocations = []
for team_id in remaining_teams:
team_slot = 팀_슬롯_데이터['teams_with_availability'][team_id]['team_slot']
# 현재 주차 범위 내에서만 찾도록 수정 필요
daytime_slot = self._find_team_daytime_slot_with_conflict_check(
team_slot, occupied_time_slots, week_start_day, week_end_day
)
if daytime_slot:
daytime_allocations.append({
'team_id': team_id,
'time_slot': daytime_slot,
'room_id': 'ROOM_DEFAULT' # 합주실 체크 간소화
})
# 시간 점유 추가
occupied_time_slots.add((daytime_slot['day'], daytime_slot['hour']))
# 되는시간 타입으로 변환
rehearsal_times = self._convert_to_되는시간_type(pair_combination, daytime_allocations)
return {
'continuous_pairs': pair_combination,
'daytime_allocations': daytime_allocations,
'rehearsal_times': rehearsal_times,
'total_teams_allocated': len(paired_teams) + len(daytime_allocations)
}
def _find_team_daytime_slot_fast(self, team_slot, occupied_slots) -> Dict:
"""팀의 낮 시간대 1시간 슬롯을 빠르게 찾기 (10시-18시)"""
days_to_check = len(team_slot.slots.data) # 전체 기간
for day in range(days_to_check):
# 낮 시간대만 체크 (10시~18시)
for hour in range(10, 18):
start_slot = hour * 4
if self._check_hour_available(team_slot.slots.data[day], start_slot):
# 이미 점유된 시간인지 체크는 room 할당 시 수행
time_slot = {
'day': day,
'hour': hour,
'start_slot': start_slot
}
# 점유 충돌 체크는 room 할당 시 수행
return time_slot
return None
def _find_team_daytime_slot_with_conflict_check(self, team_slot, occupied_time_slots: set,
week_start_day: int, week_end_day: int) -> Dict:
"""팀의 낮 시간대 1시간 슬롯을 찾기 (시간 충돌 체크 포함, 주차 범위 제한)"""
# 현재 주차 범위 내에서만 검색
for day in range(week_start_day, week_end_day):
# 낮 시간대만 체크 (10시~18시)
for hour in range(10, 18):
# 이미 점유된 시간인지 먼저 체크
if (day, hour) in occupied_time_slots:
continue
start_slot = hour * 4
# 날짜와 슬롯 범위 체크
if (day >= len(team_slot.slots.data) or
start_slot + 3 >= len(team_slot.slots.data[day])):
continue
# 1시간(4슬롯) 연속으로 가능한지 체크
available = True
for i in range(4):
if team_slot.slots.data[day][start_slot + i] != 가용상태.가능:
available = False
break
if available:
return {
'day': day,
'hour': hour,
'start_slot': start_slot
}
return None
def _convert_to_되는시간_type(self, pair_combination: List[Dict], daytime_allocations: List[Dict]) -> List[Dict]:
"""배정 결과를 되는시간 타입으로 변환"""
rehearsal_times = []
# 연속 페어들을 30분씩 나누어 처리
for pair in pair_combination:
day = pair['time_slot']['day']
hour = pair['time_slot']['hour']
# 각 팀의 30분 시간대를 되는시간으로 변환
team1_time = self._create_되는시간_for_slot(day, hour, 0, 30) # 첫 30분
team2_time = self._create_되는시간_for_slot(day, hour, 30, 60) # 다음 30분
rehearsal_times.append({
'team_id': pair['team1_id'],
'rehearsal_time': team1_time,
'room_id': pair['room_id'],
'duration': '30분'
})
rehearsal_times.append({
'team_id': pair['team2_id'],
'rehearsal_time': team2_time,
'room_id': pair['room_id'],
'duration': '30분'
})
# 낮 시간 독립 팀들 처리
for alloc in daytime_allocations:
day = alloc['time_slot']['day']
hour = alloc['time_slot']['hour']
team_time = self._create_되는시간_for_slot(day, hour, 0, 60) # 1시간
rehearsal_times.append({
'team_id': alloc['team_id'],
'rehearsal_time': team_time,
'room_id': alloc['room_id'],
'duration': '1시간'
})
return rehearsal_times
def _create_되는시간_for_slot(self, day: int, hour_float: float, start_min: int, end_min: int) -> Dict:
"""특정 슬롯을 되는시간 타입으로 변환"""
# float hour를 정수 hour + minute로 분해
hour = int(hour_float)
additional_minutes = int((hour_float - hour) * 60)
actual_start_min = start_min + additional_minutes
actual_end_min = end_min + additional_minutes
# 월요일 기준으로 날짜 계산
base_date = datetime(2024, 1, 1) # 임시 기준일 (월요일)
actual_date = base_date + timedelta(days=day)
# 시작 시간 계산
if actual_start_min >= 60:
start_hour = hour + (actual_start_min // 60)
start_minute = actual_start_min % 60
else:
start_hour = hour
start_minute = actual_start_min
start_time = actual_date.replace(hour=start_hour, minute=start_minute, second=0, microsecond=0)
# 종료 시간 계산
if actual_end_min >= 60:
end_hour = hour + (actual_end_min // 60)
end_minute = actual_end_min % 60
else:
end_hour = hour
end_minute = actual_end_min
end_time = actual_date.replace(hour=end_hour, minute=end_minute, second=0, microsecond=0)
return {
'start_time': start_time,
'end_time': end_time,
'day_of_week': day,
'time_range': f"{start_hour:02d}:{start_minute:02d}-{end_hour:02d}:{end_minute:02d}",
'room_id': 'ROOM_DEFAULT', # 기본값 (합주실 체크 제거로 인한 호환성)
'day': day,
'hour': hour_float
}
# 유틸리티 메서드들
def _calculate_total_available_hours(self, availability: 되는시간Type) -> float:
"""총 가용시간을 시간 단위로 계산 (되는시간Type용)"""
total_minutes = 0
for slot in availability.가능시간:
duration = (slot.종료시간 - slot.시작시간).total_seconds() / 60
total_minutes += duration
return total_minutes / 60
def _calculate_available_hours_from_slots(self, slots: 되는시간슬롯) -> float:
"""되는시간슬롯에서 실제 가용시간을 계산 (가용상태.가능인 슬롯만)"""
total_available_slots = 0
for day_slots in slots.data:
for slot_state in day_slots:
if slot_state == 가용상태.가능:
total_available_slots += 1
# 15분 단위 슬롯이므로 4로 나누면 시간 단위
return total_available_slots * 0.25
def _get_availability_time_range(self, availability: 되는시간Type) -> Tuple[Optional[datetime], Optional[datetime]]:
"""가용시간의 전체 범위 계산"""
if not availability.가능시간:
return None, None
earliest = min(slot.시작시간 for slot in availability.가능시간)
latest = max(slot.종료시간 for slot in availability.가능시간)
return earliest, latest
def _find_continuous_blocks(self, slots: 되는시간슬롯) -> List[List[Tuple[int, int]]]:
"""연속된 가용시간 블록들을 찾기"""
blocks = []
for day_idx, day_slots in enumerate(slots.data):
current_block = []
for slot_idx, slot_state in enumerate(day_slots):
if slot_state == 가용상태.가능:
current_block.append((day_idx, slot_idx))
else:
if current_block and len(current_block) >= 2: # 최소 30분 (2슬롯)
blocks.append(current_block)
current_block = []
# 하루 끝에서 블록 정리
if current_block and len(current_block) >= 2:
blocks.append(current_block)
return blocks
def _create_time_slot_from_block(self, block: List[Tuple[int, int]], start_date: date, duration_type: str) -> Dict:
"""블록으로부터 시간 슬롯 생성"""
start_day_idx, start_slot_idx = block[0]
end_day_idx, end_slot_idx = block[-1]
# 시작 시간 계산
start_datetime = self._slot_to_datetime((start_day_idx, start_slot_idx), start_date)
end_datetime = self._slot_to_datetime((end_day_idx, end_slot_idx + 1), start_date)
return {
'start_time': start_datetime,
'end_time': end_datetime,
'duration_type': duration_type,
'duration_hours': (end_datetime - start_datetime).total_seconds() / 3600,
'block_indices': block
}
def _slot_to_datetime(self, slot: Tuple[int, int], start_date: date) -> datetime:
"""슬롯 인덱스를 datetime으로 변환"""
day_idx, slot_idx = slot
target_date = start_date + timedelta(days=day_idx)
# 슬롯 인덱스가 96을 넘어가면 다음날로 넘어감
if slot_idx >= 96:
extra_days = slot_idx // 96
slot_idx = slot_idx % 96
target_date = target_date + timedelta(days=extra_days)
hour = slot_idx // 4
minute = (slot_idx % 4) * 15
# hour가 24를 넘어가면 다음날로 넘어감
if hour >= 24:
extra_days = hour // 24
hour = hour % 24
target_date = target_date + timedelta(days=extra_days)
return datetime.combine(target_date, datetime.min.time().replace(hour=hour, minute=minute))
def _check_room_availability(self, time_slot: Dict, 연습실별슬롯: List[연습실되는시간슬롯]) -> List[str]:
"""특정 시간에 사용 가능한 합주실 찾기"""
available_rooms = []
for room_slot in 연습실별슬롯:
if self._is_room_available_at_time(time_slot, room_slot):
available_rooms.append(room_slot.room_id)
return available_rooms
def _is_room_available_at_time(self, time_slot: Dict, room_slot: 연습실되는시간슬롯) -> bool:
"""합주실이 특정 시간에 사용 가능한지 확인"""
start_time = time_slot['start_time']
end_time = time_slot['end_time']
# 시간을 슬롯 인덱스로 변환
start_day_offset = (start_time.date() - room_slot.slots.start_date).days
end_day_offset = (end_time.date() - room_slot.slots.start_date).days
if start_day_offset < 0 or start_day_offset >= len(room_slot.slots.data):
return False
if end_day_offset < 0 or end_day_offset >= len(room_slot.slots.data):
return False
# 해당 시간 범위의 모든 슬롯이 가용한지 확인
current_time = start_time
while current_time < end_time:
day_offset = (current_time.date() - room_slot.slots.start_date).days
slot_idx = (current_time.hour * 4) + (current_time.minute // 15)
if (day_offset < 0 or day_offset >= len(room_slot.slots.data) or
slot_idx < 0 or slot_idx >= 96 or
room_slot.slots.data[day_offset][slot_idx] != 가용상태.가능):
return False
current_time += timedelta(minutes=15)
return True
def _find_continuous_pairs(self, 팀별슬롯: List, 연습실별슬롯: List) -> List[Dict]:
"""연속 30분 페어링 가능한 팀 조합 찾기"""
pairs = []
print(f" 🔍 {len(팀별슬롯)}개 팀으로 페어링 시작...")
for i, team1_slot in enumerate(팀별슬롯):
for team2_slot in 팀별슬롯[i+1:]:
# 두 팀의 교집합된 가용시간에서 연속 1시간 블록 찾기
common_blocks = self._find_common_continuous_blocks(team1_slot, team2_slot)
for block in common_blocks:
if len(block) >= 4: # 최소 1시간(4슬롯) 필요
# 이 시간대에 사용 가능한 합주실 확인
time_slot = self._create_time_slot_from_block(block[:4], team1_slot.slots.start_date, "1시간")
available_rooms = self._check_room_availability(time_slot, 연습실별슬롯)
if available_rooms:
# 30분씩 나누어 배정
team1_30min = {
'start_time': time_slot['start_time'],
'end_time': time_slot['start_time'] + timedelta(minutes=30),
'duration_type': '30분',
'duration_hours': 0.5
}
team2_30min = {
'start_time': time_slot['start_time'] + timedelta(minutes=30),
'end_time': time_slot['end_time'],
'duration_type': '30분',
'duration_hours': 0.5
}
pairs.append({
'team1_id': team1_slot.team_id,
'team2_id': team2_slot.team_id,
'team1_slot': team1_30min,
'team2_slot': team2_30min,
'room_id': available_rooms[0], # 첫 번째 가용 합주실 사용
'total_duration': 1.0,
'start_time': time_slot['start_time'],
'block_info': block[:4]
})
print(f" 📊 총 {len(pairs)}개 페어 발견")
return pairs
# 시각화 메서드들
def visualize_member_availability(self, members_data: Dict):
"""Step 1 결과 시각화"""
print(f" 📊 총 {members_data['summary']['total_members']}명 중 {members_data['summary']['members_with_availability']}명이 가용시간 등록")
print(f" 📈 등록률: {members_data['summary']['coverage_rate']:.1%}")
# 시간별 분포
hour_distribution = defaultdict(int)
for member_data in members_data['members'].values():
if member_data['has_availability'] and member_data['availability_data']:
for slot in member_data['availability_data'].가능시간:
for hour in range(slot.시작시간.hour, slot.종료시간.hour):
hour_distribution[hour] += 1
if hour_distribution:
print(" ⏰ 시간대별 가용 멤버 수:")
for hour in sorted(hour_distribution.keys()):
bar = "█" * (hour_distribution[hour] // 5 + 1)
print(f" {hour:02d}시: {bar} ({hour_distribution[hour]}명)")
def visualize_team_availability(self, teams_data: Dict):
"""Step 2 결과 시각화"""
print(f" ✅ 배정 가능한 팀: {teams_data['summary']['teams_with_availability']}개")
print(f" ❌ 배정 불가능한 팀: {teams_data['summary']['teams_without_availability']}개")
# 배정 가능한 팀들의 가용시간 분포 (실제 교집합 기준)
if teams_data['teams_with_availability']:
print(" 📋 배정 가능한 팀별 실제 가용시간 (교집합):")
sorted_teams = sorted(
teams_data['teams_with_availability'].items(),
key=lambda x: x[1]['available_hours']
)
for team_id, team_data in sorted_teams:
print(f" 팀 {team_id[-8:]}: {team_data['available_hours']:.1f}시간 ({team_data['member_count']}명)")
# 배정 불가능한 팀들의 이유
if teams_data['teams_without_availability']:
print(" ⚠️ 배정 불가능한 팀들:")
reason_counts = defaultdict(int)
for team_data in teams_data['teams_without_availability'].values():
reason_counts[team_data['reason']] += 1
for reason, count in reason_counts.items():
reason_text = {
'no_member_availability': '멤버 가용시간 없음',
'partial_member_availability': '일부 멤버 가용시간 없음',
'no_common_time': '공통 가용시간 없음 (교집합 결과 0)'
}.get(reason, reason)
print(f" {reason_text}: {count}개 팀")
# 가용시간별 분포 히스토그램
if teams_data['teams_with_availability']:
print(" 📊 가용시간 분포:")
hours_list = [data['available_hours'] for data in teams_data['teams_with_availability'].values()]
ranges = [(0, 50), (50, 100), (100, 200), (200, 300), (300, float('inf'))]
for start, end in ranges:
count = sum(1 for hours in hours_list if start <= hours < end)
if count > 0:
range_text = f"{start}-{end}시간" if end != float('inf') else f"{start}시간+"
bar = "█" * (count // 2 + 1)
print(f" {range_text}: {bar} ({count}개 팀)")
def visualize_room_availability(self, rooms_data: Dict):
"""Step 3 결과 시각화"""
print(f" 🏠 총 {rooms_data['summary']['total_rooms']}개 합주실 중 {rooms_data['summary']['rooms_with_availability']}개 사용 가능")
if rooms_data['global_availability_window']:
window = rooms_data['global_availability_window']
print(f" ⏰ 전체 가용 시간대: {window['earliest_time'].strftime('%H:%M')} ~ {window['latest_time'].strftime('%H:%M')}")
print(f" 📏 총 가용 시간: {window['total_hours']:.1f}시간")
# 합주실별 세부 정보
for room_data in rooms_data['rooms'].values():
if room_data.get('available_hours', 0) > 0:
print(f" {room_data['name']}: {room_data['available_hours']:.1f}시간 가용")
def visualize_possible_allocations(self, possible_allocations: Dict):
"""Step 4 결과 시각화 (슬롯 기반 새로운 구조)"""
summary = possible_allocations['summary']
print(f" 🎲 {summary['total_teams']}개 팀 분석")
print(f" ✅ {summary['allocated_teams']}개 팀 배정 성공")
print(f" 🤝 {summary['continuous_pairs']}개 연속 페어")
print(f" 🚶 {summary['independent_teams']}개 독립 배정")
print(f" 📊 배정 성공률: {summary['allocation_rate']:.1%}")
# 배정 결과 상세 정보
weekly_allocations = possible_allocations.get('weekly_allocations', [])
# 페어 배정 정보
pairs = [a for a in weekly_allocations if a.get('type') == 'pair']
if pairs:
print(" 🤝 연속 페어 배정:")
for i, pair in enumerate(pairs, 1):
print(f" {i}. 팀 {pair['team1_id'][-6:]} + {pair['team2_id'][-6:]}: "
f"{pair['day']}일차 {pair['start_hour']:02d}시 (30분×2)")
# 독립 팀 배정 정보
independents = [a for a in weekly_allocations if a.get('type') == 'independent']
if independents:
print(" 🚶 독립 팀 배정:")
for i, allocation in enumerate(independents, 1):
print(f" {i}. 팀 {allocation['team_id'][-6:]}: "
f"{allocation['day']}일차 {allocation['start_hour']:02d}시 (1시간)")
# 배정 실패 팀
failed = [a for a in weekly_allocations if a.get('type') == 'failed']
if failed:
print(" ❌ 배정 실패 팀:")
for i, failure in enumerate(failed, 1):
print(f" {i}. 팀 {failure['team_id'][-6:]}: {failure.get('reason', '알 수 없음')}")
# 시간대별 분포
time_distribution = defaultdict(int)
day_distribution = defaultdict(int)
for allocation in weekly_allocations:
if allocation.get('allocated'):
if 'start_hour' in allocation:
time_distribution[allocation['start_hour']] += 1
if 'day' in allocation:
day_distribution[allocation['day']] += 1
if time_distribution:
print(" ⏰ 시간대별 배정 분포:")
for hour in sorted(time_distribution.keys()):
count = time_distribution[hour]
bar = "█" * (count // 2 + 1)
print(f" {hour:02d}시: {bar} ({count}개)")
if day_distribution:
print(" 📅 요일별 배정 분포:")
days = ['월', '화', '수', '목', '금', '토', '일']
for day in sorted(day_distribution.keys()):
count = day_distribution[day]
day_name = days[day] if day < len(days) else f"{day}일"
bar = "█" * (count // 2 + 1)
print(f" {day_name}요일: {bar} ({count}개)")
def visualize_final_candidates(self, final_candidates: List[Dict]):
"""최종 후보들 시각화 (새로운 연속 알고리즘용)"""
if not final_candidates:
print(" ❌ 생성된 후보가 없습니다.")
return
print(f"\n 🎯 최적 배정 결과:")
for i, result in enumerate(final_candidates, 1):
print(f"\n 📋 결과 {i}:")
# 연속 배정 정보 (페어)
continuous_pairs = result.get('continuous_pairs', [])
max_continuous = len(continuous_pairs) * 2 # 페어당 2팀
print(f" 🔗 최대 연속 배정: {max_continuous}개팀 (30분씩)")
if continuous_pairs:
print(f" 📅 연속 배정 상세:")
for j, pair in enumerate(continuous_pairs):
day = pair['time_slot']['day']
hour = pair['time_slot']['hour']
print(f" {j+1}. 페어: {pair['team1_id'][-6:]} & {pair['team2_id'][-6:]} → {day}일 {hour}시-{hour+1}시")
# 독립 배정 정보
daytime_allocations = result.get('daytime_allocations', [])
print(f" 🚶 독립 배정: {len(daytime_allocations)}개팀 (1시간씩)")
for alloc in daytime_allocations:
day = alloc['time_slot']['day']
hour = alloc['time_slot']['hour']
print(f" - {alloc['team_id'][-6:]} → {day}일 {hour}시-{hour+1}시")
# 되는시간 타입 결과
rehearsal_times = result.get('rehearsal_times', [])
print(f" 📊 최종 배정: {len(rehearsal_times)}개 배정")
total_allocated = result.get('total_teams_allocated', 0)
print(f" ✅ 총 배정 팀 수: {total_allocated}개")
# 연속성 효율 표시
if max_continuous > 0:
efficiency = (max_continuous / total_allocated * 100) if total_allocated > 0 else 0
print(f" 📈 연속성 효율: {efficiency:.1f}% ({max_continuous}/{total_allocated})")
# 🎯 주별 스케줄 시각화 추가
self._visualize_weekly_schedule(final_candidates)
def _visualize_weekly_schedule(self, results: List[Dict]):
"""전체 주별 스케줄 시각화"""
print(f"\n 📅 전체 주간 스케줄 요약:")
print(f" {'='*80}")
if not results:
print(" ❌ 배정된 결과가 없습니다.")
return
# 모든 주의 배정을 통합
all_allocations = []
total_weeks = len(results)
total_teams_all_weeks = 0
for result in results:
week_num = result.get('week_number', 0)
# 연속 배정 (페어) 추가
for pair in result.get('continuous_pairs', []):
# 첫 번째 팀
rehearsal_time = pair['time_slot']
all_allocations.append({
'team_id': pair['team1_id'],
'day': rehearsal_time['day'],
'hour': rehearsal_time['hour'],
'duration': '30분',
'type': '연속',
'week': week_num,
'sequence': 1
})
# 두 번째 팀
all_allocations.append({
'team_id': pair['team2_id'],
'day': rehearsal_time['day'],
'hour': rehearsal_time['hour'] + 0.5, # 30분 뒤
'duration': '30분',
'type': '연속',
'week': week_num,
'sequence': 2
})
# 독립 배정 추가
for alloc in result.get('daytime_allocations', []):
rehearsal_time = alloc['time_slot']
all_allocations.append({
'team_id': alloc['team_id'],
'day': rehearsal_time['day'],
'hour': rehearsal_time['hour'],
'duration': '1시간',
'type': '독립',
'week': week_num,
'sequence': 0
})
total_teams_all_weeks += result.get('total_teams_allocated', 0)
print(f"\n 📊 전체 배정 현황: {total_weeks}주, 총 {total_teams_all_weeks}팀 배정")
# 주별로 그룹화하여 표시
days = ['월', '화', '수', '목', '금', '토', '일']
for week_num in range(1, total_weeks + 1):
week_allocations = [a for a in all_allocations if a['week'] == week_num]
if week_allocations:
print(f"\n 🗓️ {week_num}주차 상세 일정:")
print(f" {'-'*60}")
# 날짜별로 그룹화
week_days = sorted(set(a['day'] for a in week_allocations))
for day_idx in week_days:
day_allocations = [a for a in week_allocations if a['day'] == day_idx]
day_of_week = day_idx % 7
# 더 간단하고 직관적인 날짜 표시
print(f"\n 📆 {days[day_of_week]}요일 (절대 {day_idx}일):")
# 시간 순으로 정렬
day_allocations.sort(key=lambda x: (x['hour'], x['sequence']))
for alloc in day_allocations:
hour = alloc['hour']
team_id = alloc['team_id'][-6:]
duration = alloc['duration']
type_icon = '🔗' if alloc['type'] == '연속' else '🚶'
if alloc['type'] == '연속':
if isinstance(hour, float):
hour_int = int(hour)
minute = int((hour - hour_int) * 60)
end_minute = minute + 30
if end_minute >= 60:
end_hour = hour_int + 1
end_minute = 0
else:
end_hour = hour_int
time_str = f"{hour_int:02d}:{minute:02d}-{end_hour:02d}:{end_minute:02d}"
else:
time_str = f"{hour:02d}:00-{hour:02d}:30"
else:
time_str = f"{int(hour):02d}:00-{int(hour)+1:02d}:00"
print(f" {type_icon} {time_str} | {team_id} | {duration}")
# 전체 주별 통계
print(f"\n 📈 주별 배정 요약:")
for week_num in range(1, total_weeks + 1):
week_allocations = [a for a in all_allocations if a['week'] == week_num]
continuous_count = len([a for a in week_allocations if a['type'] == '연속'])
independent_count = len([a for a in week_allocations if a['type'] == '독립'])
total_count = len(week_allocations)
print(f" {week_num}주차: {total_count}팀 (연속: {continuous_count}, 독립: {independent_count})")
# 시간대별 통계
print(f"\n 📊 시간대별 배정 현황:")
hour_stats = {}
for alloc in all_allocations:
hour_key = int(alloc['hour'])
if hour_key not in hour_stats:
hour_stats[hour_key] = []
hour_stats[hour_key].append(alloc['team_id'][-4:])
for hour in sorted(hour_stats.keys()):
teams = hour_stats[hour]
print(f" {hour}시대: {len(teams)}팀 ({', '.join(teams)})")
def main():
"""메인 실행 함수"""
performance_id = "01K3R45Q2DKX2VSSE4G59CS4YZ"
tester = StepByStepTester()
try:
results = tester.run_full_test(performance_id)
print(f"\n🎉 단계별 테스트 완료!")
except Exception as e:
print(f"\n❌ 테스트 실패: {e}")
import traceback
traceback.print_exc()
if __name__ == "__main__":
main()
Python
복사

