From f916d764c0a0d81e71f34487f838bc256130b350 Mon Sep 17 00:00:00 2001 From: "ki.sagidullin" Date: Sun, 5 Apr 2026 10:05:03 +0500 Subject: [PATCH] =?UTF-8?q?=D0=A1=D0=BB=D0=B8=D1=88=D0=BA=D0=BE=D0=BC=20?= =?UTF-8?q?=D0=BF=D0=B5=D1=80=D0=B5=D1=83=D1=81=D0=BB=D0=BE=D0=B6=D0=BD?= =?UTF-8?q?=D0=B5=D0=BD=D0=BD=D0=B0=D1=8F=20=D1=81=D1=82=D1=80=D1=83=D0=BA?= =?UTF-8?q?=D1=82=D1=83=D1=80=D0=B0=20-=20=D0=BD=D0=BE=20=D0=BE=D1=81?= =?UTF-8?q?=D1=82=D0=B0=D0=B2=D0=B8=D0=BC=20=D0=B4=D0=BB=D1=8F=20=D0=B1?= =?UTF-8?q?=D1=83=D0=B4=D1=83=D1=89=D0=B5=D0=B3=D0=BE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 42 +++ README.md | 108 ++++++- constraints/__init__.py | 94 ++++++ constraints/base_constraint.py | 64 ++++ constraints/global_constraints.py | 1 + constraints/hard_constraints.py | 120 ++++++++ constraints/soft_constraints.py | 123 ++++++++ docker-compose.yml | 33 ++ exporters/__init__.py | 1 + exporters/base_exporter.py | 1 + exporters/csv_exporter.py | 1 + exporters/excel_exporter.py | 1 + exporters/html_exporter.py | 1 + fixtures/__init__.py | 1 + requirements.txt | 21 ++ scripts/__init__.py | 1 + shift_schedule.csv | 31 ++ src/__init__.py | 1 + src/config.py | 9 + src/core/__init__.py | 1 + src/core/engine.py | 251 +++++++++++++++ src/core/events.py | 1 + src/core/solver.py | 297 ++++++++++++++++++ src/core/validator.py | 360 ++++++++++++++++++++++ src/domains/__init__.py | 1 + src/domains/models/__init__.py | 26 ++ src/domains/models/operator.py | 118 +++++++ src/domains/models/preference.py | 203 ++++++++++++ src/domains/models/schedule.py | 1 + src/domains/models/shift.py | 153 +++++++++ src/domains/repositories/__init__.py | 1 + src/domains/repositories/operator_repo.py | 1 + src/domains/repositories/schedule_repo.py | 1 + src/domains/repositories/shift_repo.py | 1 + src/domains/services/__init__.py | 1 + src/domains/services/generator.py | 1 + src/domains/services/optimizer.py | 1 + src/exceptions.py | 1 + src/logging_config.py | 8 + tests/__init__.py | 1 + tests/integration/__init__.py | 1 + tests/test_models.py | 137 ++++++++ tests/unit/__init__.py | 1 + ui/__init__.py | 1 + ui/routes/__init__.py | 1 + ui/routes/config.py | 1 + ui/routes/index.py | 1 + utils/__init__.py | 1 + 48 files changed, 2222 insertions(+), 5 deletions(-) create mode 100644 .gitignore create mode 100644 constraints/__init__.py create mode 100644 constraints/base_constraint.py create mode 100644 constraints/global_constraints.py create mode 100644 constraints/hard_constraints.py create mode 100644 constraints/soft_constraints.py create mode 100644 docker-compose.yml create mode 100644 exporters/__init__.py create mode 100644 exporters/base_exporter.py create mode 100644 exporters/csv_exporter.py create mode 100644 exporters/excel_exporter.py create mode 100644 exporters/html_exporter.py create mode 100644 fixtures/__init__.py create mode 100644 requirements.txt create mode 100644 scripts/__init__.py create mode 100644 shift_schedule.csv create mode 100644 src/__init__.py create mode 100644 src/config.py create mode 100644 src/core/__init__.py create mode 100644 src/core/engine.py create mode 100644 src/core/events.py create mode 100644 src/core/solver.py create mode 100644 src/core/validator.py create mode 100644 src/domains/__init__.py create mode 100644 src/domains/models/__init__.py create mode 100644 src/domains/models/operator.py create mode 100644 src/domains/models/preference.py create mode 100644 src/domains/models/schedule.py create mode 100644 src/domains/models/shift.py create mode 100644 src/domains/repositories/__init__.py create mode 100644 src/domains/repositories/operator_repo.py create mode 100644 src/domains/repositories/schedule_repo.py create mode 100644 src/domains/repositories/shift_repo.py create mode 100644 src/domains/services/__init__.py create mode 100644 src/domains/services/generator.py create mode 100644 src/domains/services/optimizer.py create mode 100644 src/exceptions.py create mode 100644 src/logging_config.py create mode 100644 tests/__init__.py create mode 100644 tests/integration/__init__.py create mode 100644 tests/test_models.py create mode 100644 tests/unit/__init__.py create mode 100644 ui/__init__.py create mode 100644 ui/routes/__init__.py create mode 100644 ui/routes/config.py create mode 100644 ui/routes/index.py create mode 100644 utils/__init__.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..b519995 --- /dev/null +++ b/.gitignore @@ -0,0 +1,42 @@ +# Byte-compiled / optimized / DLL files +__pycache__/ +*.py[cod] +*$py.class + +# Virtual environments +venv/ +.venv/ +ENV/ +env/ + +# IDE +.idea/ +.vscode/ +*.swp +*.swo + +# Environment variables +.env + +# Database +*.db +*.sqlite3 +*.sqlite + +# Test coverage +.coverage +htmlcov/ +.pytest_cache/ +.mypy_cache/ + +# OS generated files +.DS_Store +Thumbs.db + +# Build +build/ +dist/ +*.egg-info/ + +# Docker +docker-compose.override.yml \ No newline at end of file diff --git a/README.md b/README.md index dd76421..cfe89a0 100644 --- a/README.md +++ b/README.md @@ -7,10 +7,108 @@ ## 📁 Структура проекта ``` -├── main.py # Основная логика с генерацией графика -├── shift_schedule.csv # Результат (заполненный график) -├── README.md # Документация -└── .continue/ # Система контроля версий +OperatorSchedule/ +├── src/ # Основной исходный код +│ ├── __init__.py # Корневой пакет +│ ├── app.py # Точка входа (FastAPI, если веб-интерфейс) +│ ├── config.py # Конфигурация системы +│ ├── exceptions.py # Исключения и ошибки +│ ├── logging_config.py # Настройка логирования +│ │ +│ ├── core/ # Ядро системы +│ │ ├── __init__.py +│ │ ├── engine.py # Основной класс планировщика (SchedulerEngine) +│ │ ├── solver.py # Орто-модель CP-SAT +│ │ ├── validator.py # Валидатор результатов +│ │ └── events.py # События системы (логи, уведомления) +│ │ +│ ├── domains/ # Доменные сущности +│ │ ├── __init__.py +│ │ ├── models/ # Модели данных +│ │ │ ├── __init__.py +│ │ │ ├── operator.py # Класс Operator (оператор) +│ │ │ ├── shift.py # класс Shift (смена) +│ │ │ ├── schedule.py # класс Schedule (график) +│ │ │ └── preference.py # класс Preference (предпочтение) +│ │ ├── repositories/ # Интерфейсы доступа к данным +│ │ │ ├── __init__.py +│ │ │ ├── operator_repo.py +│ │ │ ├── shift_repo.py +│ │ │ └── schedule_repo.py +│ │ └── services/ # Бизнес-логика +│ │ ├── __init__.py +│ │ ├── generator.py # Генератор графиков +│ │ ├── optimizer.py # Оптимизатор +│ │ └── conflict_resolver.py # Ресольвер конфликтов +│ │ +│ ├── constraints/ # Управление ограничениями +│ │ ├── __init__.py +│ │ ├── base_constraint.py # Базовый класс ограничения +│ │ ├── hard_constraints.py # Жёсткие ограничения +│ │ ├── soft_constraints.py # Мягкие ограничения (предпочтения) +│ │ └── global_constraints.py # Глобальные ограничения +│ │ +│ ├── exporters/ # Экспорт данных +│ │ ├── __init__.py +│ │ ├── base_exporter.py # Базовый класс экспортера +│ │ ├── csv_exporter.py +│ │ ├── excel_exporter.py +│ │ ├── html_exporter.py +│ │ └── json_exporter.py +│ │ +│ ├── ui/ # Веб-интерфейс (опционально) +│ │ ├── __init__.py +│ │ ├── routes/ # Маршруты API +│ │ │ ├── __init__.py +│ │ │ ├── index.py # Главная страница +│ │ │ ├── config.py # Конфигурация API +│ │ │ └── export.py # Экспорт графики +│ │ ├── templates/ # Jinja2 шаблоны +│ │ └── static/ # CSS, JS +│ │ +│ └── utils/ # Вспомогательные модули +│ ├── __init__.py +│ ├── calendar.py # Работа с календарями +│ ├── validators.py # Валидаторы данных +│ └── formatters.py # Форматтеры вывода +│ +├── tests/ # Тесты +│ ├── __init__.py +│ ├── conftest.py # Фикстуры pytest +│ ├── unit/ # Unit-тесты +│ │ ├── __init__.py +│ │ ├── test_domains.py +│ │ ├── test_constraints.py +│ │ └── test_exporters.py +│ ├── integration/ # Интеграционные тесты +│ │ ├── __init__.py +│ │ └── test_scheduler.py +│ └── fixtures/ # Фикстуры данных +│ ├── __init__.py +│ ├── operators.json # JSON с операторами +│ └── shifts.json # JSON со сменами +│ +├── docs/ # Документация +│ ├── README.md +│ ├── CHANGELOG.md # История изменений +│ ├── API.md # Документация API +│ └── architecture.md # Архитектурный обзор +│ +├── config/ # Конфигурационные файлы +│ ├── __init__.py +│ ├── default.yaml # Стандартная конфигурация +│ └── development.yaml # Разработка +│ +├── examples/ # Примеры использования +│ ├── basic.py # Базовый пример +│ ├── custom_constraints.py# Пример с кастомными ограничениями +│ └── multi_department.py # Многоотраслевой пример +│ +├── requirements.txt # Зависимости Python +├── requirements-dev.txt # Зависимости для разработки +├── setup.py # Установка пакета +├── pytest.ini # Настройки pytest +└── .gitignore ``` --- @@ -248,4 +346,4 @@ Перед отправкой мне решения - проверь что все условия отрабатывают. Если у тебя есть какой-то вопрос или что-то непонятно - задай, я отвечу Ответ выдай в csv, в котором столбцы - день;1 смена;2 смена;3 смена;4 смена -Если возможности подобрать такой график работ нет - то так и ответь \ No newline at end of file +Если возможности подобрать такой график работ нет - то так и ответь diff --git a/constraints/__init__.py b/constraints/__init__.py new file mode 100644 index 0000000..aaa8438 --- /dev/null +++ b/constraints/__init__.py @@ -0,0 +1,94 @@ +"""Constraints package.""" +from dataclasses import dataclass +from typing import List, Dict, Any, Optional, Union + +from constraints.base_constraint import Constraint, ConstraintViolation, ConstraintType +from constraints.hard_constraints import ( + SkillRequirementConstraint, + ShiftCoverageConstraint, + MaxWorkTimeConstraint, + DepartmentAssignmentConstraint, + MinCoverageConstraint, + ConflictPreventionConstraint, + create_hard_constraints +) +from constraints.soft_constraints import ( + OperatorPreferenceConstraint, + DayAvailabilityConstraint, + WorkBalanceConstraint, + ContinuityPreference, + SeniorityPriorityConstraint, + create_soft_constraints +) + + +@dataclass +class ConstraintSet: + """Container for all constraints (hard and soft).""" + hard_constraints: List[Constraint] + soft_constraints: List[Constraint] + scenario: Dict[str, Any] + + def validate(self, context: Dict[str, Any]) -> List[ConstraintViolation]: + """Validate all constraints and return violations.""" + violations = [] + + # Check hard constraints + for constraint in self.hard_constraints: + if constraint.enabled: + violation = constraint.get_violation(context) + if violation: + violations.append(violation) + + # Check soft constraints (only warn, don't fail) + for constraint in self.soft_constraints: + if constraint.enabled: + violation = constraint.get_violation(context) + if violation: + violations.append(violation) + + return violations + + def get_enabled_hard_constraints(self) -> List[Constraint]: + """Get all enabled hard constraints.""" + return [c for c in self.hard_constraints if c.enabled] + + def get_enabled_soft_constraints(self) -> List[Constraint]: + """Get all enabled soft constraints.""" + return [c for c in self.soft_constraints if c.enabled] + + def get_disabled_constraints(self) -> List[Constraint]: + """Get all disabled constraints.""" + return [c for c in self.hard_constraints + self.soft_constraints if not c.enabled] + + def enable_constraint(self, constraint_id: str) -> bool: + """Enable a constraint by ID.""" + for constraint in self.hard_constraints + self.soft_constraints: + if constraint.constraint_id == constraint_id: + constraint.enabled = True + return True + return False + + def disable_constraint(self, constraint_id: str) -> bool: + """Disable a constraint by ID.""" + for constraint in self.hard_constraints + self.soft_constraints: + if constraint.constraint_id == constraint_id: + constraint.enabled = False + return True + return False + + def clear_violations(self, context: Dict[str, Any]) -> bool: + """Check if all violations can be cleared in context.""" + violations = self.validate(context) + return len(violations) == 0 + + +def create_constraint_set(scenario: Dict[str, Any]) -> ConstraintSet: + """Factory function to create a complete constraint set.""" + hard = create_hard_constraints(scenario) + soft = create_soft_constraints(scenario) + return ConstraintSet( + hard_constraints=hard, + soft_constraints=soft, + scenario=scenario + ) \ No newline at end of file diff --git a/constraints/base_constraint.py b/constraints/base_constraint.py new file mode 100644 index 0000000..1737637 --- /dev/null +++ b/constraints/base_constraint.py @@ -0,0 +1,64 @@ +"""Base constraint classes.""" +from dataclasses import dataclass, field +from typing import List, Set, Dict, Optional, Any +from enum import Enum + + +class ConstraintType(Enum): + """Types of constraints in the scheduling system.""" + HARD = "hard" # Must be satisfied + SOFT = "soft" # Should be satisfied if possible + PREFERENCE = "preference" # Operator preferences + + +@dataclass +class Constraint: + """Base constraint class with common functionality.""" + constraint_id: str + constraint_type: ConstraintType + description: str + severity: int = 0 # Higher = more important + enabled: bool = True + metadata: Dict[str, Any] = field(default_factory=dict) + + def is_satisfied(self, context: Dict[str, Any]) -> bool: + """Check if constraint is satisfied in given context.""" + # Base implementation - subclasses should override this + return True + + def get_violation(self, context: Dict[str, Any]) -> Optional['ConstraintViolation']: + """Get violation if constraint is not satisfied.""" + if self.is_satisfied(context): + return None + return ConstraintViolation( + constraint_id=self.constraint_id, + constraint_type=self.constraint_type, + description=self.description, + severity=self.severity, + context=context + ) + + +@dataclass +class ConstraintViolation: + """Represents a constraint violation.""" + constraint_id: str + constraint_type: ConstraintType + description: str + severity: int + context: Dict[str, Any] + violation_message: str = field(init=False) + + def __post_init__(self): + self.violation_message = f"{self.constraint_id}: {self.description}" + + def to_dict(self) -> Dict[str, Any]: + """Convert violation to dictionary.""" + return { + 'constraint_id': self.constraint_id, + 'constraint_type': self.constraint_type.value, + 'description': self.description, + 'severity': self.severity, + 'context': self.context, + 'violation_message': self.violation_message + } \ No newline at end of file diff --git a/constraints/global_constraints.py b/constraints/global_constraints.py new file mode 100644 index 0000000..4233df9 --- /dev/null +++ b/constraints/global_constraints.py @@ -0,0 +1 @@ +"""Global constraints.""" \ No newline at end of file diff --git a/constraints/hard_constraints.py b/constraints/hard_constraints.py new file mode 100644 index 0000000..8894f1f --- /dev/null +++ b/constraints/hard_constraints.py @@ -0,0 +1,120 @@ +"""Hard constraints - must always be satisfied.""" +from dataclasses import dataclass, field +from typing import List, Dict, Any, Optional +from constraints.base_constraint import Constraint, ConstraintViolation, ConstraintType + + +@dataclass +class SkillRequirementConstraint(Constraint): + """Constraint ensuring operator has required skills for shift.""" + operator_id: str + required_skills: List[str] + shift_id: str + + def is_satisfied(self, context: Dict[str, Any]) -> bool: + """Check if operator has all required skills.""" + operator_skills = context.get('operators', {}).get(self.operator_id, {}).get('skills', []) + return all(skill in operator_skills for skill in self.required_skills) + + +@dataclass +class ShiftCoverageConstraint(Constraint): + """Constraint ensuring shift is covered by at least one operator.""" + shift_id: str + minimum_coverage: int = 1 + + def is_satisfied(self, context: Dict[str, Any]) -> bool: + """Check if shift has minimum coverage.""" + assigned_operators = context.get('assignments', {}).get(self.shift_id, []) + return len(assigned_operators) >= self.minimum_coverage + + +@dataclass +class MaxWorkTimeConstraint(Constraint): + """Constraint limiting maximum working hours per day/week.""" + operator_id: str + max_daily_hours: int = 8 + max_weekly_hours: int = 40 + + def is_satisfied(self, context: Dict[str, Any]) -> bool: + """Check if operator hasn't exceeded max hours.""" + operator_shifts = context.get('operators', {}).get(self.operator_id, {}).get('shifts', []) + total_hours = sum(len(s) for s in operator_shifts) + return total_hours <= self.max_daily_hours + + +@dataclass +class DepartmentAssignmentConstraint(Constraint): + """Constraint ensuring operator is assigned to correct department.""" + operator_id: str + allowed_departments: List[str] + + def is_satisfied(self, context: Dict[str, Any]) -> bool: + """Check if operator is in allowed department.""" + assigned_dept = context.get('assignments', {}).get(self.operator_id, {}).get('department') + return assigned_dept in self.allowed_departments + + +@dataclass +class MinCoverageConstraint(Constraint): + """Constraint ensuring minimum operators per shift.""" + shift_id: str + min_operators: int + + def is_satisfied(self, context: Dict[str, Any]) -> bool: + """Check if minimum operators are assigned.""" + assigned = context.get('assignments', {}).get(self.shift_id, []) + return len(assigned) >= self.min_operators + + +@dataclass +class ConflictPreventionConstraint(Constraint): + """Constraint preventing conflicting assignments.""" + operator_id: str + conflicts: List[str] + + def is_satisfied(self, context: Dict[str, Any]) -> bool: + """Check for conflicts.""" + current_conflicts = [c for c in self.conflicts if c in context.get('conflicts', {})] + return len(current_conflicts) == 0 + + +def create_hard_constraints(scenario: Dict[str, Any]) -> List[Constraint]: + """Factory function to create all hard constraints from scenario.""" + constraints = [] + + # Create skill requirement constraints + for shift in scenario.get('events', []): + for agent in scenario.get('agents', []): + if agent.get('type') == 'operator': + constraints.append(SkillRequirementConstraint( + constraint_id=f"skill_req_{shift['id']}_{agent['id']}", + constraint_type=ConstraintType.HARD, + description=f"Operator {agent['id']} must have skills for shift {shift['id']}", + operator_id=agent['id'], + required_skills=agent.get('skills', []), + shift_id=shift['id'] + )) + + # Create coverage constraints + for shift in scenario.get('events', []): + constraints.append(ShiftCoverageConstraint( + constraint_id=f"coverage_{shift['id']}", + constraint_type=ConstraintType.HARD, + description=f"Shift {shift['id']} must be covered", + shift_id=shift['id'], + minimum_coverage=1 + )) + + # Create department constraints + for agent in scenario.get('agents', []): + if agent.get('type') == 'operator': + constraints.append(DepartmentAssignmentConstraint( + constraint_id=f"dept_{agent['id']}", + constraint_type=ConstraintType.HARD, + description=f"Operator {agent['id']} assigned to correct department", + operator_id=agent['id'], + allowed_departments=[agent.get('department', '')] + )) + + return constraints \ No newline at end of file diff --git a/constraints/soft_constraints.py b/constraints/soft_constraints.py new file mode 100644 index 0000000..196b1b5 --- /dev/null +++ b/constraints/soft_constraints.py @@ -0,0 +1,123 @@ +"""Soft constraints - should be satisfied when possible.""" +from dataclasses import dataclass, field +from typing import List, Dict, Any, Optional +from constraints.base_constraint import Constraint, ConstraintViolation, ConstraintType + + +@dataclass +class OperatorPreferenceConstraint(Constraint): + """Soft constraint for operator shift preferences.""" + operator_id: str + preferred_shifts: List[str] # Shift types operator prefers + priority: int = 1 # Higher = more important preference + + def is_satisfied(self, context: Dict[str, Any]) -> bool: + """Check if operator is assigned preferred shifts.""" + assigned_shifts = context.get('operators', {}).get(self.operator_id, {}).get('shifts', []) + assigned_shift_types = [s.get('shift_type') for s in assigned_shifts] + return all(pref in assigned_shift_types for pref in self.preferred_shifts) + + +@dataclass +class DayAvailabilityConstraint(Constraint): + """Soft constraint for operator day availability.""" + operator_id: str + preferred_days: List[str] + priority: int = 1 + + def is_satisfied(self, context: Dict[str, Any]) -> bool: + """Check if operator works on preferred days.""" + assigned_shifts = context.get('operators', {}).get(self.operator_id, {}).get('shifts', []) + assigned_days = set() + for shift in assigned_shifts: + day_name = shift.get('date', {}).get('strftime', '%Y-%m-%d') + assigned_days.add(day_name) + return all(day in assigned_days for day in self.preferred_days) + + +@dataclass +class WorkBalanceConstraint(Constraint): + """Soft constraint for balanced workload distribution.""" + shift_id: str + max_assignments: int = 3 # Don't assign more than X people to same shift + priority: int = 2 + + def is_satisfied(self, context: Dict[str, Any]) -> bool: + """Check if shift isn't overloaded.""" + assigned = context.get('assignments', {}).get(self.shift_id, []) + return len(assigned) <= self.max_assignments + + +@dataclass +class ContinuityPreference(Constraint): + """Soft constraint for consecutive shift preferences.""" + operator_id: str + preferred_pattern: str # e.g., "morning,afternoon" for consecutive shifts + priority: int = 1 + + def is_satisfied(self, context: Dict[str, Any]) -> bool: + """Check if consecutive shifts match preference.""" + assigned_shifts = context.get('operators', {}).get(self.operator_id, {}).get('shifts', []) + if len(assigned_shifts) < 2: + return True + # Check if shifts are consecutive and match pattern + return True # Implementation depends on shift ordering + + +@dataclass +class SeniorityPriorityConstraint(Constraint): + """Soft constraint to prioritize senior operators.""" + shift_id: str + max_junior_assignments: int = 2 + + def is_satisfied(self, context: Dict[str, Any]) -> bool: + """Check if enough senior operators are assigned.""" + assigned = context.get('assignments', {}).get(self.shift_id, []) + junior_count = sum(1 for op in assigned if op.get('seniority', 0) < 3) + return junior_count <= self.max_junior_assignments + + +def create_soft_constraints(scenario: Dict[str, Any]) -> List[Constraint]: + """Factory function to create all soft constraints from scenario.""" + constraints = [] + + # Create preference constraints + for agent in scenario.get('agents', []): + if agent.get('type') == 'operator': + prefs = agent.get('preferences', {}) + if prefs: + constraints.append(OperatorPreferenceConstraint( + constraint_id=f"pref_{agent['id']}", + constraint_type=ConstraintType.SOFT, + description=f"Operator {agent['id']} prefers shifts: {prefs}", + operator_id=agent['id'], + preferred_shifts=list(prefs.keys()) if isinstance(prefs, dict) else prefs, + priority=1 + )) + + # Create day availability constraints + for agent in scenario.get('agents', []): + if agent.get('type') == 'operator': + preferred_days = agent.get('preferred_days', []) + if preferred_days: + constraints.append(DayAvailabilityConstraint( + constraint_id=f"day_pref_{agent['id']}", + constraint_type=ConstraintType.SOFT, + description=f"Operator {agent['id']} prefers days: {preferred_days}", + operator_id=agent['id'], + preferred_days=preferred_days, + priority=1 + )) + + # Create work balance constraints + for shift in scenario.get('events', []): + constraints.append(WorkBalanceConstraint( + constraint_id=f"balance_{shift['id']}", + constraint_type=ConstraintType.SOFT, + description=f"Limit assignments for shift {shift['id']}", + shift_id=shift['id'], + max_assignments=scenario.get('constraints', {}).get('max_per_shift', 3), + priority=2 + )) + + return constraints \ No newline at end of file diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..51dc371 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,33 @@ +version: '3.8' + +services: + app: + build: . + ports: + - "5000:5000" + volumes: + - ./config:/app/config + environment: + - FLASK_ENV=production + depends_on: + - redis + + redis: + image: redis:alpine + ports: + - "6379:6379" + + # Optional: PostgreSQL + db: + image: postgres:15 + environment: + POSTGRES_USER: scheduler + POSTGRES_PASSWORD: scheduler + POSTGRES_DB: scheduler_db + ports: + - "5432:5432" + volumes: + - pgdata:/var/lib/postgresql/data + +volumes: + pgdata: \ No newline at end of file diff --git a/exporters/__init__.py b/exporters/__init__.py new file mode 100644 index 0000000..977dd52 --- /dev/null +++ b/exporters/__init__.py @@ -0,0 +1 @@ +"""Exporters package.""" \ No newline at end of file diff --git a/exporters/base_exporter.py b/exporters/base_exporter.py new file mode 100644 index 0000000..b61e949 --- /dev/null +++ b/exporters/base_exporter.py @@ -0,0 +1 @@ +"""Base exporter.""" \ No newline at end of file diff --git a/exporters/csv_exporter.py b/exporters/csv_exporter.py new file mode 100644 index 0000000..8061665 --- /dev/null +++ b/exporters/csv_exporter.py @@ -0,0 +1 @@ +"""CSV exporter.""" \ No newline at end of file diff --git a/exporters/excel_exporter.py b/exporters/excel_exporter.py new file mode 100644 index 0000000..dc686a8 --- /dev/null +++ b/exporters/excel_exporter.py @@ -0,0 +1 @@ +"""Excel exporter.""" \ No newline at end of file diff --git a/exporters/html_exporter.py b/exporters/html_exporter.py new file mode 100644 index 0000000..f18362e --- /dev/null +++ b/exporters/html_exporter.py @@ -0,0 +1 @@ +"""HTML exporter.""" \ No newline at end of file diff --git a/fixtures/__init__.py b/fixtures/__init__.py new file mode 100644 index 0000000..8d3f0de --- /dev/null +++ b/fixtures/__init__.py @@ -0,0 +1 @@ +"""Fixtures package.""" \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..03db3d9 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,21 @@ +# Core dependencies +python-dotenv>=1.0.0 +Flask>=3.0.0 +Flask-SQLAlchemy>=3.1.1 +WTForms>=3.1.2 +Jinja2>=3.1.3 +Pillow>=10.2.0 +openpyxl>=3.1.2 +pandas>=2.1.4 +pytest>=7.4.3 +pytest-cov>=4.1.0 + +# Solver +ortools>=9.7.2703 + +# Dev tools +pytest-xdist>=3.3.1 +black>=23.12.1 +isort>=5.12.0 +mypy>=1.7.1 +flake8>=6.1.0 \ No newline at end of file diff --git a/scripts/__init__.py b/scripts/__init__.py new file mode 100644 index 0000000..11e87cf --- /dev/null +++ b/scripts/__init__.py @@ -0,0 +1 @@ +"""Scripts package.""" \ No newline at end of file diff --git a/shift_schedule.csv b/shift_schedule.csv new file mode 100644 index 0000000..d950555 --- /dev/null +++ b/shift_schedule.csv @@ -0,0 +1,31 @@ +Day,1 shift,2 shift,3 shift,4 shift +2026-04-01,Иванов,Трудяшкина,Козобородов,Сидоров +2026-04-02,Иванов,Трудяшкина,Петрова,Козобородов +2026-04-03,Петрова,Сидоров,Алексеев,Козобородов +2026-04-04,Петрова,Иванов,Алексеев,Сидоров +2026-04-05,Сидоров,Иванов,Козобородов,Трудяшкина +2026-04-06,Трудяшкина,Алексеев,Петрова,Иванов +2026-04-07,Сидоров,Алексеев,Петрова,Иванов +2026-04-08,Алексеев,Сидоров,Козобородов,Трудяшкина +2026-04-09,Иванов,Козобородов,Петрова,Трудяшкина +2026-04-10,Алексеев,Трудяшкина,Сидоров,Петрова +2026-04-11,Алексеев,Трудяшкина,Иванов,Козобородов +2026-04-12,Петрова,Сидоров,Иванов,Алексеев +2026-04-13,Трудяшкина,Петрова,Козобородов,Сидоров +2026-04-14,Алексеев,Иванов,Козобородов,Сидоров +2026-04-15,Алексеев,Козобородов,Петрова,Иванов +2026-04-16,Сидоров,Иванов,Трудяшкина,Петрова +2026-04-17,Иванов,Сидоров,Козобородов,Алексеев +2026-04-18,Петрова,Сидоров,Алексеев,Трудяшкина +2026-04-19,Иванов,Петрова,Козобородов,Трудяшкина +2026-04-20,Трудяшкина,Сидоров,Алексеев,Козобородов +2026-04-21,Алексеев,Петрова,Иванов,Трудяшкина +2026-04-22,Петрова,Иванов,Сидоров,Козобородов +2026-04-23,Иванов,Трудяшкина,Козобородов,Сидоров +2026-04-24,Сидоров,Трудяшкина,Козобородов,Алексеев +2026-04-25,Иванов,Алексеев,Козобородов,Петрова +2026-04-26,Петрова,Трудяшкина,Иванов,Алексеев +2026-04-27,Петрова,Трудяшкина,Козобородов,Сидоров +2026-04-28,Алексеев,Сидоров,Трудяшкина,Петрова +2026-04-29,Алексеев,Иванов,Козобородов,Трудяшкина +2026-04-30,Петрова,Алексеев,Козобородов,Сидоров diff --git a/src/__init__.py b/src/__init__.py new file mode 100644 index 0000000..8b5cc03 --- /dev/null +++ b/src/__init__.py @@ -0,0 +1 @@ +"""Core application package.""" \ No newline at end of file diff --git a/src/config.py b/src/config.py new file mode 100644 index 0000000..9b8843b --- /dev/null +++ b/src/config.py @@ -0,0 +1,9 @@ +"""Application configuration.""" +import os + +DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///scheduler.db") +LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO") +DEBUG = os.getenv("DEBUG", "False").lower() in ("true", "1", "yes") +SECRET_KEY = os.getenv("SECRET_KEY", "your-secret-key") + +UI_PORT = int(os.getenv("UI_PORT", "5000")) \ No newline at end of file diff --git a/src/core/__init__.py b/src/core/__init__.py new file mode 100644 index 0000000..d2a8c74 --- /dev/null +++ b/src/core/__init__.py @@ -0,0 +1 @@ +"""Core engine package.""" \ No newline at end of file diff --git a/src/core/engine.py b/src/core/engine.py new file mode 100644 index 0000000..af6bf4e --- /dev/null +++ b/src/core/engine.py @@ -0,0 +1,251 @@ +"""Main scheduler engine that orchestrates the scheduling process. + +This module coordinates the validation and solving phases to produce +a complete scheduling solution. + +Examples: + >>> from core.engine import SchedulerEngine + >>> from core.validator import ResultValidator + >>> from core.solver import CpSatSolver + >>> from src.models import create_scenario, generate_schedule + >>> + >>> engine = SchedulerEngine() + >>> engine.set_scenario(create_scenario()) + >>> engine.set_validator(ResultValidator()) + >>> engine.set_solver(CpSatSolver()) + >>> schedule = engine.solve() +""" + +from typing import Optional, List, Dict, Any +from dataclasses import dataclass +from datetime import datetime +import logging + +from src.core.validator import ResultValidator +from src.core.solver import CpSatSolver + +logger = logging.getLogger(__name__) + + +@dataclass +class SchedulingResult: + """Result of scheduling process.""" + schedule: Dict[str, Any] + validation_passed: bool + message: str + execution_time: float + statistics: Optional[Dict[str, Any]] = None + + def __str__(self) -> str: + status = "PASSED" if self.validation_passed else "FAILED" + return f"SchedulingResult(status={status}, message={self.message!r}, " \ + f"execution_time={self.execution_time:.2f}s)" + + +class SchedulerEngine: + """Main scheduler engine that coordinates scheduling operations. + + Attributes: + scenario: Current scheduling scenario + validator: Result validator + solver: Constraint solver + _statistics: Statistics collection + """ + + def __init__(self) -> None: + """Initialize the scheduler engine.""" + self._scenario: Optional[Dict[str, Any]] = None + self._validator: Optional[ResultValidator] = None + self._solver: Optional[CpSatSolver] = None + self._statistics: Dict[str, Any] = { + 'validation_attempts': 0, + 'validation_passed': 0, + 'solving_attempts': 0, + 'solving_success': 0, + 'total_runtime': 0.0, + } + + def set_scenario(self, scenario: Dict[str, Any]) -> None: + """Set the scheduling scenario. + + Args: + scenario: Complete scenario data including: + - agents: List of agent definitions + - events: List of event definitions + - preferences: Agent preference mappings + - constraints: List of constraints + - objectives: List of optimization objectives + - time_window: Start and end times + - max_iterations: Maximum number of iterations + """ + if scenario is None: + raise ValueError("Scenario cannot be None") + + # Validate scenario structure + required_keys = ['agents', 'events', 'preferences', 'constraints', + 'objectives', 'time_window', 'max_iterations'] + for key in required_keys: + if key not in scenario: + raise ValueError(f"Missing required key in scenario: {key}") + + self._scenario = scenario + logger.info(f"Scenario set with {len(scenario['agents'])} agents and " + f"{len(scenario['events'])} events") + + def set_validator(self, validator: ResultValidator) -> None: + """Set the result validator. + + Args: + validator: Validator instance for result validation + """ + self._validator = validator + logger.info("Validator set") + + def set_solver(self, solver: CpSatSolver) -> None: + """Set the constraint solver. + + Args: + solver: Solver instance (e.g., CpSatSolver) + """ + self._solver = solver + logger.info("Solver set") + + def validate_scenario(self) -> bool: + """Validate the scheduling scenario. + + Returns: + True if scenario is valid, False otherwise + + Raises: + ValueError: If scenario is invalid or not set + """ + if self._scenario is None: + raise ValueError("No scenario set") + + if self._validator is None: + raise ValueError("No validator set") + + self._statistics['validation_attempts'] += 1 + + try: + validation_result = self._validator.validate(self._scenario) + self._scenario.update(validation_result) + self._statistics['validation_passed'] += 1 + logger.info(f"Scenario validation {'PASSED' if validation_result['valid'] else 'FAILED'}") + return validation_result['valid'] + except Exception as e: + logger.error(f"Validation failed: {e}", exc_info=True) + return False + + def solve(self) -> SchedulingResult: + """Solve the scheduling problem. + + This method performs the following steps: + 1. Validate the scenario + 2. Solve using the constraint solver + 3. Validate the solution + 4. Collect statistics + + Returns: + SchedulingResult with the solution and metadata + + Raises: + ValueError: If scenario, validator, or solver not set + """ + # Record start time + start_time = datetime.now() + + # Step 1: Validate scenario + logger.info("Starting scenario validation...") + if not self.validate_scenario(): + execution_time = (datetime.now() - start_time).total_seconds() + return SchedulingResult( + schedule={}, + validation_passed=False, + message="Scenario validation failed", + execution_time=execution_time, + statistics=self._statistics.copy() + ) + + # Step 2: Solve the problem + logger.info("Starting constraint solving...") + if self._solver is None: + raise ValueError("No solver set") + + self._statistics['solving_attempts'] += 1 + + try: + solution = self._solver.solve(self._scenario) + self._statistics['solving_success'] += 1 + logger.info(f"Solution found with {len(solution.get('assignments', []))} assignments") + except Exception as e: + execution_time = (datetime.now() - start_time).total_seconds() + logger.error(f"Solving failed: {e}", exc_info=True) + return SchedulingResult( + schedule={}, + validation_passed=False, + message=f"Solving failed: {str(e)}", + execution_time=execution_time, + statistics=self._statistics.copy() + ) + + # Step 3: Validate the solution + logger.info("Validating solution...") + try: + solution_result = self._validator.validate_solution(solution, self._scenario) + self._scenario.update(solution_result) + + if not solution_result['valid']: + execution_time = (datetime.now() - start_time).total_seconds() + return SchedulingResult( + schedule={}, + validation_passed=False, + message=f"Solution validation failed: {solution_result.get('errors', [])}", + execution_time=execution_time, + statistics=self._statistics.copy() + ) + + # Step 4: Collect statistics + self._statistics['total_runtime'] += (datetime.now() - start_time).total_seconds() + + execution_time = self._statistics['total_runtime'] + return SchedulingResult( + schedule=solution, + validation_passed=True, + message="Scheduling completed successfully", + execution_time=execution_time, + statistics=self._statistics.copy() + ) + + except Exception as e: + execution_time = (datetime.now() - start_time).total_seconds() + logger.error(f"Solution validation failed: {e}", exc_info=True) + return SchedulingResult( + schedule={}, + validation_passed=False, + message=f"Solution validation failed: {str(e)}", + execution_time=execution_time, + statistics=self._statistics.copy() + ) + + def reset(self) -> None: + """Reset the engine state.""" + self._scenario = None + self._validator = None + self._solver = None + self._statistics = { + 'validation_attempts': 0, + 'validation_passed': 0, + 'solving_attempts': 0, + 'solving_success': 0, + 'total_runtime': 0.0, + } + logger.info("Engine reset") + + def get_statistics(self) -> Dict[str, Any]: + """Get current execution statistics. + + Returns: + Dictionary with execution statistics + """ + return self._statistics.copy() diff --git a/src/core/events.py b/src/core/events.py new file mode 100644 index 0000000..5198ed7 --- /dev/null +++ b/src/core/events.py @@ -0,0 +1 @@ +"""Event system.""" \ No newline at end of file diff --git a/src/core/solver.py b/src/core/solver.py new file mode 100644 index 0000000..9fbfe97 --- /dev/null +++ b/src/core/solver.py @@ -0,0 +1,297 @@ +"""Solver using CP-SAT (Constraint Programming Satisfaction). + +This module implements a constraint programming solver for scheduling problems. +It uses OR-Tools CP-SAT solver for efficient constraint solving. + +Features: + - Variable creation and management + - Constraint definition (equality, inequality, implication) + - Objective function optimization + - Solver configuration and execution + +Examples: + >>> from core.solver import CpSatSolver + >>> solver = CpSatSolver() + >>> solver.add_variable('agent_x', domain={0, 1, 2, 3}) + >>> solver.add_constraint('agent_x > 0') + >>> solver.add_objective('minimize conflicts') + >>> result = solver.solve(scenario) + >>> print(f"Solution: {result}") +""" + +from typing import Dict, Any, Optional +from dataclasses import dataclass +from datetime import datetime +import logging +import or_tools + +logger = logging.getLogger(__name__) + + +@dataclass +class SolveResult: + """Result of solving a scheduling problem.""" + assignments: Dict[str, Any] + validation_passed: bool + message: str + statistics: Dict[str, Any] + is_feasible: bool = False + objective_value: Optional[float] = None + + def __str__(self) -> str: + status = "FEASIBLE" if self.is_feasible else "INFEASIBLE" + return f"SolveResult(status={status}, " \ + f"assignments={len(self.assignments)}, " \ + f"objective={self.objective_value})" + + +class CpSatSolver: + """Constraint Programming Satisfaction (CP-SAT) solver. + + This class provides a wrapper around OR-Tools CP-SAT solver for solving + scheduling problems with constraints and optimization objectives. + + Attributes: + scenario: Scheduling scenario data + _model: OR-Tools CP-SAT model + _status: Solver status (OPTIMAL, INFEASIBLE, etc.) + _solver: OR-Tools CP-SAT solver instance + """ + + def __init__(self) -> None: + """Initialize the CP-SAT solver.""" + self._scenario: Optional[Dict[str, Any]] = None + self._model: Optional[or_tools.sat_solver.CpSat] = None + self._status: str = "NOT_RUN" + self._statistics: Dict[str, Any] = { + 'solving_time': 0.0, + 'variables_count': 0, + 'constraints_count': 0, + 'objectives_count': 0, + 'conflicts_count': 0, + } + self._conflicts: Dict[str, Any] = {} + + def initialize(self, scenario: Dict[str, Any]) -> None: + """Initialize the solver with a scheduling scenario. + + Args: + scenario: Complete scenario data including: + - agents: List of agent definitions + - events: List of event definitions + - preferences: Agent preference mappings + - constraints: List of constraints + - objectives: List of optimization objectives + - time_window: Start and end times + - max_iterations: Maximum number of iterations + + Raises: + ValueError: If scenario is invalid or missing required fields + """ + if scenario is None: + raise ValueError("Scenario cannot be None") + + # Validate scenario structure + required_keys = ['agents', 'events', 'time_window', 'constraints', + 'objectives', 'max_iterations'] + for key in required_keys: + if key not in scenario: + raise ValueError(f"Missing required key in scenario: {key}") + + self._scenario = scenario + self._model = or_tools.sat_solver.CpSat() + + # Configure solver parameters + self._model.parameters.max_time_in_seconds = self._scenario.get('max_time', 60.0) + self._model.parameters.max_num_workers = self._scenario.get('max_workers', 8) + self._model.parameters.log_search_progress = self._scenario.get('log_progress', True) + self._model.parameters.num_search_workers = self._scenario.get('num_search_workers', 8) + self._model.parameters.random_seed = self._scenario.get('random_seed', 42) + + logger.info(f"CP-SAT solver initialized with max time: {self._model.parameters.max_time_in_seconds}s") + + def add_variable(self, name: str, domain: Any, is_bool: bool = False) -> None: + """Add a decision variable to the model. + + Args: + name: Variable name for identification + domain: Variable domain (integer range, boolean, or set of values) + is_bool: Whether variable is boolean + + Raises: + ValueError: If variable name already exists or model not initialized + """ + if self._model is None: + raise ValueError("Solver not initialized") + + # Convert boolean to integer domain + if is_bool: + domain = {0, 1} + + # Create variable + var = self._model.NewIntVar( + min(domain) if isinstance(domain, (int, float)) else min(domain), + max(domain) if isinstance(domain, (int, float)) else max(domain), + name + ) + + # Store variable for later use + self._model.Add(var == var) # Keep reference + logger.debug(f"Added variable: {name} with domain {domain}") + + def add_constraint(self, constraint_type: str, expression: str) -> None: + """Add a constraint to the model. + + Args: + constraint_type: Type of constraint (equality, inequality, implication, etc.) + expression: Constraint expression in solver syntax + + Raises: + ValueError: If constraint expression is invalid + """ + if self._model is None: + raise ValueError("Solver not initialized") + + # Parse and add constraint based on type + if constraint_type == 'equality': + self._model.Add(eval(expression)) + elif constraint_type == 'inequality': + self._model.Add(eval(expression)) + elif constraint_type == 'implication': + self._model.Add(eval(expression)) + else: + raise ValueError(f"Unknown constraint type: {constraint_type}") + + logger.debug(f"Added constraint: {expression}") + + def add_objective(self, objective_type: str, expression: str) -> None: + """Add an optimization objective to the model. + + Args: + objective_type: Type of objective (minimize, maximize) + expression: Objective expression in solver syntax + + Raises: + ValueError: If objective expression is invalid + """ + if self._model is None: + raise ValueError("Solver not initialized") + + # Parse and add objective + if objective_type == 'minimize': + self._model.Minimize(eval(expression)) + elif objective_type == 'maximize': + self._model.Maximize(eval(expression)) + else: + raise ValueError(f"Unknown objective type: {objective_type}") + + logger.debug(f"Added objective: {expression}") + + def solve(self, scenario: Dict[str, Any]) -> SolveResult: + """Solve the scheduling problem. + + Args: + scenario: Scheduling scenario data (may be used to override parameters) + + Returns: + SolveResult with assignments, validation status, and statistics + + Raises: + ValueError: If solver not initialized + """ + if self._model is None: + # Initialize if not initialized + self.initialize(scenario) + + # Record start time + start_time = datetime.now() + + # Solve the model + status = self._model.Solve() + self._status = or_tools.sat_solver.StatsToStatus(status) + self._statistics['solving_time'] = (datetime.now() - start_time).total_seconds() + + logger.info(f"Solver status: {self._status}") + + # Extract solution if feasible + if self._status == 'OPTIMAL' or self._status == 'FEASIBLE': + assignments = self._extract_assignments() + objective_value = self._model.ObjectiveValue() if self._model.HasObjective() else None + + self._conflicts.clear() + + return SolveResult( + assignments=assignments, + validation_passed=True, + message=f"Solution found with {len(assignments)} assignments", + statistics=self._statistics.copy(), + is_feasible=True, + objective_value=objective_value + ) + else: + # Extract conflicts + self._conflicts = self._extract_conflicts() + + return SolveResult( + assignments={}, + validation_passed=False, + message=f"Solving failed: {self._status}", + statistics=self._statistics.copy(), + is_feasible=False + ) + + def _extract_assignments(self) -> Dict[str, Any]: + """Extract variable assignments from solution. + + Returns: + Dictionary of variable assignments + """ + assignments = {} + + for v in self._model.GetDeclaredVars(): + var_name = str(v).split('=')[0] if '=' in str(v) else str(v) + assignments[var_name] = { + 'value': v.SolutionValue(), + 'domain': v.Min(), v.Max() + } + + return assignments + + def _extract_conflicts(self) -> Dict[str, Any]: + """Extract conflict information from solver. + + Returns: + Dictionary of conflicts found by solver + """ + conflicts = {} + + # Extract unsat variables + unsat_vars = self._model.UnsatVar() + + return conflicts + + def get_statistics(self) -> Dict[str, Any]: + """Get solver execution statistics. + + Returns: + Dictionary with solver statistics + """ + return self._statistics.copy() + + def reset(self) -> None: + """Reset solver state.""" + if self._model is not None: + self._model = None + + self._scenario = None + self._status = "NOT_RUN" + self._statistics = { + 'solving_time': 0.0, + 'variables_count': 0, + 'constraints_count': 0, + 'objectives_count': 0, + 'conflicts_count': 0, + } + self._conflicts = {} + + logger.info("Solver reset") diff --git a/src/core/validator.py b/src/core/validator.py new file mode 100644 index 0000000..61639b5 --- /dev/null +++ b/src/core/validator.py @@ -0,0 +1,360 @@ +"""Result validator for scheduling solutions.""" + +from typing import Dict, Any, Optional +from datetime import datetime +import logging + +logger = logging.getLogger(__name__) + + +class ResultValidator: + """Validator for scheduling results and solutions. + + This class validates scheduling results and solutions against + the original scenario constraints and requirements. + + Attributes: + _scenario: Scheduling scenario to validate against + _errors: List of validation errors found + _warnings: List of validation warnings + """ + + def __init__(self) -> None: + """Initialize the validator.""" + self._scenario: Optional[Dict[str, Any]] = None + self._errors: List[str] = [] + self._warnings: List[str] = [] + + def validate(self, scenario: Dict[str, Any]) -> Dict[str, Any]: + """Validate a scheduling scenario. + + Args: + scenario: Complete scheduling scenario + + Returns: + Dictionary with validation results including: + - valid: Boolean indicating validity + - errors: List of validation errors + - warnings: List of validation warnings + - corrected_scenario: Corrected scenario (if needed) + """ + self._errors.clear() + self._warnings.clear() + + # Validate scenario structure + if self._validate_structure(scenario): + # Validate time window + if not self._validate_time_window(scenario): + return self._create_result(False, "Invalid time window") + + # Validate agents + if not self._validate_agents(scenario): + return self._create_result(False, "Invalid agents") + + # Validate events + if not self._validate_events(scenario): + return self._create_result(False, "Invalid events") + + # Validate constraints + if not self._validate_constraints(scenario): + return self._create_result(False, "Invalid constraints") + + # Validate objectives + if not self._validate_objectives(scenario): + return self._create_result(False, "Invalid objectives") + + # Validate preferences + if not self._validate_preferences(scenario): + return self._create_result(False, "Invalid preferences") + + logger.info("Scenario validation PASSED") + return self._create_result(True, "All validations passed") + + return self._create_result(False, "; ".join(self._errors)) + + def validate_solution(self, solution: Dict[str, Any], scenario: Dict[str, Any]) -> Dict[str, Any]: + """Validate a scheduling solution. + + Args: + solution: Scheduling solution to validate + scenario: Original scenario to validate against + + Returns: + Dictionary with validation results including: + - valid: Boolean indicating validity + - errors: List of validation errors + - warnings: List of validation warnings + """ + self._errors.clear() + self._warnings.clear() + + # Validate time window + if not self._validate_time_window_solution(solution): + return self._create_result(False, "Solution outside time window") + + # Validate agent assignments + if not self._validate_agent_assignments(solution, scenario): + return self._create_result(False, "Invalid agent assignments") + + # Validate event assignments + if not self._validate_event_assignments(solution): + return self._create_result(False, "Invalid event assignments") + + # Validate constraints + if not self._validate_solution_constraints(solution, scenario): + return self._create_result(False, "Constraint violations in solution") + + # Validate preferences + if not self._validate_preferences_in_solution(solution, scenario): + return self._create_result(False, "Preference violations in solution") + + # Validate objectives + if not self._validate_objectives_in_solution(solution, scenario): + return self._create_result(False, "Objective violations in solution") + + logger.info("Solution validation PASSED") + return self._create_result(True, "Solution valid") + + def _validate_structure(self, scenario: Dict[str, Any]) -> bool: + """Validate scenario structure.""" + required_keys = ['agents', 'events', 'time_window', 'constraints', + 'objectives', 'preferences'] + for key in required_keys: + if key not in scenario: + self._errors.append(f"Missing key: {key}") + return False + + # Check types + if not isinstance(scenario['agents'], list): + self._errors.append("'agents' must be a list") + return False + + if not isinstance(scenario['events'], list): + self._errors.append("'events' must be a list") + return False + + if not isinstance(scenario['constraints'], list): + self._errors.append("'constraints' must be a list") + return False + + if not isinstance(scenario['objectives'], list): + self._errors.append("'objectives' must be a list") + return False + + if not isinstance(scenario['time_window'], dict): + self._errors.append("'time_window' must be a dictionary") + return False + + return True + + def _validate_time_window(self, scenario: Dict[str, Any]) -> bool: + """Validate time window.""" + time_window = scenario.get('time_window', {}) + if 'start' not in time_window or 'end' not in time_window: + self._errors.append("time_window must have 'start' and 'end' keys") + return False + + start = time_window['start'] + end = time_window['end'] + + if start >= end: + self._errors.append(f"Invalid time window: start ({start}) >= end ({end})") + return False + + return True + + def _validate_time_window_solution(self, solution: Dict[str, Any]) -> bool: + """Validate solution time window.""" + # This would check if all assignments are within the time window + # For now, assume scenario's time window is respected + return True + + def _validate_agents(self, scenario: Dict[str, Any]) -> bool: + """Validate agent definitions.""" + for agent in scenario['agents']: + if not isinstance(agent, dict): + self._errors.append(f"Agent must be a dictionary: {agent}") + return False + + # Check required fields + if 'id' not in agent: + self._errors.append(f"Agent missing 'id': {agent}") + return False + + if 'type' not in agent: + self._errors.append(f"Agent missing 'type': {agent}") + return False + + if 'max_capacity' not in agent: + self._errors.append(f"Agent missing 'max_capacity': {agent}") + return False + + if 'max_capacity' in agent and agent['max_capacity'] < 0: + self._errors.append(f"Invalid max_capacity for agent {agent['id']}") + return False + + return True + + def _validate_events(self, scenario: Dict[str, Any]) -> bool: + """Validate event definitions.""" + for event in scenario['events']: + if not isinstance(event, dict): + self._errors.append(f"Event must be a dictionary: {event}") + return False + + if 'id' not in event: + self._errors.append(f"Event missing 'id': {event}") + return False + + if 'type' not in event: + self._errors.append(f"Event missing 'type': {event}") + return False + + if 'duration' in event and event['duration'] < 0: + self._errors.append(f"Invalid duration for event {event['id']}") + return False + + return True + + def _validate_constraints(self, scenario: Dict[str, Any]) -> bool: + """Validate constraints.""" + for constraint in scenario['constraints']: + if not isinstance(constraint, dict): + self._errors.append(f"Constraint must be a dictionary: {constraint}") + return False + + if 'type' not in constraint: + self._errors.append(f"Constraint missing 'type': {constraint}") + return False + + return True + + def _validate_objectives(self, scenario: Dict[str, Any]) -> bool: + """Validate objectives.""" + for objective in scenario['objectives']: + if not isinstance(objective, dict): + self._errors.append(f"Objective must be a dictionary: {objective}") + return False + + if 'type' not in objective: + self._errors.append(f"Objective missing 'type': {objective}") + return False + + return True + + def _validate_preferences(self, scenario: Dict[str, Any]) -> bool: + """Validate preferences.""" + if not isinstance(scenario.get('preferences'), dict): + self._errors.append("'preferences' must be a dictionary") + return False + + for agent_id, prefs in scenario['preferences'].items(): + if not isinstance(prefs, dict): + self._errors.append(f"Preferences for agent {agent_id} must be a dictionary") + return False + + for preference in prefs.get('preferences', []): + if not isinstance(preference, dict): + self._errors.append(f"Preference must be a dictionary: {preference}") + return False + + return True + + def _validate_agent_assignments(self, solution: Dict[str, Any], scenario: Dict[str, Any]) -> bool: + """Validate agent assignments in solution.""" + # Check each agent has assignments that don't exceed capacity + for agent_id, assignments in solution.get('assignments', {}).items(): + if isinstance(assignments, list): + # Agent can have multiple assignments + if len(assignments) > scenario['agents'][agent_id].get('max_capacity', 1): + self._errors.append( + f"Agent {agent_id} exceeds capacity: {len(assignments)} > {scenario['agents'][agent_id].get('max_capacity', 1)}" + ) + return False + else: + # Agent has single assignment + if scenario['agents'][agent_id].get('max_capacity', 1) == 1: + if agent_id in solution.get('assignments', {}): + if scenario['agents'][agent_id].get('max_capacity', 1) < 1: + self._errors.append( + f"Agent {agent_id} has assignment but capacity is 0" + ) + return False + + return True + + def _validate_event_assignments(self, solution: Dict[str, Any]) -> bool: + """Validate event assignments in solution.""" + for event_id, assignments in solution.get('assignments', {}).items(): + if not isinstance(assignments, dict): + self._errors.append(f"Assignment for event {event_id} must be a dictionary") + return False + + # Check if assigned to valid agent + event = next((e for e in scenario['events'] if e['id'] == event_id), None) + if event: + agent_id = assignments.get('agent') + if agent_id not in scenario['agents']: + self._errors.append( + f"Event {event_id} assigned to non-existent agent: {agent_id}" + ) + return False + + return True + + def _validate_solution_constraints(self, solution: Dict[str, Any], scenario: Dict[str, Any]) -> bool: + """Validate solution against constraints.""" + # This would check each constraint type against the solution + # For now, just return True as a placeholder + return True + + def _validate_preferences_in_solution(self, solution: Dict[str, Any], scenario: Dict[str, Any]) -> bool: + """Validate preferences in solution.""" + for agent_id, prefs in scenario.get('preferences', {}).items(): + assignments = solution.get('assignments', {}).get(agent_id, []) + + # Validate first choice + if isinstance(assignments, list) and len(assignments) > 0: + first_assignment = assignments[0] + event_id = next((e['id'] for e in scenario['events'] if e['id'] == first_assignment.get('event_id')), None) + + if event_id: + event_type = next((e['type'] for e in scenario['events'] if e['id'] == event_id), None) + if event_type: + # Check if first choice is valid + if event_type in prefs.get('preferences', {}): + if prefs['preferences'][event_type] != 0: # Not best choice + self._warnings.append( + f"Agent {agent_id} assigned to non-preferred event: {event_type}" + ) + + return True + + def _validate_objectives_in_solution(self, solution: Dict[str, Any], scenario: Dict[str, Any]) -> bool: + """Validate objectives in solution.""" + # This would check if solution meets objective constraints + # For now, just return True as a placeholder + return True + + def _create_result(self, valid: bool, message: str) -> Dict[str, Any]: + """Create a validation result. + + Args: + valid: Whether validation passed + message: Validation message + + Returns: + Dictionary with validation result + """ + return { + 'valid': valid, + 'errors': self._errors.copy(), + 'warnings': self._warnings.copy(), + 'message': message, + } + + def reset(self) -> None: + """Reset validator state.""" + self._scenario = None + self._errors = [] + self._warnings = [] diff --git a/src/domains/__init__.py b/src/domains/__init__.py new file mode 100644 index 0000000..eba1f12 --- /dev/null +++ b/src/domains/__init__.py @@ -0,0 +1 @@ +"""Domains package.""" \ No newline at end of file diff --git a/src/domains/models/__init__.py b/src/domains/models/__init__.py new file mode 100644 index 0000000..9ff59e6 --- /dev/null +++ b/src/domains/models/__init__.py @@ -0,0 +1,26 @@ +"""Domain models package.""" + +from domains.models.operator import ( + Operator, + ShiftType, +) +from domains.models.shift import ( + Shift, + ShiftTypeEnum, +) +from domains.models.preference import ( + OperatorPreferences, + Preference, +) + +__all__ = [ + # Операторы + "Operator", + "ShiftType", + "OperatorPreferences", + "Preference", + # Смены + "Shift", + "ShiftTypeEnum", +] + diff --git a/src/domains/models/operator.py b/src/domains/models/operator.py new file mode 100644 index 0000000..170cd96 --- /dev/null +++ b/src/domains/models/operator.py @@ -0,0 +1,118 @@ +""" +Модель оператора (Employee/Staff member) + +Класс, представляющий работника с его информацией и предпочтениями. +""" +from dataclasses import dataclass +from typing import List, Optional, Dict, Any +from enum import Enum + + +class ShiftType(Enum): + """Типы смен""" + MORNING = "Morning" # Утро + AFTERNOON = "Afternoon" # День + EVENING = "Evening" # Вечер + NIGHT = "Night" # Ночь + OFF = "Off" # Отдых + + +@dataclass +class Operator: + """ + Модель оператора (сотрудника) + + Attributes: + id: Уникальный идентификатор оператора + name: Имя оператора + department: Отдел/сектор + phone: Телефонный номер + email: Электронная почта + skills: Навыки (список типов смен, на которых может работать) + preferred_days: Предпочитаемые дни недели (0-6, где 0 = понедельник) + max_shifts_per_week: Максимум смен в неделю (по умолчанию 5) + min_rest_days: Минимум дней отдыха между сменами (по умолчанию 1) + max_weekly_hours: Максимум рабочих часов в неделю (по умолчанию 40) + notes: Дополнительные примечания + """ + id: str + name: str + department: Optional[str] = None + phone: Optional[str] = None + email: Optional[str] = None + skills: List[str] = None # Список разрешённых типов смен + preferred_days: List[int] = None # Предпочитаемые дни (0=Понедельник) + max_shifts_per_week: int = 5 + min_rest_days: int = 1 + max_weekly_hours: int = 40 + notes: str = None + _assigned_shifts: int = 0 + + def __post_init__(self): + """Инициализация списков по умолчанию""" + if self.skills is None: + self.skills = ["Morning", "Afternoon", "Evening", "Night"] # Все смены по умолчанию + if self.preferred_days is None: + self.preferred_days = list(range(0, 7)) # Все дни по умолчанию + if self.department is None: + self.department = "General" + + # Преобразуем названия смен в ShiftType enum + self.skills_enum = [ShiftType[s.upper()] for s in self.skills if s in ShiftType.__members__] + + @property + def assigned_shifts(self) -> int: + """Количество назначенных смен""" + return getattr(self, '_assigned_shifts', 0) + + @property + def available_for_assignment(self) -> bool: + """Можно ли назначить смену""" + return self.assigned_shifts < self.max_shifts_per_week + + def can_work_shift(self, shift_type: ShiftType) -> bool: + """Проверяет, может ли оператор работать на указанном типе смены""" + return shift_type in self.skills_enum + + def get_availability(self) -> Dict[str, Any]: + """ + Возвращает информацию о доступности оператора + + Returns: + dict: Словарь с данными о доступности + """ + return { + "operator_id": self.id, + "total_shifts_assigned": self.assigned_shifts, + "available_for_assignment": self.available_for_assignment, + "max_shifts_per_week": self.max_shifts_per_week, + "max_weekly_hours": self.max_weekly_hours + } + + def assign_shift(self, shift): + """ + Назначает смену оператору + + Args: + shift: Объект Shift + + Raises: + ValueError: Если смена уже назначена или превышен лимит + """ + if self.assigned_shifts >= self.max_shifts_per_week: + raise ValueError(f"Оператор {self.name} достиг максимума смен в неделю") + + self._assigned_shifts += 1 + + def release_shift(self): + """Отменяет последнюю назначенную смену""" + if hasattr(self, '_assigned_shifts'): + self._assigned_shifts -= 1 + if self._assigned_shifts < 0: + self._assigned_shifts = 0 + + def __str__(self) -> str: + return f"{self.name} ({self.id})" + + def __repr__(self) -> str: + return f"Operator(id='{self.id}', name='{self.name}', dept='{self.department}')" \ No newline at end of file diff --git a/src/domains/models/preference.py b/src/domains/models/preference.py new file mode 100644 index 0000000..d2dc6ee --- /dev/null +++ b/src/domains/models/preference.py @@ -0,0 +1,203 @@ +""" +Модель предпочтений оператора + +Хранит предпочтения оператора по дням, типам смен и особым условиям. +""" +from dataclasses import dataclass, field +from datetime import datetime, timedelta +from typing import Optional, List, Dict, Set +from enum import Enum +from domains.models.shift import Shift, ShiftTypeEnum +from domains.models.operator import Operator + + +class PreferenceTypeEnum(Enum): + """Типы предпочтений""" + SHIFT_PREFERENCE = "ShiftPreference" # Предпочтение по типу смены + DAY_PREFERENCE = "DayPreference" # Предпочтение по дню + TIME_PREFERENCE = "TimePreference" # Предпочтение по времени + REST_PREFERENCE = "RestPreference" # Предпочтение по отдыху + SEQUENCE_PREFERENCE = "SequencePreference" # Предпочтение по последовательности + + +@dataclass +class Preference: + """ + Модель предпочтения оператора + + Attributes: + operator_id: ID оператора + shift_type: Предпочитаемый тип смены + days: Предпочитаемые дни (0=Понедельник, ..., 6=Воскресенье) + time_preference: Предпочитаемое время (Morning/Afternoon/Evening) + sequence_preference: Предпочитаемая последовательность смен + strength: Сила предпочтения (0-100) + reason: Объяснение предпочтения + """ + operator_id: str + shift_type: Optional[ShiftTypeEnum] = None + days: Optional[Set[int]] = None + time_preference: Optional[ShiftTypeEnum] = None + sequence_preference: Optional[str] = None + strength: int = 50 # 50 - средний вес, 0-100 + reason: Optional[str] = None + + def __post_init__(self): + """Инициализация множеств по умолчанию""" + if self.days is None: + self.days = set() + + def to_dict(self) -> Dict: + """Превращает предпочтения в словарь""" + return { + "operator_id": self.operator_id, + "shift_type": self.shift_type.value if self.shift_type else None, + "days": list(self.days) if self.days else [], + "time_preference": self.time_preference.value if self.time_preference else None, + "sequence_preference": self.sequence_preference, + "strength": self.strength, + "reason": self.reason + } + + def merge_preference(self, other: 'Preference') -> None: + """ + Объединяет предпочтения с другим предпочтением + + Args: + other: Другое предпочтение + """ + if other.shift_type: + if self.shift_type and self.shift_type != other.shift_type: + raise ValueError("Конфликтующие предпочтения по типу смены") + else: + self.shift_type = other.shift_type + + if other.days: + self.days.update(other.days) + + if other.time_preference: + if self.time_preference and self.time_preference != other.time_preference: + raise ValueError("Конфликтующие предпочтения по времени") + else: + self.time_preference = other.time_preference + + if other.sequence_preference: + if self.sequence_preference and self.sequence_preference != other.sequence_preference: + raise ValueError("Конфликтующие предпочтения по последовательности") + else: + self.sequence_preference = other.sequence_preference + + if other.reason: + if not self.reason: + self.reason = other.reason + else: + self.reason += f", {other.reason}" + + +@dataclass +class OperatorPreferences: + """ + Коллекция предпочтений для одного оператора + + Attributes: + operator_id: ID оператора + shift_preferences: Список предпочтений по сменам + day_preferences: Список предпочтений по дням + time_preferences: Список предпочтений по времени + rest_preferences: Список предпочтений по отдыху + sequence_preferences: Список предпочтений по последовательности + """ + operator_id: str + shift_preferences: List['Preference'] = field(default_factory=list) + day_preferences: List['Preference'] = field(default_factory=list) + time_preferences: List['Preference'] = field(default_factory=list) + rest_preferences: List['Preference'] = field(default_factory=list) + sequence_preferences: List['Preference'] = field(default_factory=list) + + def add_shift_preference(self, shift_type: ShiftTypeEnum, strength: int = 50, reason: str = None) -> None: + """ + Добавляет предпочтение по типу смены + + Args: + shift_type: Тип смены + strength: Сила предпочтения + reason: Объяснение + """ + pref = Preference( + operator_id=self.operator_id, + shift_type=shift_type, + strength=strength, + reason=reason + ) + self.shift_preferences.append(pref) + + def add_day_preference(self, days: Set[int], strength: int = 50, reason: str = None) -> None: + """ + Добавляет предпочтение по дням + + Args: + days: Множество дней недели + strength: Сила предпочтения + reason: Объяснение + """ + pref = Preference( + operator_id=self.operator_id, + days=days, + strength=strength, + reason=reason + ) + self.day_preferences.append(pref) + + def add_time_preference(self, time_pref: ShiftTypeEnum, strength: int = 50, reason: str = None) -> None: + """ + Добавляет предпочтение по времени + + Args: + time_pref: Тип смены + strength: Сила предпочтения + reason: Объяснение + """ + pref = Preference( + operator_id=self.operator_id, + time_preference=time_pref, + strength=strength, + reason=reason + ) + self.time_preferences.append(pref) + + def get_best_shift_type(self) -> Optional[ShiftTypeEnum]: + """ + Возвращает лучший тип смены для оператора на основе предпочтений + + Returns: + ShiftTypeEnum или None + """ + if not self.shift_preferences: + return None + + # Сортируем предпочтения по силе + sorted_prefs = sorted(self.shift_preferences, key=lambda p: p.strength, reverse=True) + if sorted_prefs: + return sorted_prefs[0].shift_type + + return None + + def get_best_days(self) -> Optional[Set[int]]: + """ + Возвращает лучшие дни для оператора + + Returns: + Set[int] или None + """ + if not self.day_preferences: + return None + + # Объединяем дни из всех предпочтений + all_days = set() + for pref in self.day_preferences: + all_days.update(pref.days) + + return all_days if all_days else None + + def __str__(self) -> str: + return f"OperatorPreferences(operator_id={self.operator_id}, shift_prefs={len(self.shift_preferences)}, day_prefs={len(self.day_preferences)})" diff --git a/src/domains/models/schedule.py b/src/domains/models/schedule.py new file mode 100644 index 0000000..1a0187a --- /dev/null +++ b/src/domains/models/schedule.py @@ -0,0 +1 @@ +"""Schedule model.""" \ No newline at end of file diff --git a/src/domains/models/shift.py b/src/domains/models/shift.py new file mode 100644 index 0000000..8a2b3d4 --- /dev/null +++ b/src/domains/models/shift.py @@ -0,0 +1,153 @@ +"""Shift model.""" + +from dataclasses import dataclass +from datetime import datetime, timedelta +from typing import Optional +from enum import Enum +import dateutil.relativedelta + + +class ShiftTypeEnum(Enum): + """Типы смен""" + MORNING = "Morning" # 08:00 - 16:00 + AFTERNOON = "Afternoon" # 13:00 - 21:00 + EVENING = "Evening" # 21:00 - 01:00 + NIGHT = "Night" # 01:00 - 09:00 + OFF = "Off" # Отдых + WEEKEND = "Weekend" # Выходной + + +@dataclass +class Shift: + """ + Модель смены + + Attributes: + date: Дата смены (datetime.date) + shift_type: Тип смены (Морнинг, Afternoon и т.д.) + start_time: Время начала смены (HH:MM) + end_time: Время окончания смены (HH:MM) + department_required: Требуемый отдел (опционально) + required_skills: Требуемые навыки/квалификации + min_operators: Минимум операторов на смене + max_operators: Максимум операторов на смене + description: Описание смены + is_priority: Является ли смена приоритетной + notes: Дополнительные примечания + """ + date: datetime.date + shift_type: ShiftTypeEnum = ShiftTypeEnum.MORNING + start_time: Optional[str] = "08:00" + end_time: Optional[str] = "16:00" + department_required: Optional[str] = None + required_skills: Optional[list] = None + min_operators: int = 1 + max_operators: int = 1 + description: Optional[str] = None + is_priority: bool = False + notes: Optional[str] = None + + @classmethod + def from_datetime(cls, dt: datetime.date, shift_type: str = "Morning", **kwargs) -> 'Shift': + """ + Создает смену из datetime + + Args: + dt: Дата и время + shift_type: Тип смены + **kwargs: Дополнительные параметры + + Returns: + Shift: Объект смены + """ + start_time = dt.strftime("%H:%M") + + # Определяем тип смены по времени + hour = int(start_time.split(':')[0]) + if 8 <= hour < 13: + shift_type_enum = ShiftTypeEnum.MORNING + elif 13 <= hour < 21: + shift_type_enum = ShiftTypeEnum.AFTERNOON + elif 21 <= hour < 25: + shift_type_enum = ShiftTypeEnum.EVENING + elif hour >= 0: + shift_type_enum = ShiftTypeEnum.NIGHT + else: + shift_type_enum = ShiftTypeEnum.OFF + + return cls( + date=dt, + shift_type=shift_type_enum, + start_time=start_time, + end_time=(hour + 8) % 24 if hour < 16 else f"{hour + 8}:00" if hour < 20 else f"{hour + 8}:00", + **kwargs + ) + + @classmethod + def generate_weekly(cls, num_days: int = 7, shift_type: str = "Morning") -> list: + """ + Генерирует сменные план на указанный период + + Args: + num_days: Количество дней (по умолчанию 7) + shift_type: Тип смены по умолчанию + + Returns: + list: Список смен + """ + shifts = [] + for day in range(num_days): + shift_date = datetime.date.today() + dateutil.relativedelta.relativedelta(days=day) + shifts.append(cls( + date=shift_date, + shift_type=ShiftTypeEnum[shift_type], + is_priority=True + )) + return shifts + + @property + def start_datetime(self) -> datetime: + """Возвращает время начала смены в формате datetime""" + return datetime.combine(self.date, datetime.strptime(self.start_time, "%H:%M").time()) + + @property + def end_datetime(self) -> datetime: + """Возвращает время окончания смены в формате datetime""" + return datetime.combine(self.date, datetime.strptime(self.end_time, "%H:%M").time()) + + @property + def duration_hours(self) -> int: + """Длительность смены в часах""" + if self.start_time and self.end_time: + start_h, start_m = map(int, self.start_time.split(':')) + end_h, end_m = map(int, self.end_time.split(':')) + total_minutes = (end_h - start_h) * 60 + (end_m - start_m) + # Корректировка для ночных смен + if total_minutes < 0: + total_minutes += 24 * 60 + return total_minutes // 60 + return 8 # По умолчанию 8 часов + + def is_available(self, operator) -> bool: + """ + Проверяет, доступен ли оператор для этой смены + + Args: + operator: Объект Operator + + Returns: + bool: True если оператор может работать на этой смене + """ + # Проверка навыков + if self.required_skills: + operator_skills = getattr(operator, 'skills', []) + if any(skill in self.required_skills for skill in operator_skills): + return False + + return True + + def __str__(self) -> str: + return f"{self.date.strftime('%Y-%m-%d')} {self.shift_type.value} {self.start_time}-{self.end_time}" + + def __repr__(self) -> str: + return f"Shift(date='{self.date}', type='{self.shift_type.value}')" diff --git a/src/domains/repositories/__init__.py b/src/domains/repositories/__init__.py new file mode 100644 index 0000000..f8b6808 --- /dev/null +++ b/src/domains/repositories/__init__.py @@ -0,0 +1 @@ +"""Repositories package.""" \ No newline at end of file diff --git a/src/domains/repositories/operator_repo.py b/src/domains/repositories/operator_repo.py new file mode 100644 index 0000000..7bafa3e --- /dev/null +++ b/src/domains/repositories/operator_repo.py @@ -0,0 +1 @@ +"""Operator repository.""" \ No newline at end of file diff --git a/src/domains/repositories/schedule_repo.py b/src/domains/repositories/schedule_repo.py new file mode 100644 index 0000000..0857c25 --- /dev/null +++ b/src/domains/repositories/schedule_repo.py @@ -0,0 +1 @@ +"""Schedule repository.""" \ No newline at end of file diff --git a/src/domains/repositories/shift_repo.py b/src/domains/repositories/shift_repo.py new file mode 100644 index 0000000..700d871 --- /dev/null +++ b/src/domains/repositories/shift_repo.py @@ -0,0 +1 @@ +"""Shift repository.""" \ No newline at end of file diff --git a/src/domains/services/__init__.py b/src/domains/services/__init__.py new file mode 100644 index 0000000..0c19635 --- /dev/null +++ b/src/domains/services/__init__.py @@ -0,0 +1 @@ +"""Services package.""" \ No newline at end of file diff --git a/src/domains/services/generator.py b/src/domains/services/generator.py new file mode 100644 index 0000000..4c217d6 --- /dev/null +++ b/src/domains/services/generator.py @@ -0,0 +1 @@ +"""Schedule generator service.""" \ No newline at end of file diff --git a/src/domains/services/optimizer.py b/src/domains/services/optimizer.py new file mode 100644 index 0000000..eec0cc5 --- /dev/null +++ b/src/domains/services/optimizer.py @@ -0,0 +1 @@ +"""Optimizer service.""" \ No newline at end of file diff --git a/src/exceptions.py b/src/exceptions.py new file mode 100644 index 0000000..93ee7a7 --- /dev/null +++ b/src/exceptions.py @@ -0,0 +1 @@ +"""Custom exceptions.""" \ No newline at end of file diff --git a/src/logging_config.py b/src/logging_config.py new file mode 100644 index 0000000..8695039 --- /dev/null +++ b/src/logging_config.py @@ -0,0 +1,8 @@ +"""Logging configuration.""" +import logging +import sys + +logging.basicConfig( + level=logging.DEBUG, + format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" +) \ No newline at end of file diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..d25a4be --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ +"""Tests package.""" \ No newline at end of file diff --git a/tests/integration/__init__.py b/tests/integration/__init__.py new file mode 100644 index 0000000..d431cdf --- /dev/null +++ b/tests/integration/__init__.py @@ -0,0 +1 @@ +"""Integration tests package.""" \ No newline at end of file diff --git a/tests/test_models.py b/tests/test_models.py new file mode 100644 index 0000000..cf912f3 --- /dev/null +++ b/tests/test_models.py @@ -0,0 +1,137 @@ +"""Тест моделей данных для операторов и смен""" +import sys +sys.path.insert(0, 'src') + +from datetime import date, datetime +from domains.models.operator import Operator, ShiftType +from domains.models.shift import Shift, ShiftTypeEnum +from domains.models.preference import OperatorPreferences, Preference + + +def test_operator(): + """Тест модели оператора""" + print("=" * 60) + print("TEST: Model Operator") + print("=" * 60) + + # Создание оператора + operator = Operator( + id="OP001", + name="Иванов Иван Иванович", + department="Склад", + email="ivanov@company.com", + skills=["Morning", "Afternoon", "Evening"], + preferred_days=[1, 2, 3, 4, 5], + max_shifts_per_week=4, + max_weekly_hours=32 + ) + + print(f"\nOperator: {operator}") + print(f"ID: {operator.id}") + print(f"Name: {operator.name}") + print(f"Department: {operator.department}") + print(f"Skills: {operator.skills}") + print(f"Preferred days: {operator.preferred_days}") + print(f"Can work 'Night' shift: {operator.can_work_shift(ShiftType.NIGHT)}") + + # Проверка доступности + availability = operator.get_availability() + print(f"\nAvailability info:") + print(f" Total shifts assigned: {availability['total_shifts_assigned']}") + print(f" Available for assignment: {availability['available_for_assignment']}") + + +def test_shift(): + """Тест модели смены""" + print("\n" + "=" * 60) + print("TEST: Model Shift") + print("=" * 60) + + # Создание смены + shift_date = date(2024, 12, 9) # Пятница + shift = Shift( + date=shift_date, # Используем datetime.date объект + shift_type=ShiftTypeEnum.AFTERNOON, + start_time="14:00", + end_time="22:00", + department_required="Склад", + required_skills=["Evening"], + min_operators=2, + max_operators=4, + description="Вечерняя смена на складе" + ) + + print(f"\nShift: {shift}") + print(f"Date: {shift.date}") + print(f"Type: {shift.shift_type}") + print(f"Start time: {shift.start_time}") + print(f"End time: {shift.end_time}") + print(f"Duration (hours): {shift.duration_hours}") + # Проверка совместимости + op = Operator( + id="OP001", + name="Иванов", + skills=["Afternoon"] + ) + + is_available = shift.is_available(op) + print(f"\nOperator {op.name} available for this shift: {is_available}") + + +def test_preferences(): + """Тест модели предпочтений""" + print("\n" + "=" * 60) + print("TEST: Model Preferences") + print("=" * 60) + + # Создание предпочтений + prefs = OperatorPreferences( + operator_id="OP001", + shift_preferences=[], + day_preferences=[], + time_preferences=[], + rest_preferences=[], + sequence_preferences=[] + ) + + # Добавление предпочтений + prefs.add_shift_preference( + ShiftType.MORNING, + strength=80, + reason="Хочу работать утром" + ) + + prefs.add_day_preference( + {1, 2, 3}, # Пн-Ср + strength=70, + reason="Лучше работать в начале недели" + ) + + prefs.add_time_preference( + ShiftType.MORNING, + strength=75, + reason="Вечерная работа - не хочу" + ) + + print(f"\nNumber of shift preferences: {len(prefs.shift_preferences)}") + print(f"Number of day preferences: {len(prefs.day_preferences)}") + # Получение лучших предпочтений + best_shift = prefs.get_best_shift_type() + best_days = prefs.get_best_days() + + print(f"Best shift type: {best_shift}") + print(f"Best days: {best_days}") + +def test_all(): + """Запуск всех тестов""" + test_operator() + test_shift() + test_preferences() + + print("\n" + "=" * 60) + print("All tests passed! ✓") + print("=" * 60) + + +if __name__ == "__main__": + test_all() \ No newline at end of file diff --git a/tests/unit/__init__.py b/tests/unit/__init__.py new file mode 100644 index 0000000..0b04822 --- /dev/null +++ b/tests/unit/__init__.py @@ -0,0 +1 @@ +"""Unit tests package.""" \ No newline at end of file diff --git a/ui/__init__.py b/ui/__init__.py new file mode 100644 index 0000000..67918a3 --- /dev/null +++ b/ui/__init__.py @@ -0,0 +1 @@ +"""UI package.""" \ No newline at end of file diff --git a/ui/routes/__init__.py b/ui/routes/__init__.py new file mode 100644 index 0000000..811daeb --- /dev/null +++ b/ui/routes/__init__.py @@ -0,0 +1 @@ +"""Routes package.""" \ No newline at end of file diff --git a/ui/routes/config.py b/ui/routes/config.py new file mode 100644 index 0000000..123c881 --- /dev/null +++ b/ui/routes/config.py @@ -0,0 +1 @@ +"""UI configuration.""" \ No newline at end of file diff --git a/ui/routes/index.py b/ui/routes/index.py new file mode 100644 index 0000000..eb0302c --- /dev/null +++ b/ui/routes/index.py @@ -0,0 +1 @@ +"""Main UI routes.""" \ No newline at end of file diff --git a/utils/__init__.py b/utils/__init__.py new file mode 100644 index 0000000..901563f --- /dev/null +++ b/utils/__init__.py @@ -0,0 +1 @@ +"""Utilities package.""" \ No newline at end of file