core/ecosystem_engine.py - master - Odisseywolf/BioSphere

core/ecosystem_engine.py
344 строки14 KB
Odisseywolf
Create: difficulty_levels.py, adaptive_engine.py, difficulty_manager.py, schema.sql, common_species.sql, queries.py, Update: ecosystem_engine.py
09 июн 2026, 14:42
Верифицирован
09 июн 2026, 14:42e0bb572
# FILE: ecosystem_engine.py
# AUTHOR: BioSphere Core Team
# DESCRIPTION: Обновлённый движок экосистемы с поддержкой сложностей и глубокой биологии

import random
import math
from typing import List, Dict, Tuple, Optional
from dataclasses import dataclass, field
from datetime import datetime

from difficulty_system.difficulty_manager import DifficultyManager
from species_db.manager import SpeciesDBManager
from models import Zone, User

# Инициализация менеджеров
difficulty_manager = DifficultyManager()
db_manager = SpeciesDBManager()


@dataclass
class SpeciesInstance:
    """Инстанс вида в конкретной зоне"""
    species_id: int
    population: int
    age_distribution: Dict[str, int]  # "juvenile", "adult", "senior"
    sex_ratio: float  # M/F ratio
    last_breeding: int  # год последнего размножения


class EcosystemEngine:
    """
    Основной движок экосистемы.
    Учитывает:
    - Уровень сложности
    - Параметры биома (показываемые и скрытые)
    - Трофические уровни и пищевые цепи
    - Циклы размножения
    - Конкуренцию и симбиоз
    - Человеческое воздействие
    """

    def __init__(self, level_id: int = 1):
        self.difficulty = difficulty_manager.set_level(level_id)
        self.zone_cache: Dict[int, Zone] = {}
        self.last_day_of_year = 365
        self.current_year = 0
        self.current_day = 0

    def simulate_day(self, zones: List[Zone], year: int, day: int) -> List[Zone]:
        """Симуляция одного дня во всех зонах"""
        self.current_year = year
        self.current_day = day
        self._apply_seasonal_factors(year, day)

        updated_zones = []
        for zone in zones:
            zone = self._simulate_zone(zone)
            updated_zones.append(zone)
        return updated_zones

    def _simulate_zone(self, zone: Zone) -> Zone:
        """Симуляция экосистемы в одной зоне"""
        # 1. Обновление скрытых параметров (автокоррекция)
        zone = self._process_hidden_parameters(zone)

        # 2. Обновление параметров биома
        zone = self._update_abiotic_factors(zone)

        # 3. Размножение и динамика популяций
        zone.species_instances = self._handle_reproduction(zone)
        zone.species_instances = self._handle_mortality(zone)

        # 4. Взаимодействия видов
        zone = self._handle_species_interactions(zone)

        # 5. Человеческое влияние
        zone = self._apply_human_impact(zone)

        # 6. Сброс на случайные события (катастрофы)
        zone = self._handle_random_events(zone)

        return zone

    def _process_hidden_parameters(self, zone: Zone) -> Zone:
        """
        Коррекция скрытых параметров по правилам уровня сложности
        """
        hidden_params = difficulty_manager.level.hidden_params

        if "microbiome" in hidden_params:
            if zone.soil_organic < 1.0:
                # Автодобавление микробов
                zone.soil_organic += 0.02
                zone.soil_N += 0.001
                # Добавляем бактерии и грибы (скрыто)
                if "bacteria" not in zone.species:
                    zone.species.append("bacteria")
                    zone.species_instances.append(SpeciesInstance(
                        species_id=db_manager.get_species_id("bacteria"),
                        population=1000,
                        age_distribution={"juvenile": 500, "adult": 400, "senior": 100},
                        sex_ratio=1.0,
                        last_breeding=self.current_year - 1
                    ))

        if "seasonality" in hidden_params:
            # Если зима — уменьшаем N, увеличиваем лесной покров
            season = self._get_season()
            if season == "winter":
                zone.soil_N *= 0.8
                zone.light_availability *= 0.6
            elif season == "autumn":
                zone.litter_layer += 0.5

        return zone

    def _update_abiotic_factors(self, zone: Zone) -> Zone:
        """Обновление видимых и скрытых параметров биома"""

        # Суточный и сезонный цикл
        base_temp = {
            "tropical": 28,
            "savanna": 30,
            "desert": 35,
            "temperate": 15,
            "boreal": -2,
            "tundra": -10
        }.get(zone.biome, 15)

        seasonal_amplitude = 10 * math.sin(2 * math.pi * self.current_day / 365)
        zone.temp = base_temp + seasonal_amplitude + random.uniform(-2, 2)

        # Влажность и осадки
        if self.current_day - zone.last_rain < 3:
            zone.humidity = min(100, zone.humidity + 10)
        else:
            zone.humidity = max(20, zone.humidity - 0.5)
        zone.water_level = zone.humidity * 0.5

        # Свет
        day_length = 12 + 4 * math.sin(2 * math.pi * self.current_day / 365 - math.pi / 2)
        zone.light_availability = day_length / 24

        # Проверка скрытых параметров
        if "soil_pH" in difficulty_manager.level.hidden_params:
            zone.soil_pH = 6.5 + random.uniform(-0.5, 0.5)  # Скрытая коррекция

        # Другие параметры (в зависимости от уровня)
        if "nutrient_cycles_N" in difficulty_manager.level.visible_params:
            zone.soil_N += 0.01 * self._calculate_decomposition_rate(zone)

        return zone

    def _handle_reproduction(self, zone: Zone) -> List[SpeciesInstance]:
        """Обработка размножения"""
        new_instances = []
        current_year = self.current_year

        for instance in zone.species_instances:
            species_data = db_manager.get_species_by_id(instance.species_id)
            if not species_data:
                continue

            breeding_season = self._parse_season(species_data.breeding_season)
            if not breeding_season:
                continue

            start_month, end_month = breeding_season
            current_month = self._get_month()

            if start_month <= current_month <= end_month:
                # Размножение
                if current_year - instance.last_breeding >= 1:
                    # Проверка баланса полов
                    if random.random() < instance.sex_ratio:
                        offsprings = int(species_data.fecundity * instance.population * 0.01)
                        new_instances.append(SpeciesInstance(
                            species_id=instance.species_id,
                            population=offsprings,
                            age_distribution={"juvenile": offsprings, "adult": 0, "senior": 0},
                            sex_ratio=species_data.sex_ratio,
                            last_breeding=current_year
                        ))

        return zone.species_instances + new_instances

    def _handle_mortality(self, zone: Zone) -> List[SpeciesInstance]:
        """Обработка смертности"""
        surviving_instances = []
        for instance in zone.species_instances:
            species_data = db_manager.get_species_by_id(instance.species_id)
            if not species_data:
                continue

            # Смерть от неподходящих условий
            if not self._is_habitat_suitable(instance.species_id, zone):
                continue

            # Естественная смертность
            mortality_rate = {
                "juvenile": 0.2,
                "adult": 0.05,
                "senior": 0.3
            }

            age_groups = instance.age_distribution
            new_groups = {}
            for age, count in age_groups.items():
                new_count = int(count * (1 - mortality_rate.get(age, 0)))
                if new_count > 0:
                    new_groups[age] = new_count
            if new_groups:
                instance.age_distribution = new_groups
                instance.population = sum(new_groups.values())
                surviving_instances.append(instance)

        return surviving_instances

    def _handle_species_interactions(self, zone: Zone) -> Zone:
        """Обработка конкуренции, симбиоза, хищничества"""
        species_map = {i.species_id: i for i in zone.species_instances}

        for instance in zone.species_instances:
            species_data = db_manager.get_species_by_id(instance.species_id)
            if not species_data:
                continue

            # Хищничество
            if species_data.trophic_level > 1:
                prey_ids = db_manager.get_prey_species(species_data.id)
                for prey_id in prey_ids:
                    if prey_id in species_map:
                        prey = species_map[prey_id]
                        # 5% шанс поедания за день
                        if random.random() < 0.05 and prey.population > 10:
                            eaten = int(prey.population * 0.01)
                            prey.population -= eaten
                            instance.population += int(eaten * 0.7)  # Эффективность

            # Конкуренция
            if species_data.competitor_ids:
                competitor_ids = eval(species_data.competitor_ids)
                for cid in competitor_ids:
                    if cid in species_map:
                        competitor = species_map[cid]
                        if instance.species_id != cid:
                            # Общий ресурс
                            resource_limit = int(1000 * zone.soil_organic)
                            if instance.population + competitor.population > resource_limit:
                                # Уменьшение популяций
                                reduction = int((instance.population + competitor.population - resource_limit) / 2)
                                instance.population -= reduction
                                competitor.population -= reduction

            # Симбиоз
            if species_data.symbiosis:
                for symb in eval(species_data.symbiosis):
                    if symb["species_id"] in species_map:
                        symb_instance = species_map[symb["species_id"]]
                        if symb["type"] == "mutualism":
                            instance.population += int(instance.population * 0.02)
                            symb_instance.population += int(symb_instance.population * 0.02)

        # Обновление зоны
        zone.species_instances = list(species_map.values())
        return zone

    def _apply_human_impact(self, zone: Zone) -> Zone:
        """Влияние человека (при наличии видов с человеком)"""
        for instance in zone.species_instances:
            species_data = db_manager.get_species_by_id(instance.species_id)
            if not species_data:
                continue

            if species_data.human_relationship == "invasive":
                instance.population *= 1.05  # Рост за счёт активности человека
            elif species_data.human_relationship in ["beneficial", "neutral"]:
                # Без изменений
                pass
            elif species_data.human_relationship == "harmful":
                instance.population *= 0.95  # Уменьшение из-за уничтожения

        return zone

    def _handle_random_events(self, zone: Zone) -> Zone:
        """Редкие катастрофы (только на высоких уровнях)"""
        if difficulty_manager.level.id >= 5 and random.random() < 0.005:  # 0.5% шанс в день
            event = random.choice(["fire", "flood", "drought"])
            if event == "fire":
                zone.species = [s for s in zone.species if s not in ["tree", "shrub"]]
                zone.soil_organic *= 0.2
            elif event == "flood":
                zone.soil_N *= 2.0
                zone.humidity = 100
        return zone

    # --- Вспомогательные методы ---

    def _is_habitat_suitable(self, species_id: int, zone: Zone) -> bool:
        species_data = db_manager.get_species_by_id(species_id)
        if not species_data:
            return False

        # Проверка температуры
        if not (species_data.min_temp <= zone.temp <= species_data.max_temp):
            return False

        # Проверка влажности
        if not (species_data.min_humidity <= zone.humidity <= species_data.max_humidity):
            return False

        return True

    def _calculate_decomposition_rate(self, zone: Zone) -> float:
        return zone.soil_organic * 0.1 * (1 + 0.05 * zone.temp)

    def _get_season(self) -> str:
        if 0 <= self.current_day < 90:
            return "spring"
        elif 90 <= self.current_day < 181:
            return "summer"
        elif 181 <= self.current_day < 273:
            return "autumn"
        else:
            return "winter"

    def _get_month(self) -> int:
        return self.current_day // 30 + 1

    def _parse_season(self, season_str: str) -> Optional[Tuple[int, int]]:
        if not season_str:
            return None
        season_data = eval(season_str)
        return season_data.get("start", 1), season_data.get("end", 12)

    def _apply_seasonal_factors(self, year: int, day: int):
        pass  # Можно добавить глобальные сезонные эффекты


# --- Старая функция для обратной совместимости ---
def update_ecosystem_day(zones: List[Zone], year: int, day: int, level_id: int = 1) -> List[Zone]:
    """Упрощённый вызов для простых сценариев"""
    engine = EcosystemEngine(level_id)
    return engine.simulate_day(zones, year, day)