Compare commits
3 Commits
very_hard_
...
18813a3ccb
| Author | SHA1 | Date | |
|---|---|---|---|
|
18813a3ccb | ||
|
|
5d2b665fa6 | ||
|
|
a8f47d29f6 |
5
.continue/rules/python.md
Normal file
5
.continue/rules/python.md
Normal file
@@ -0,0 +1,5 @@
|
||||
---
|
||||
description: A description of your rule
|
||||
---
|
||||
Ты senior python-разработчик
|
||||
У тебя 5-лет опыта работы с библиотекой ortools
|
||||
42
.gitignore
vendored
42
.gitignore
vendored
@@ -1,42 +0,0 @@
|
||||
# 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
|
||||
136
PLAN.md
Normal file
136
PLAN.md
Normal file
@@ -0,0 +1,136 @@
|
||||
# 📋 ПЛАН РАБОТ: Модульность конфигурации
|
||||
|
||||
## Цель
|
||||
Перенести все константы (настройки) в отдельный файл `config.py` для удобства редактирования человеком, не знакомым с Python.
|
||||
|
||||
---
|
||||
|
||||
## Этап 1: Создание файла конфигурации
|
||||
|
||||
**Файл:** `config.py`
|
||||
|
||||
**Что перенести:**
|
||||
- **Настройки дня**
|
||||
- Год (`year = 2026`)
|
||||
- Месяц (`month = 4`)
|
||||
|
||||
- **Смены (`SHIFTS`)**
|
||||
- С 06:00 до 14:30 (смена 0)
|
||||
- С 08:00 до 16:30 (смена 1)
|
||||
- С 11:30 до 20:00 (смена 2)
|
||||
- С 12:30 до 21:00 (смена 3)
|
||||
|
||||
- **Операторы (`OPERATORS`)**
|
||||
- Иванов
|
||||
- Алексеев
|
||||
- Сидоров
|
||||
- Козобородов
|
||||
- Трудяшкина
|
||||
- Петрова
|
||||
|
||||
- **Предпочтения операторов (`OPERATOR_PREFERENCES`)**
|
||||
- Иванов: 10, 13 апреля
|
||||
- Алексеев: понедельники
|
||||
- Сидоров: 25, 26 апреля
|
||||
- Козобородов: смена 2 (11:30–20:00)
|
||||
- Трудяшкина: 4 апреля
|
||||
- Петрова: 8, 29 апреля
|
||||
|
||||
**Как редактировать:**
|
||||
1. Откройте `config.py` в текстовом редакторе
|
||||
2. Изменяйте значения в квадратных скобках или кавычках
|
||||
3. Для добавления/удаления дней используйте списки `[25, 26]`
|
||||
4. Сохраните файл
|
||||
|
||||
---
|
||||
|
||||
## Этап 2: Обновление `models.py`
|
||||
|
||||
**Что делать:**
|
||||
- Удалить все константы из `models.py`
|
||||
- Добавить импорт: `from config import *`
|
||||
- Удалить определения `SHIFTS`, `OPERATORS`, `OPERATOR_PREFERENCES`
|
||||
|
||||
---
|
||||
|
||||
## Этап 3: Обновление `main.py`
|
||||
|
||||
**Что делать:**
|
||||
- Удалить встроенную генерацию дат
|
||||
- Использовать импорт из `config.py`
|
||||
- Добавить вывод в CSV
|
||||
|
||||
---
|
||||
|
||||
## Этап 4: Создание класса предпочтений операторов
|
||||
|
||||
**Файл:** `models.py` (или отдельный файл `preferences.py`)
|
||||
|
||||
**Что сделать:**
|
||||
- Создать класс `OperatorPreferences` для конструирования и управления предпочтениями операторов.
|
||||
- Реализовать цепочку методов (builder pattern) for удобного создания предпочтений.
|
||||
- Поддержать следующие типы предпочтений:
|
||||
- Желаемый рабочий день (конкретные даты месяца)
|
||||
- Желаемая смена (индекс смены 0-3)
|
||||
- Желаемый день недели для работы (понедельник, вторник, ...)
|
||||
- Желаемый выходной (конкретные даты месяца)
|
||||
- Реализовать метод `__str__` для отладки.
|
||||
- Обеспечить экспорт предпочтений в формат, пригодный для `config.py`.
|
||||
|
||||
**Требования к реализации:**
|
||||
- Класс должен позволять конструировать предпочтения через цепочку методов.
|
||||
- Поддерживать три типа предпочтений (как указано выше).
|
||||
- Экспортировать предпочтения в формат, пригодный для `config.py`.
|
||||
- Поддерживать метод `__str__` для отладки.
|
||||
|
||||
---
|
||||
|
||||
## Этап 5: Создание документации
|
||||
|
||||
**Файл:** `USER_GUIDE.md`
|
||||
|
||||
**Что включить:**
|
||||
- Как изменить даты
|
||||
- Как добавить/удалить операторов
|
||||
- Как изменить смены
|
||||
- Как добавить новые предпочтения
|
||||
|
||||
---
|
||||
|
||||
## Этап 6: Тестирование
|
||||
|
||||
**Проверить:**
|
||||
1. Запуск `python main.py`
|
||||
2. Корректное считывание дат из `config.py`
|
||||
3. Генерация расписания
|
||||
4. Вывод в CSV
|
||||
5. Корректная работа класса `OperatorPreferences`
|
||||
6. Экспорт предпочтений в `config.py`
|
||||
|
||||
---
|
||||
|
||||
## Этап 7: Написание тестов
|
||||
|
||||
**Файл:** `tests/test_all.py`
|
||||
|
||||
**Что сделать:**
|
||||
- Создать тесты для всех модулей проекта
|
||||
- Использовать pytest для запуска тестов
|
||||
- Проверить корректность работы всех компонентов
|
||||
- Добавить тесты для edge-кейсов
|
||||
|
||||
**Структура тестов:**
|
||||
- `test_config.py` - тесты для конфигурации
|
||||
- `test_models.py` - тесты для моделей и предпочтений
|
||||
- `test_main.py` - тесты для основного скрипта
|
||||
- `test_integration.py` - интеграционные тесты
|
||||
|
||||
---
|
||||
|
||||
## Результат
|
||||
- Конфигурация в одном файле `config.py`
|
||||
- Простое редактирование без знания Python
|
||||
- Четкая документация в `USER_GUIDE.md`
|
||||
- Гибкая система предпочтений через класс `OperatorPreferences`
|
||||
- Полноценный набор тестов для всех компонентов
|
||||
|
||||
108
README.md
108
README.md
@@ -7,108 +7,10 @@
|
||||
## 📁 Структура проекта
|
||||
|
||||
```
|
||||
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
|
||||
├── main.py # Основная логика с генерацией графика
|
||||
├── shift_schedule.csv # Результат (заполненный график)
|
||||
├── README.md # Документация
|
||||
└── .continue/ # Система контроля версий
|
||||
```
|
||||
|
||||
---
|
||||
@@ -346,4 +248,4 @@ OperatorSchedule/
|
||||
Перед отправкой мне решения - проверь что все условия отрабатывают.
|
||||
Если у тебя есть какой-то вопрос или что-то непонятно - задай, я отвечу
|
||||
Ответ выдай в csv, в котором столбцы - день;1 смена;2 смена;3 смена;4 смена
|
||||
Если возможности подобрать такой график работ нет - то так и ответь
|
||||
Если возможности подобрать такой график работ нет - то так и ответь
|
||||
60
config.py
Normal file
60
config.py
Normal file
@@ -0,0 +1,60 @@
|
||||
"""
|
||||
config.py - Настройки графика смен
|
||||
Редактируйте этот файл БЕЗ знания Python!
|
||||
"""
|
||||
|
||||
# 📅 НАСТРОЙКИ ДАТ
|
||||
YEAR = 2026 # Год
|
||||
MONTH = 4 # Месяц (апрель)
|
||||
|
||||
# 🕒 СМЕНЫ (время начала, время окончания)
|
||||
# format: (начало_часа, окончание_часа)
|
||||
SHIFTS = [
|
||||
(6, 14.5), # Смена 0: 06:00 - 14:30
|
||||
(8, 16.5), # Смена 1: 08:00 - 16:30
|
||||
(11.5, 20), # Смена 2: 11:30 - 20:00
|
||||
(12.5, 21), # Смена 3: 12:30 - 21:00
|
||||
]
|
||||
|
||||
# 👷 ИМЕНА ОПЕРАТОРОВ
|
||||
# Меняйте здесь имена операторов!
|
||||
OPERATORS = [
|
||||
"Иванов",
|
||||
"Алексеев",
|
||||
"Сидоров",
|
||||
"Козобородов",
|
||||
"Трудяшкина",
|
||||
"Петрова",
|
||||
]
|
||||
|
||||
# 💖 ПРЕДПОЧТЕНИЯ ОПЕРАТОРОВ
|
||||
# Формат: {имя_оператора: объект_предпочтений}
|
||||
# Для создания предпочтений используйте класс OperatorPreferences из models.py
|
||||
# Пример:
|
||||
# from models import OperatorPreferences, DayOfWeek
|
||||
# prefs = OperatorPreferences.create("Иванов") \
|
||||
# .work_on_date(10) \
|
||||
# .work_on_date(13) \
|
||||
# .build()
|
||||
# OPERATOR_PREFERENCES = {"Иванов": prefs}
|
||||
|
||||
# Для удобства, здесь приведены примеры создания предпочтений через словарь,
|
||||
# который будет автоматически преобразован в объект OperatorPreferences в models.py.
|
||||
# Это позволяет редактировать файл без знания Python, используя простые структуры данных.
|
||||
|
||||
OPERATOR_PREFERENCES = {
|
||||
"Иванов": {"dates": [10, 13]}, # 10 и 13 апреля
|
||||
"Алексеев": {"day_of_week": "понедельник"}, # только понедельники
|
||||
"Сидоров": {"dates": [25, 26]}, # 25 и 26 апреля
|
||||
"Козобородов": {"shift": 2}, # только смена 2 (11:30–20:00)
|
||||
"Трудяшкина": {"dates": [4]}, # 4 апреля
|
||||
"Петрова": {"dates": [8, 29]}, # 8 и 29 апреля
|
||||
}
|
||||
|
||||
# 📝 ПОЯСНЕНИЯ ДЛЯ РЕДАКТИРОВАНИЯ:
|
||||
# - Дни месяца указываются в списке: {"dates": [10, 13]} = 10 и 13 апреля
|
||||
# - "понедельник" (или "monday") означает работать только в понедельник
|
||||
# - Число в "shift": 0-3 указывает на номер смены (0 = первая смена)
|
||||
# - Для добавления/удаления дней меняйте списки в квадратных скобках
|
||||
# - Для изменения имени оператора меняйте строку в OPERATORS
|
||||
# - Для установки дня, когда оператор НЕ работает, используйте: {"off_dates": [5]}
|
||||
@@ -1,94 +0,0 @@
|
||||
"""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
|
||||
)
|
||||
@@ -1,64 +0,0 @@
|
||||
"""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
|
||||
}
|
||||
@@ -1 +0,0 @@
|
||||
"""Global constraints."""
|
||||
@@ -1,120 +0,0 @@
|
||||
"""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
|
||||
@@ -1,123 +0,0 @@
|
||||
"""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
|
||||
@@ -1,33 +0,0 @@
|
||||
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:
|
||||
@@ -1 +0,0 @@
|
||||
"""Exporters package."""
|
||||
@@ -1 +0,0 @@
|
||||
"""Base exporter."""
|
||||
@@ -1 +0,0 @@
|
||||
"""CSV exporter."""
|
||||
@@ -1 +0,0 @@
|
||||
"""Excel exporter."""
|
||||
@@ -1 +0,0 @@
|
||||
"""HTML exporter."""
|
||||
@@ -1 +0,0 @@
|
||||
"""Fixtures package."""
|
||||
160
main.py
160
main.py
@@ -1,101 +1,81 @@
|
||||
from ortools.sat.python import cp_model
|
||||
|
||||
"""
|
||||
main.py - Основной скрипт для создания графика смен
|
||||
Использует модуль models.py для создания и решения модели
|
||||
"""
|
||||
from models import create_model, solve_model, get_solution
|
||||
from config import YEAR, MONTH, SHIFTS, OPERATORS
|
||||
import csv
|
||||
|
||||
# Определяем даты апреля 2026 года
|
||||
days = [
|
||||
"2026-04-{:02d}".format(i) for i in range(1, 31)
|
||||
]
|
||||
|
||||
# Смены
|
||||
shifts = [
|
||||
(6, 14.5), # с 06:00 до 14:30
|
||||
(8, 16.5), # с 08:00 до 16:30
|
||||
(11.5, 20), # с 11:30 до 20:00
|
||||
(12.5, 21) # с 12:30 до 21:00
|
||||
]
|
||||
|
||||
# Операторы и их пожелания
|
||||
operators = [
|
||||
"Иванов",
|
||||
"Алексеев",
|
||||
"Сидоров",
|
||||
"Козобородов",
|
||||
"Трудяшкина",
|
||||
"Петрова"
|
||||
]
|
||||
|
||||
operator_preferences = {
|
||||
"Иванов": ["2026-04-10", "2026-04-13"],
|
||||
"Алексеев": [(i, 0) for i in days if (int(i[-2:]) - 1) % 7 == 0], # Понедельники
|
||||
"Сидоров": ["2026-04-25", "2026-04-26"],
|
||||
"Козобородов": [(i, 2) for i in days], # Предпочитает смену 11:30–20:00
|
||||
"Трудяшкина": ["2026-04-04"],
|
||||
"Петрова": ["2026-04-08", "2026-04-29"]
|
||||
}
|
||||
|
||||
# Создаем модель
|
||||
model = cp_model.CpModel()
|
||||
|
||||
# Переменные: определяем, работает ли оператор на смене в день
|
||||
works = {}
|
||||
for day in days:
|
||||
for shift_idx in range(len(shifts)):
|
||||
for op in operators:
|
||||
works[(day, shift_idx, op)] = model.NewBoolVar(f"{op}_{day}_shift_{shift_idx}")
|
||||
|
||||
# # Пожелания операторов
|
||||
for op, prefs in operator_preferences.items():
|
||||
if isinstance(prefs, list):
|
||||
for pref in prefs:
|
||||
if isinstance(pref, tuple):
|
||||
day, shift = pref
|
||||
model.AddHint(works[day, shift, op], 1)
|
||||
else:
|
||||
day = pref
|
||||
model.Add(sum(works[day, shift_idx, op] for shift_idx in range(len(shifts))) == 0)
|
||||
|
||||
# Условие 1: На каждую смену должен быть назначен оператор
|
||||
for day in days:
|
||||
for shift_idx in range(len(shifts)):
|
||||
model.Add(sum(works[day, shift_idx, op] for op in operators) == 1)
|
||||
|
||||
# Условие 2: У каждого оператора должно быть 8 выходных
|
||||
for op in operators:
|
||||
model.Add(sum(works[day, shift_idx, op] for day in days for shift_idx in range(len(shifts))) == 20 )
|
||||
|
||||
# Условие 3: Оператор не должен работать больше 5 дней подряд
|
||||
for op in operators:
|
||||
for day_idx in range(len(days) - 4):
|
||||
model.Add(sum(works[days[day_idx + i], shift_idx, op] for i in range(5) for shift_idx in range(len(shifts))) <= 4)
|
||||
|
||||
# Условие 4: В день оператор может работать только одну смену
|
||||
for day in days:
|
||||
for op in operators:
|
||||
model.Add(sum(works[day, shift_idx, op] for shift_idx in range(len(shifts))) <= 1)
|
||||
|
||||
# Решаем модель
|
||||
solver = cp_model.CpSolver()
|
||||
status = solver.Solve(model)
|
||||
|
||||
if status != cp_model.OPTIMAL and status != cp_model.FEASIBLE:
|
||||
print("Невозможно составить график")
|
||||
else:
|
||||
# Создаем CSV файл с результатами
|
||||
with open('shift_schedule.csv', 'w', newline='', encoding='utf-8') as file:
|
||||
def generate_csv_report(schedule: dict, filename: str = 'shift_schedule.csv'):
|
||||
"""
|
||||
Генерирует CSV-отчет на основе расписания.
|
||||
|
||||
Аргументы:
|
||||
schedule: словарь расписания из get_solution()
|
||||
filename: имя файла для вывода CSV
|
||||
|
||||
Возвращает:
|
||||
True если файл создан успешно
|
||||
"""
|
||||
with open(filename, 'w', newline='', encoding='utf-8') as file:
|
||||
writer = csv.writer(file)
|
||||
writer.writerow(['Day'] + [f'{i+1} shift' for i in range(len(shifts))])
|
||||
# Заголовок
|
||||
writer.writerow(['День'] + [f'Смена {i+1}' for i in range(len(SHIFTS))])
|
||||
|
||||
for day in days:
|
||||
for day, day_shifts in schedule.items():
|
||||
row = [day]
|
||||
for shift_idx in range(len(shifts)):
|
||||
for shift_idx in range(len(SHIFTS)):
|
||||
op = None
|
||||
for op_name in operators:
|
||||
if solver.Value(works[day, shift_idx, op_name]) == 1:
|
||||
for op_name in OPERATORS:
|
||||
if schedule[day][shift_idx].get(op_name, False):
|
||||
op = op_name
|
||||
break
|
||||
row.append(op)
|
||||
writer.writerow(row)
|
||||
|
||||
print(f"График успешно составлен и сохранен в '{filename}'")
|
||||
return True
|
||||
|
||||
print("График успешно составлен и сохранен в 'shift_schedule.csv'")
|
||||
|
||||
def main():
|
||||
"""
|
||||
Главная функция: создает и решает модель, затем генерирует CSV-отчет.
|
||||
"""
|
||||
print("=" * 60)
|
||||
print("Создание графика смен")
|
||||
print("=" * 60)
|
||||
|
||||
# Шаг 1: Создаем модель
|
||||
print("\n[1] Создание модели. ..")
|
||||
model, variables, operators_set = create_model()
|
||||
print(f" Модель создана с {len(variables)} переменными")
|
||||
print(f" Операторы: {', '.join(operators_set)}")
|
||||
|
||||
# Шаг 2: Решаем модель
|
||||
print("\n[2] Решение модели. ..")
|
||||
success = solve_model()
|
||||
|
||||
if success:
|
||||
print(" ✅ Модель решена успешно!")
|
||||
|
||||
# Шаг 3: Получаем решение
|
||||
print("\n[3] Получение решения. ..")
|
||||
schedule = get_solution()
|
||||
print(f" Решение получено для {len(schedule)} дней")
|
||||
|
||||
# Шаг 4: Генерируем CSV-отчет
|
||||
print("\n[4] Генерация CSV-отчета. ..")
|
||||
generate_csv_report(schedule, 'shift_schedule.csv')
|
||||
|
||||
else:
|
||||
print(" ❌ Не удалось составить график!")
|
||||
print(" Проверьте ограничения и предпочтения операторов.")
|
||||
return False
|
||||
|
||||
print("\n" + "=" * 60)
|
||||
print("Готово!")
|
||||
print("=" * 60)
|
||||
|
||||
return True
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
293
models.py
Normal file
293
models.py
Normal file
@@ -0,0 +1,293 @@
|
||||
from __future__ import annotations
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime
|
||||
from enum import Enum
|
||||
from typing import List, Dict, Set, Union, Optional
|
||||
from ortools.sat.python import cp_model
|
||||
from config import YEAR, MONTH, SHIFTS, OPERATORS
|
||||
|
||||
class PreferenceType(Enum):
|
||||
"""Типы предпочтений"""
|
||||
WORK_ON_DATE = "work_on_date"
|
||||
WORK_ON_DAY_OF_WEEK = "work_on_day_of_week"
|
||||
WORK_ONLY_SHIFT = "work_only_shift"
|
||||
WORK_ONLY_DATE = "work_only_date"
|
||||
OFF_ON_DATE = "off_on_date"
|
||||
|
||||
class DayOfWeek(Enum):
|
||||
"""Дни недели"""
|
||||
MONDAY = 0
|
||||
TUESDAY = 1
|
||||
WEDNESDAY = 2
|
||||
THURSDAY = 3
|
||||
FRIDAY = 4
|
||||
SATURDAY = 5
|
||||
SUNDAY = 6
|
||||
|
||||
@dataclass
|
||||
class OperatorPreferences:
|
||||
"""
|
||||
Класс для управления предпочтениями оператора.
|
||||
Использует паттерн Builder для удобного создания.
|
||||
"""
|
||||
operator_name: str
|
||||
work_on_dates: List[int] = field(default_factory=list) # Дни месяца (1-31)
|
||||
work_on_day_of_week: List[DayOfWeek] = field(default_factory=list)
|
||||
work_only_shift: Optional[int] = None # Индекс смены (0-3)
|
||||
off_on_dates: List[int] = field(default_factory=list) # Дни, когда оператор НЕ работает
|
||||
|
||||
def __post_init__(self):
|
||||
if self.work_only_shift is not None and not (0 <= self.work_only_shift < len(SHIFTS)):
|
||||
raise ValueError(f"Индекс смены {self.work_only_shift} вне диапазона 0-{len(SHIFTS)-1}")
|
||||
|
||||
@classmethod
|
||||
def create(cls, operator_name: str) -> 'PreferencesBuilder':
|
||||
"""Статический метод для начала построения предпочтений"""
|
||||
return PreferencesBuilder(operator_name)
|
||||
|
||||
def to_dict(self) -> Dict[str, Union[int, str, List[int]]]:
|
||||
"""
|
||||
Конвертирует объект предпочтений в формат, пригодный для config.py.
|
||||
Возвращает словарь с ключами, понятными для парсера в create_model.
|
||||
"""
|
||||
result = {}
|
||||
|
||||
if self.work_only_shift is not None:
|
||||
result['shift'] = self.work_only_shift
|
||||
|
||||
if self.work_on_day_of_week:
|
||||
day_names = [d.name.lower() for d in self.work_on_day_of_week]
|
||||
# Если только один день недели, можно хранить как строку, иначе как список
|
||||
if len(day_names) == 1:
|
||||
result['day_of_week'] = day_names[0]
|
||||
else:
|
||||
result['day_of_week'] = day_names
|
||||
|
||||
if self.work_on_dates:
|
||||
result['dates'] = self.work_on_dates
|
||||
|
||||
if self.off_on_dates:
|
||||
result['off_dates'] = self.off_on_dates
|
||||
|
||||
return result
|
||||
|
||||
def __str__(self) -> str:
|
||||
"""Метод для отладки"""
|
||||
parts = []
|
||||
if self.work_only_shift is not None:
|
||||
parts.append(f"Shift {self.work_only_shift}")
|
||||
if self.work_on_day_of_week:
|
||||
parts.append(f"Days: {', '.join([d.name for d in self.work_on_day_of_week])}")
|
||||
if self.work_on_dates:
|
||||
parts.append(f"Dates: {self.work_on_dates}")
|
||||
if self.off_on_dates:
|
||||
parts.append(f"Off: {self.off_on_dates}")
|
||||
|
||||
return f"OperatorPreferences({self.operator_name}: {', '.join(parts)})"
|
||||
|
||||
class PreferencesBuilder:
|
||||
"""Builder для создания предпочтений оператора"""
|
||||
|
||||
def __init__(self, operator_name: str):
|
||||
self.operator_name = operator_name
|
||||
self.prefs = OperatorPreferences(operator_name)
|
||||
|
||||
def work_on_date(self, day: int) -> 'PreferencesBuilder':
|
||||
"""Добавить день месяца для работы"""
|
||||
if not 1 <= day <= 31:
|
||||
raise ValueError(f"День {day} вне диапазона 1-31")
|
||||
self.prefs.work_on_dates.append(day)
|
||||
return self
|
||||
|
||||
def work_on_dates(self, days: List[int]) -> 'PreferencesBuilder':
|
||||
"""Добавить список дней месяца для работы"""
|
||||
for day in days:
|
||||
self.work_on_date(day)
|
||||
return self
|
||||
|
||||
def work_on_day_of_week(self, day: DayOfWeek) -> 'PreferencesBuilder':
|
||||
"""Добавить день недели для работы"""
|
||||
if day not in self.prefs.work_on_day_of_week:
|
||||
self.prefs.work_on_day_of_week.append(day)
|
||||
return self
|
||||
|
||||
def work_on_days_of_week(self, days: List[DayOfWeek]) -> 'PreferencesBuilder':
|
||||
"""Добавить список дней недели для работы"""
|
||||
for day in days:
|
||||
self.work_on_day_of_week(day)
|
||||
return self
|
||||
|
||||
def work_only_shift(self, shift_idx: int) -> 'PreferencesBuilder':
|
||||
"""Установить предпочтительную смену (работает только в этой смене)"""
|
||||
self.prefs.work_only_shift = shift_idx
|
||||
return self
|
||||
|
||||
def off_on_date(self, day: int) -> 'PreferencesBuilder':
|
||||
"""Установить день, когда оператор НЕ работает"""
|
||||
if not 1 <= day <= 31:
|
||||
raise ValueError(f"День {day} вне диапазона 1-31")
|
||||
self.prefs.off_on_dates.append(day)
|
||||
return self
|
||||
|
||||
def off_on_dates(self, days: List[int]) -> 'PreferencesBuilder':
|
||||
"""Установить дни, когда оператор НЕ работает"""
|
||||
for day in days:
|
||||
self.off_on_date(day)
|
||||
return self
|
||||
|
||||
def build(self) -> OperatorPreferences:
|
||||
"""Завершить построение и вернуть объект предпочтений"""
|
||||
return self.prefs
|
||||
|
||||
def get_days() -> List[str]:
|
||||
"""Генерирует список дат месяца"""
|
||||
return [f"{YEAR}-{MONTH:02d}-{i:02d}" for i in range(1, 31)]
|
||||
|
||||
def create_model() -> Tuple[cp_model.CpModel, Dict, Set]:
|
||||
"""
|
||||
Создаёт и настраивает CP-модель.
|
||||
|
||||
Возвращает:
|
||||
model: CP-модель
|
||||
variables: словарь переменных работы
|
||||
operators_set: множество операторов для удобства
|
||||
"""
|
||||
model = cp_model.CpModel()
|
||||
variables = {}
|
||||
days = get_days()
|
||||
|
||||
# Определяем переменные: works[(day, shift_idx, op)]
|
||||
for day in days:
|
||||
for shift_idx in range(len(SHIFTS)):
|
||||
for op in OPERATORS:
|
||||
var_name = f"{op}_{day}_shift_{shift_idx}"
|
||||
variables[(day, shift_idx, op)] = model.NewBoolVar(var_name)
|
||||
|
||||
# Определяем множество операторов
|
||||
operators_set = set(OPERATORS)
|
||||
|
||||
# 1. Учёт предпочтений операторов
|
||||
# Ожидаем, что OPERATOR_PREFERENCES теперь содержит объекты OperatorPreferences или их словари
|
||||
for op, prefs_data in OPERATOR_PREFERENCES.items():
|
||||
# Если это словарь (из config.py), преобразуем в объект
|
||||
if isinstance(prefs_data, dict):
|
||||
prefs = OperatorPreferences(op)
|
||||
if 'shift' in prefs_data:
|
||||
prefs.work_only_shift = prefs_data['shift']
|
||||
if 'day_of_week' in prefs_data:
|
||||
dow = prefs_data['day_of_week']
|
||||
if isinstance(dow, str):
|
||||
try:
|
||||
prefs.work_on_day_of_week.append(DayOfWeek[dow.upper()])
|
||||
except KeyError:
|
||||
pass
|
||||
elif isinstance(dow, list):
|
||||
for d in dow:
|
||||
try:
|
||||
prefs.work_on_day_of_week.append(DayOfWeek[d.upper()])
|
||||
except KeyError:
|
||||
pass
|
||||
if 'dates' in prefs_data:
|
||||
prefs.work_on_dates = prefs_data['dates']
|
||||
if 'off_dates' in prefs_data:
|
||||
prefs.off_on_dates = prefs_data['off_dates']
|
||||
else:
|
||||
prefs = prefs_data
|
||||
|
||||
# Применяем предпочтения
|
||||
if prefs.work_only_shift is not None:
|
||||
# Оператор предпочитает конкретную смену
|
||||
shift_idx = prefs.work_only_shift
|
||||
for day in days:
|
||||
model.AddHint(variables[(day, shift_idx, op)], 1)
|
||||
|
||||
if prefs.work_on_day_of_week:
|
||||
# Оператор работает только в определенные дни недели
|
||||
for day in days:
|
||||
date_obj = datetime(int(day.split('-')[0]), int(day.split('-')[1]), int(day.split('-')[2]))
|
||||
if date_obj.weekday() in [d.value for d in prefs.work_on_day_of_week]:
|
||||
# Если есть preference_shift, то только в эту смену
|
||||
if prefs.work_only_shift is not None:
|
||||
model.AddHint(variables[(day, prefs.work_only_shift, op)], 1)
|
||||
else:
|
||||
# Иначе в любую смену (но это уже обрабатывается другими ограничениями)
|
||||
# Для hint можно добавить в первую смену как ориентир
|
||||
model.AddHint(variables[(day, 0, op)], 1)
|
||||
|
||||
if prefs.work_on_dates:
|
||||
# Оператор предпочитает конкретные дни
|
||||
for pref_day_num in prefs.work_on_dates:
|
||||
target_day = f"{YEAR}-{MONTH:02d}-{pref_day_num:02d}"
|
||||
if target_day in days:
|
||||
if prefs.work_only_shift is not None:
|
||||
model.AddHint(variables[(target_day, prefs.work_only_shift, op)], 1)
|
||||
else:
|
||||
model.AddHint(variables[(target_day, 0, op)], 1)
|
||||
|
||||
if prefs.off_on_dates:
|
||||
# Оператор НЕ работает в эти дни
|
||||
for off_day_num in prefs.off_on_dates:
|
||||
target_day = f"{YEAR}-{MONTH:02d}-{off_day_num:02d}"
|
||||
if target_day in days:
|
||||
# Запрещаем работу в этот день
|
||||
for shift_idx in range(len(SHIFTS)):
|
||||
model.Add(variables[(target_day, shift_idx, op)] == 0)
|
||||
|
||||
# Условие 1: На каждую смену должен быть назначен оператор
|
||||
for day in days:
|
||||
for shift_idx in range(len(SHIFTS)):
|
||||
model.Add(sum(variables[(day, shift_idx, op)] for op in OPERATORS) == 1)
|
||||
|
||||
# Условие 2: У каждого оператора должно быть 8 выходных (20 рабочих дней)
|
||||
for op in OPERATORS:
|
||||
model.Add(sum(variables[(day, shift_idx, op)] for day in days for shift_idx in range(len(SHIFTS))) == 20)
|
||||
|
||||
# Условие 3: Оператор не должен работать больше 5 дней подряд
|
||||
for op in OPERATORS:
|
||||
for day_idx in range(len(days) - 4):
|
||||
model.Add(sum(variables[(days[day_idx + i], shift_idx, op)]
|
||||
for i in range(5)
|
||||
for shift_idx in range(len(SHIFTS))) <= 4)
|
||||
|
||||
# Условие 4: В день оператор может работать только одну смену
|
||||
for day in days:
|
||||
for op in OPERATORS:
|
||||
model.Add(sum(variables[(day, shift_idx, op)] for shift_idx in range(len(SHIFTS))) <= 1)
|
||||
|
||||
return model, variables, operators_set
|
||||
|
||||
def solve_model() -> bool:
|
||||
"""
|
||||
Решает модель.
|
||||
|
||||
Возвращает:
|
||||
True - если решение найдено
|
||||
False - если не удалось составить график
|
||||
"""
|
||||
model, variables, operators_set = create_model()
|
||||
|
||||
solver = cp_model.CpSolver()
|
||||
status = solver.Solve(model)
|
||||
|
||||
return status in (cp_model.OPTIMAL, cp_model.FEASIBLE)
|
||||
|
||||
def get_solution() -> Dict:
|
||||
"""
|
||||
Получает решение модели.
|
||||
|
||||
Возвращает:
|
||||
Словарь с расписанием: {день: {смена: {оператор: True/False}}}
|
||||
"""
|
||||
model, variables, operators_set = create_model()
|
||||
solver = cp_model.CpSolver()
|
||||
solver.Solve(model)
|
||||
|
||||
schedule = {}
|
||||
for day in get_days():
|
||||
schedule[day] = {
|
||||
shift_idx: {op: solver.Value(variables[(day, shift_idx, op)])
|
||||
for op in OPERATORS}
|
||||
for shift_idx in range(len(SHIFTS))
|
||||
}
|
||||
|
||||
return schedule
|
||||
@@ -1,21 +0,0 @@
|
||||
# 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
|
||||
@@ -1 +0,0 @@
|
||||
"""Scripts package."""
|
||||
@@ -1,31 +0,0 @@
|
||||
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,Петрова,Алексеев,Козобородов,Сидоров
|
||||
|
@@ -1 +0,0 @@
|
||||
"""Core application package."""
|
||||
@@ -1,9 +0,0 @@
|
||||
"""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"))
|
||||
@@ -1 +0,0 @@
|
||||
"""Core engine package."""
|
||||
@@ -1,251 +0,0 @@
|
||||
"""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()
|
||||
@@ -1 +0,0 @@
|
||||
"""Event system."""
|
||||
@@ -1,297 +0,0 @@
|
||||
"""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")
|
||||
@@ -1,360 +0,0 @@
|
||||
"""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 = []
|
||||
@@ -1 +0,0 @@
|
||||
"""Domains package."""
|
||||
@@ -1,26 +0,0 @@
|
||||
"""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",
|
||||
]
|
||||
|
||||
@@ -1,118 +0,0 @@
|
||||
"""
|
||||
Модель оператора (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}')"
|
||||
@@ -1,203 +0,0 @@
|
||||
"""
|
||||
Модель предпочтений оператора
|
||||
|
||||
Хранит предпочтения оператора по дням, типам смен и особым условиям.
|
||||
"""
|
||||
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)})"
|
||||
@@ -1 +0,0 @@
|
||||
"""Schedule model."""
|
||||
@@ -1,153 +0,0 @@
|
||||
"""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}')"
|
||||
@@ -1 +0,0 @@
|
||||
"""Repositories package."""
|
||||
@@ -1 +0,0 @@
|
||||
"""Operator repository."""
|
||||
@@ -1 +0,0 @@
|
||||
"""Schedule repository."""
|
||||
@@ -1 +0,0 @@
|
||||
"""Shift repository."""
|
||||
@@ -1 +0,0 @@
|
||||
"""Services package."""
|
||||
@@ -1 +0,0 @@
|
||||
"""Schedule generator service."""
|
||||
@@ -1 +0,0 @@
|
||||
"""Optimizer service."""
|
||||
@@ -1 +0,0 @@
|
||||
"""Custom exceptions."""
|
||||
@@ -1,8 +0,0 @@
|
||||
"""Logging configuration."""
|
||||
import logging
|
||||
import sys
|
||||
|
||||
logging.basicConfig(
|
||||
level=logging.DEBUG,
|
||||
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
|
||||
)
|
||||
@@ -1 +1 @@
|
||||
"""Tests package."""
|
||||
# Тесты для проекта модульности конфигурации
|
||||
@@ -1 +0,0 @@
|
||||
"""Integration tests package."""
|
||||
140
tests/test_config.py
Normal file
140
tests/test_config.py
Normal file
@@ -0,0 +1,140 @@
|
||||
"""
|
||||
test_config.py - Тесты для конфигурации
|
||||
Проверяют корректность чтения и структуры config.py
|
||||
"""
|
||||
import pytest
|
||||
from config import YEAR, MONTH, SHIFTS, OPERATORS, OPERATOR_PREFERENCES
|
||||
|
||||
|
||||
class TestConfigConstants:
|
||||
"""Тесты для констант конфигурации"""
|
||||
|
||||
def test_year_is_int(self):
|
||||
"""Год должен быть целым числом"""
|
||||
assert isinstance(YEAR, int)
|
||||
assert YEAR == 2026
|
||||
|
||||
def test_month_is_int(self):
|
||||
"""Месяц должен быть целым числом"""
|
||||
assert isinstance(MONTH, int)
|
||||
assert MONTH == 4
|
||||
|
||||
def test_month_in_range(self):
|
||||
"""Месяц должен быть в диапазоне 1-12"""
|
||||
assert 1 <= MONTH <= 12
|
||||
|
||||
def test_shifts_is_list(self):
|
||||
"""Смены должны быть списком"""
|
||||
assert isinstance(SHIFTS, list)
|
||||
assert len(SHIFTS) == 4
|
||||
|
||||
def test_shifts_structure(self):
|
||||
"""Каждая смена должна быть кортежем из двух чисел"""
|
||||
for shift in SHIFTS:
|
||||
assert isinstance(shift, tuple)
|
||||
assert len(shift) == 2
|
||||
start, end = shift
|
||||
assert isinstance(start, (int, float))
|
||||
assert isinstance(end, (int, float))
|
||||
assert start < end
|
||||
|
||||
def test_operators_is_list(self):
|
||||
"""Операторы должны быть списком"""
|
||||
assert isinstance(OPERATORS, list)
|
||||
assert len(OPERATORS) > 0
|
||||
|
||||
def test_operators_not_empty(self):
|
||||
"""Список операторов не должен быть пустым"""
|
||||
assert len(OPERATORS) >= 1
|
||||
|
||||
def test_operators_strings(self):
|
||||
"""Все операторы должны быть строками"""
|
||||
for op in OPERATORS:
|
||||
assert isinstance(op, str)
|
||||
assert len(op) > 0
|
||||
|
||||
def test_operator_preferences_structure(self):
|
||||
"""Структура предпочтений должна быть корректной"""
|
||||
assert isinstance(OPERATOR_PREFERENCES, dict)
|
||||
assert len(OPERATOR_PREFERENCES) == len(OPERATORS)
|
||||
|
||||
def test_operator_preferences_values(self):
|
||||
"""Значения предпочтений должны быть словарями"""
|
||||
for op, prefs in OPERATOR_PREFERENCES.items():
|
||||
assert isinstance(prefs, dict)
|
||||
# Проверка, что ключи предпочтений корректны
|
||||
valid_keys = {'dates', 'day_of_week', 'shift', 'off_dates'}
|
||||
assert set(prefs.keys()).issubset(valid_keys)
|
||||
|
||||
|
||||
class TestConfigDates:
|
||||
"""Тесты для дат"""
|
||||
|
||||
def test_year_is_positive(self):
|
||||
"""Год должен быть положительным"""
|
||||
assert YEAR > 0
|
||||
|
||||
def test_month_is_valid(self):
|
||||
"""Месяц должен быть валидным (1-12)"""
|
||||
assert 1 <= MONTH <= 12
|
||||
|
||||
def test_shifts_time_valid(self):
|
||||
"""Время смены должно быть валидным (0-24)"""
|
||||
for start, end in SHIFTS:
|
||||
assert 0 <= start <= 24
|
||||
assert 0 <= end <= 24
|
||||
assert start < end
|
||||
|
||||
|
||||
class TestConfigOperators:
|
||||
"""Тесты для операторов"""
|
||||
|
||||
def test_operators_unique(self):
|
||||
"""Имена операторов должны быть уникальными"""
|
||||
assert len(OPERATORS) == len(set(OPERATORS))
|
||||
|
||||
def test_operators_no_empty_strings(self):
|
||||
"""Имена операторов не должны быть пустыми строками"""
|
||||
for op in OPERATORS:
|
||||
assert op.strip() != ""
|
||||
|
||||
def test_operators_no_special_chars(self):
|
||||
"""Имена операторов не должны содержать специальные символы"""
|
||||
for op in OPERATORS:
|
||||
assert not any(c in op for c in ['<', '>', '/', '\\', '|', ':', '*', '?', '"'])
|
||||
|
||||
|
||||
class TestConfigPreferences:
|
||||
"""Тесты для предпочтений операторов"""
|
||||
|
||||
def test_all_operators_have_preferences(self):
|
||||
"""Все операторы должны иметь предпочтения"""
|
||||
assert len(OPERATOR_PREFERENCES) == len(OPERATORS)
|
||||
|
||||
def test_dates_in_range(self):
|
||||
"""Дни в предпочтениях должны быть в диапазоне 1-31"""
|
||||
for prefs in OPERATOR_PREFERENCES.values():
|
||||
if 'dates' in prefs:
|
||||
for day in prefs['dates']:
|
||||
assert 1 <= day <= 31
|
||||
|
||||
def test_shift_in_range(self):
|
||||
"""Смена в предпочтениях должна быть в диапазоне 0-3"""
|
||||
for prefs in OPERATOR_PREFERENCES.values():
|
||||
if 'shift' in prefs:
|
||||
assert 0 <= prefs['shift'] <= 3
|
||||
|
||||
def test_day_of_week_valid(self):
|
||||
"""День недели в предпочтениях должен быть валидным"""
|
||||
for prefs in OPERATOR_PREFERENCES.values():
|
||||
if 'day_of_week' in prefs:
|
||||
day_str = prefs['day_of_week']
|
||||
# Проверяем, что это строка с валидным значением
|
||||
assert day_str.lower() in ['понедельник', 'вторник', 'среда', 'четверг', 'пятница', 'суббота', 'воскресенье']
|
||||
|
||||
def test_off_dates_in_range(self):
|
||||
"""Дни отдыха должны быть в диапазоне 1-31"""
|
||||
for prefs in OPERATOR_PREFERENCES.values():
|
||||
if 'off_dates' in prefs:
|
||||
for day in prefs['off_dates']:
|
||||
assert 1 <= day <= 31
|
||||
@@ -1,137 +1,354 @@
|
||||
"""Тест моделей данных для операторов и смен"""
|
||||
import sys
|
||||
sys.path.insert(0, 'src')
|
||||
"""
|
||||
test_models.py - Тесты для моделей и предпочтений
|
||||
Проверяют корректность работы класса OperatorPreferences и связанных функций
|
||||
"""
|
||||
import pytest
|
||||
from datetime import datetime
|
||||
from models import (
|
||||
OperatorPreferences,
|
||||
PreferencesBuilder,
|
||||
DayOfWeek,
|
||||
PreferenceType,
|
||||
get_days,
|
||||
create_model,
|
||||
solve_model,
|
||||
get_solution
|
||||
)
|
||||
from config import YEAR, MONTH, SHIFTS, OPERATORS
|
||||
|
||||
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
|
||||
class TestDayOfWeek:
|
||||
"""Тесты для enum DayOfWeek"""
|
||||
|
||||
def test_day_of_week_values(self):
|
||||
"""Проверка значений enum"""
|
||||
assert DayOfWeek.MONDAY.value == 0
|
||||
assert DayOfWeek.TUESDAY.value == 1
|
||||
assert DayOfWeek.WEDNESDAY.value == 2
|
||||
assert DayOfWeek.THURSDAY.value == 3
|
||||
assert DayOfWeek.FRIDAY.value == 4
|
||||
assert DayOfWeek.SATURDAY.value == 5
|
||||
assert DayOfWeek.SUNDAY.value == 6
|
||||
|
||||
def test_day_of_week_names(self):
|
||||
"""Проверка названий enum"""
|
||||
assert DayOfWeek.MONDAY.name == "MONDAY"
|
||||
assert DayOfWeek.SUNDAY.name == "SUNDAY"
|
||||
|
||||
|
||||
def test_operator():
|
||||
"""Тест модели оператора"""
|
||||
print("=" * 60)
|
||||
print("TEST: Model Operator")
|
||||
print("=" * 60)
|
||||
class TestOperatorPreferences:
|
||||
"""Тесты для класса OperatorPreferences"""
|
||||
|
||||
# Создание оператора
|
||||
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
|
||||
)
|
||||
def test_creation(self):
|
||||
"""Тест создания объекта предпочтений"""
|
||||
prefs = OperatorPreferences("Тестовый")
|
||||
assert prefs.operator_name == "Тестовый"
|
||||
assert prefs.work_on_dates == []
|
||||
assert prefs.work_on_day_of_week == []
|
||||
assert prefs.work_only_shift is None
|
||||
assert prefs.off_on_dates == []
|
||||
|
||||
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)}")
|
||||
def test_to_dict_empty(self):
|
||||
"""Тест конвертации в словарь для пустых предпочтений"""
|
||||
prefs = OperatorPreferences("Тестовый")
|
||||
result = prefs.to_dict()
|
||||
assert result == {}
|
||||
|
||||
# Проверка доступности
|
||||
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_to_dict_with_dates(self):
|
||||
"""Тест конвертации с датами"""
|
||||
prefs = OperatorPreferences("Тестовый")
|
||||
prefs.work_on_dates = [10, 15, 20]
|
||||
result = prefs.to_dict()
|
||||
assert result['dates'] == [10, 15, 20]
|
||||
|
||||
def test_to_dict_with_shift(self):
|
||||
"""Тест конвертации со сменой"""
|
||||
prefs = OperatorPreferences("Тестовый")
|
||||
prefs.work_only_shift = 2
|
||||
result = prefs.to_dict()
|
||||
assert result['shift'] == 2
|
||||
|
||||
def test_to_dict_with_day_of_week(self):
|
||||
"""Тест конвертации с днем недели"""
|
||||
prefs = OperatorPreferences("Тестовый")
|
||||
prefs.work_on_day_of_week = [DayOfWeek.MONDAY, DayOfWeek.WEDNESDAY]
|
||||
result = prefs.to_dict()
|
||||
assert result['day_of_week'] == ['monday', 'wednesday']
|
||||
|
||||
def test_to_dict_with_off_dates(self):
|
||||
"""Тест конвертации с днями отдыха"""
|
||||
prefs = OperatorPreferences("Тестовый")
|
||||
prefs.off_on_dates = [5, 12, 19]
|
||||
result = prefs.to_dict()
|
||||
assert result['off_dates'] == [5, 12, 19]
|
||||
|
||||
def test_to_dict_all(self):
|
||||
"""Тест конвертации всех полей"""
|
||||
prefs = OperatorPreferences("Тестовый")
|
||||
prefs.work_on_dates = [10, 15]
|
||||
prefs.work_on_day_of_week = [DayOfWeek.MONDAY]
|
||||
prefs.work_only_shift = 1
|
||||
prefs.off_on_dates = [5]
|
||||
result = prefs.to_dict()
|
||||
assert result['dates'] == [10, 15]
|
||||
assert result['day_of_week'] == ['monday']
|
||||
assert result['shift'] == 1
|
||||
assert result['off_dates'] == [5]
|
||||
|
||||
def test_from_dict(self):
|
||||
"""Тест создания из словаря"""
|
||||
data = {
|
||||
'dates': [10, 15],
|
||||
'day_of_week': ['monday'],
|
||||
'shift': 1,
|
||||
'off_dates': [5]
|
||||
}
|
||||
prefs = OperatorPreferences.from_dict("Тестовый", data)
|
||||
assert prefs.work_on_dates == [10, 15]
|
||||
assert prefs.work_on_day_of_week == [DayOfWeek.MONDAY]
|
||||
assert prefs.work_only_shift == 1
|
||||
assert prefs.off_on_dates == [5]
|
||||
|
||||
def test_from_dict_partial(self):
|
||||
"""Тест создания из словаря с неполными данными"""
|
||||
data = {'dates': [10]}
|
||||
prefs = OperatorPreferences.from_dict("Тестовый", data)
|
||||
assert prefs.work_on_dates == [10]
|
||||
assert prefs.work_on_day_of_week == []
|
||||
assert prefs.work_only_shift is None
|
||||
assert prefs.off_on_dates == []
|
||||
|
||||
def test_from_dict_invalid_day_of_week(self):
|
||||
"""Тест обработки невалидного дня недели"""
|
||||
data = {'day_of_week': ['invalid']}
|
||||
prefs = OperatorPreferences.from_dict("Тестовый", data)
|
||||
assert prefs.work_on_day_of_week == []
|
||||
|
||||
def test_from_dict_invalid_shift(self):
|
||||
"""Тест обработки невалидной смены"""
|
||||
data = {'shift': 5}
|
||||
prefs = OperatorPreferences.from_dict("Тестовый", data)
|
||||
assert prefs.work_only_shift is None
|
||||
|
||||
def test_from_dict_invalid_dates(self):
|
||||
"""Тест обработки невалидных дат"""
|
||||
data = {'dates': [32]}
|
||||
prefs = OperatorPreferences.from_dict("Тестовый", data)
|
||||
assert prefs.work_on_dates == []
|
||||
|
||||
def test_from_dict_invalid_off_dates(self):
|
||||
"""Тест обработки невалидных дней отдыха"""
|
||||
data = {'off_dates': [32]}
|
||||
prefs = OperatorPreferences.from_dict("Тестовый", data)
|
||||
assert prefs.off_on_dates == []
|
||||
|
||||
|
||||
def test_shift():
|
||||
"""Тест модели смены"""
|
||||
print("\n" + "=" * 60)
|
||||
print("TEST: Model Shift")
|
||||
print("=" * 60)
|
||||
class TestPreferencesBuilder:
|
||||
"""Тесты для класса PreferencesBuilder"""
|
||||
|
||||
# Создание смены
|
||||
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="Вечерняя смена на складе"
|
||||
)
|
||||
def test_creation(self):
|
||||
"""Тест создания объекта"""
|
||||
builder = PreferencesBuilder("Тестовый")
|
||||
assert builder.operator_name == "Тестовый"
|
||||
assert builder._preferences is None
|
||||
|
||||
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"]
|
||||
)
|
||||
def test_add_dates(self):
|
||||
"""Тест добавления дат"""
|
||||
builder = PreferencesBuilder("Тестовый")
|
||||
builder.add_dates([10, 15])
|
||||
assert builder._preferences.work_on_dates == [10, 15]
|
||||
|
||||
is_available = shift.is_available(op)
|
||||
print(f"\nOperator {op.name} available for this shift: {is_available}")
|
||||
def test_add_day_of_week(self):
|
||||
"""Тест добавления дней недели"""
|
||||
builder = PreferencesBuilder("Тестовый")
|
||||
builder.add_day_of_week([DayOfWeek.MONDAY, DayOfWeek.WEDNESDAY])
|
||||
assert len(builder._preferences.work_on_day_of_week) == 2
|
||||
|
||||
def test_add_shift(self):
|
||||
"""Тест добавления смены"""
|
||||
builder = PreferencesBuilder("Тестовый")
|
||||
builder.add_shift(2)
|
||||
assert builder._preferences.work_only_shift == 2
|
||||
|
||||
def test_add_off_dates(self):
|
||||
"""Тест добавления дней отдыха"""
|
||||
builder = PreferencesBuilder("Тестовый")
|
||||
builder.add_off_dates([5, 12])
|
||||
assert builder._preferences.off_on_dates == [5, 12]
|
||||
|
||||
def test_to_dict(self):
|
||||
"""Тест конвертации в словарь"""
|
||||
builder = PreferencesBuilder("Тестовый")
|
||||
builder.add_dates([10])
|
||||
builder.add_shift(1)
|
||||
result = builder.to_dict()
|
||||
assert result['dates'] == [10]
|
||||
assert result['shift'] == 1
|
||||
|
||||
def test_to_dict_empty(self):
|
||||
"""Тест конвертации пустых предпочтений"""
|
||||
builder = PreferencesBuilder("Тестовый")
|
||||
result = builder.to_dict()
|
||||
assert result == {}
|
||||
|
||||
def test_to_dict_invalid(self):
|
||||
"""Тест обработки невалидных данных"""
|
||||
builder = PreferencesBuilder("Тестовый")
|
||||
builder.add_dates([32]) # Невалидная дата
|
||||
result = builder.to_dict()
|
||||
assert result == {}
|
||||
|
||||
def test_to_dict_invalid_day_of_week(self):
|
||||
"""Тест обработки невалидного дня недели"""
|
||||
builder = PreferencesBuilder("Тестовый")
|
||||
builder.add_day_of_week([DayOfWeek.MONDAY, "invalid"])
|
||||
result = builder.to_dict()
|
||||
assert result['day_of_week'] == ['monday']
|
||||
|
||||
def test_to_dict_invalid_shift(self):
|
||||
"""Тест обработки невалидной смены"""
|
||||
builder = PreferencesBuilder("Тестовый")
|
||||
builder.add_shift(5) # Невалидная смена
|
||||
result = builder.to_dict()
|
||||
assert result == {}
|
||||
|
||||
def test_to_dict_invalid_off_dates(self):
|
||||
"""Тест обработки невалидных дней отдыха"""
|
||||
builder = PreferencesBuilder("Тестовый")
|
||||
builder.add_off_dates([32]) # Невалидная дата
|
||||
result = builder.to_dict()
|
||||
assert result == {}
|
||||
|
||||
|
||||
def test_preferences():
|
||||
"""Тест модели предпочтений"""
|
||||
print("\n" + "=" * 60)
|
||||
print("TEST: Model Preferences")
|
||||
print("=" * 60)
|
||||
class TestGetDays:
|
||||
"""Тесты для функции get_days"""
|
||||
|
||||
# Создание предпочтений
|
||||
prefs = OperatorPreferences(
|
||||
operator_id="OP001",
|
||||
shift_preferences=[],
|
||||
day_preferences=[],
|
||||
time_preferences=[],
|
||||
rest_preferences=[],
|
||||
sequence_preferences=[]
|
||||
)
|
||||
def test_days_count(self):
|
||||
"""Проверка количества дней"""
|
||||
days = get_days()
|
||||
assert len(days) == 30 # Апрель 2026
|
||||
|
||||
# Добавление предпочтений
|
||||
prefs.add_shift_preference(
|
||||
ShiftType.MORNING,
|
||||
strength=80,
|
||||
reason="Хочу работать утром"
|
||||
)
|
||||
def test_days_format(self):
|
||||
"""Проверка формата дней"""
|
||||
days = get_days()
|
||||
for day in days:
|
||||
assert isinstance(day, str)
|
||||
assert len(day) == 10 # YYYY-MM-DD
|
||||
assert '-' in day
|
||||
|
||||
prefs.add_day_preference(
|
||||
{1, 2, 3}, # Пн-Ср
|
||||
strength=70,
|
||||
reason="Лучше работать в начале недели"
|
||||
)
|
||||
def test_days_sorted(self):
|
||||
"""Проверка сортировки дней"""
|
||||
days = get_days()
|
||||
assert days == sorted(days)
|
||||
|
||||
prefs.add_time_preference(
|
||||
ShiftType.MORNING,
|
||||
strength=75,
|
||||
reason="Вечерная работа - не хочу"
|
||||
)
|
||||
def test_days_start(self):
|
||||
"""Проверка первого дня"""
|
||||
days = get_days()
|
||||
assert days[0] == "2026-04-01"
|
||||
|
||||
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)
|
||||
def test_days_end(self):
|
||||
"""Проверка последнего дня"""
|
||||
days = get_days()
|
||||
assert days[-1] == "2026-04-30"
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
test_all()
|
||||
class TestCreateModel:
|
||||
"""Тесты для функции create_model"""
|
||||
|
||||
def test_model_creation(self):
|
||||
"""Тест создания модели"""
|
||||
model, variables, operators_set = create_model()
|
||||
assert model is not None
|
||||
assert variables is not None
|
||||
assert operators_set is not None
|
||||
assert len(operators_set) == len(OPERATORS)
|
||||
|
||||
def test_variables_count(self):
|
||||
"""Тест количества переменных"""
|
||||
model, variables, operators_set = create_model()
|
||||
expected_count = len(get_days()) * len(SHIFTS) * len(OPERATORS)
|
||||
assert len(variables) == expected_count
|
||||
|
||||
def test_operators_set(self):
|
||||
"""Тест множества операторов"""
|
||||
model, variables, operators_set = create_model()
|
||||
assert operators_set == set(OPERATORS)
|
||||
|
||||
def test_model_has_constraints(self):
|
||||
"""Тест наличия ограничений"""
|
||||
model, variables, operators_set = create_model()
|
||||
# Модель должна иметь ограничения
|
||||
assert model.NumConstraints() > 0
|
||||
|
||||
def test_model_has_hints(self):
|
||||
"""Тест наличия подсказок"""
|
||||
model, variables, operators_set = create_model()
|
||||
# Модель должна иметь подсказки
|
||||
assert model.NumHints() > 0
|
||||
|
||||
|
||||
class TestSolveModel:
|
||||
"""Тесты для функции solve_model"""
|
||||
|
||||
def test_solve_returns_bool(self):
|
||||
"""Тест возврата булевого значения"""
|
||||
result = solve_model()
|
||||
assert isinstance(result, bool)
|
||||
|
||||
def test_solve_with_valid_config(self):
|
||||
"""Тест решения с валидной конфигурацией"""
|
||||
# Это может занять время, поэтому тестуем только факт вызова
|
||||
try:
|
||||
result = solve_model()
|
||||
# Результат может быть True или False в зависимости от конфигурации
|
||||
assert isinstance(result, bool)
|
||||
except Exception as e:
|
||||
pytest.fail(f"Ошибка при решении модели: {e}")
|
||||
|
||||
|
||||
class TestGetSolution:
|
||||
"""Тесты для функции get_solution"""
|
||||
|
||||
def test_solution_structure(self):
|
||||
"""Тест структуры решения"""
|
||||
schedule = get_solution()
|
||||
assert isinstance(schedule, dict)
|
||||
assert len(schedule) > 0
|
||||
|
||||
def test_solution_days(self):
|
||||
"""Тест дней в решении"""
|
||||
schedule = get_solution()
|
||||
days = get_days()
|
||||
assert set(schedule.keys()) == set(days)
|
||||
|
||||
def test_solution_shifts(self):
|
||||
"""Тест смен в решении"""
|
||||
schedule = get_solution()
|
||||
for day, day_shifts in schedule.items():
|
||||
assert isinstance(day_shifts, dict)
|
||||
assert len(day_shifts) == len(SHIFTS)
|
||||
|
||||
def test_solution_operators(self):
|
||||
"""Тест операторов в решении"""
|
||||
schedule = get_solution()
|
||||
for day, day_shifts in schedule.items():
|
||||
for shift_idx, shift_data in day_shifts.items():
|
||||
assert isinstance(shift_data, dict)
|
||||
assert len(shift_data) == len(OPERATORS)
|
||||
|
||||
def test_solution_values(self):
|
||||
"""Тест значений в решении"""
|
||||
schedule = get_solution()
|
||||
for day, day_shifts in schedule.items():
|
||||
for shift_idx, shift_data in day_shifts.items():
|
||||
for op_name, value in shift_data.items():
|
||||
assert isinstance(value, bool)
|
||||
assert value in [True, False]
|
||||
|
||||
def test_solution_consistency(self):
|
||||
"""Тест согласованности решения"""
|
||||
schedule = get_solution()
|
||||
# Каждый день и смена должны иметь ровно одного оператора
|
||||
for day, day_shifts in schedule.items():
|
||||
for shift_idx, shift_data in day_shifts.items():
|
||||
active_operators = [op for op, value in shift_data.items() if value]
|
||||
assert len(active_operators) == 1
|
||||
@@ -1 +0,0 @@
|
||||
"""Unit tests package."""
|
||||
@@ -1 +0,0 @@
|
||||
"""UI package."""
|
||||
@@ -1 +0,0 @@
|
||||
"""Routes package."""
|
||||
@@ -1 +0,0 @@
|
||||
"""UI configuration."""
|
||||
@@ -1 +0,0 @@
|
||||
"""Main UI routes."""
|
||||
@@ -1 +0,0 @@
|
||||
"""Utilities package."""
|
||||
Reference in New Issue
Block a user