New functionality added inspired by Wicked Quest game.

This commit is contained in:
Storm Dragon
2025-09-16 01:21:20 -04:00
parent a96f9744a9
commit 5e5d33256e
6 changed files with 1554 additions and 25 deletions

329
README.md
View File

@@ -14,6 +14,9 @@ A Python library to make creating audio games easier.
- Input handling and keyboard controls
- Menu systems and text display
- GUI initialization
- **NEW**: Statistics tracking with level/total separation
- **NEW**: Save/load management with atomic operations
- **NEW**: Combat systems with weapons and projectiles
## Installation
@@ -129,6 +132,9 @@ The library is organized into modules, each with a specific focus:
- **display**: Text display and GUI functionality
- **menu**: Menu systems
- **utils**: Utility functions and Game class
- **stat_tracker**: Statistics tracking system
- **save_manager**: Save/load management
- **combat**: Weapon and projectile systems
## Core Classes
@@ -297,6 +303,131 @@ Where `game-name` is the lowercase, hyphenated version of your game title. For e
"My Awesome Game" would use the directory `my-awesome-game`.
### StatTracker
Flexible statistics tracking with separate level and total counters:
```python
# Initialize with default stats
stats = sg.StatTracker({
"kills": 0,
"deaths": 0,
"score": 0,
"time_played": 0.0
})
# Update stats during gameplay
stats.update_stat("kills", 1) # Increment kills
stats.update_stat("time_played", 1.5) # Add playtime
# Access current values
level_kills = stats.get_stat("kills") # Current level kills
total_kills = stats.get_stat("kills", from_total=True) # All-time kills
# Reset level stats for new level (keeps totals)
stats.reset_level()
# Add new stats dynamically
stats.add_stat("boss_kills", 0)
# Serialization for save/load
stats_data = stats.to_dict()
restored_stats = sg.StatTracker.from_dict(stats_data)
```
### SaveManager
Atomic save/load operations with corruption detection:
```python
# Initialize save manager
save_manager = sg.SaveManager("my-rpg-game")
# Create a comprehensive save
game_state = {
"player": {
"level": 5,
"hp": 80,
"inventory": ["sword", "health_potion"]
},
"world": {
"current_area": "enchanted_forest",
"completed_quests": ["tutorial", "first_boss"]
},
"stats": stats.to_dict() # Include StatTracker data
}
# Add metadata for save selection screen
metadata = {
"display_name": "Enchanted Forest - Level 5",
"level": 5,
"playtime": "2.5 hours"
}
# Create the save (atomic operation)
save_path = save_manager.create_save(game_state, metadata)
# List and load saves
save_files = save_manager.get_save_files() # Newest first
if save_files:
loaded_data, loaded_metadata = save_manager.load_save(save_files[0])
restored_stats = sg.StatTracker.from_dict(loaded_data["stats"])
```
### Combat Systems
#### Weapons
```python
# Create weapons using factory methods
sword = sg.Weapon.create_sword("Iron Sword", damage=15)
dagger = sg.Weapon.create_dagger("Steel Dagger", damage=12)
# Custom weapon with stat bonuses
bow = sg.Weapon(
name="Elvish Bow", damage=18, range_value=8,
cooldown=600, stat_bonuses={"speed": 1.15}
)
# Combat usage
if sword.can_attack(): # Check cooldown
damage = sword.attack() # Start attack
if sword.can_hit_target(enemy_pos, player_pos, facing_right, "enemy1"):
actual_damage = sword.hit_target("enemy1")
sg.speak(f"Hit for {actual_damage} damage!")
```
#### Projectiles
```python
# Create projectiles
arrow = sg.Projectile.create_arrow(
start_pos=(player_x, player_y),
direction=(1, 0) # Moving right
)
# Game loop integration
active_projectiles = []
def update_projectiles():
for projectile in active_projectiles[:]:
if not projectile.update(): # Move and check range
active_projectiles.remove(projectile)
continue
# Check enemy collisions
for enemy in enemies:
if projectile.check_collision(enemy.position, enemy.size):
damage = projectile.hit()
enemy.take_damage(damage)
active_projectiles.remove(projectile)
break
```
## Key Functions
### Game Initialization and Control
@@ -644,40 +775,60 @@ def create_sound_environment(player_x, player_y):
```
### Complete Game Structure with Class-Based Architecture
### Complete Game Structure with New Systems
```python
import libstormgames as sg
import pygame
import random
class MyGame:
class ModernRPGGame:
def __init__(self):
# Create a Game instance that manages all subsystems
self.game = sg.Game("My Advanced Game").initialize()
self.game = sg.Game("Modern RPG Demo").initialize()
# Initialize new game systems
self.player_stats = sg.StatTracker({
"level": 1, "exp": 0, "hp": 100, "mp": 50,
"kills": 0, "deaths": 0, "playtime": 0.0,
"items_found": 0, "gold": 0
})
self.save_manager = sg.SaveManager("modern-rpg-demo")
# Combat system
self.player_weapon = sg.Weapon.create_sword("Starting Sword", damage=10)
self.projectiles = []
self.enemies = []
# Game state
self.player_x = 5
self.player_y = 5
self.current_area = "village"
self.difficulty = "normal"
# Load settings
try:
self.difficulty = self.game.config_service.local_config.get("settings", "difficulty")
self.difficulty = self.game.configService.localConfig.get("settings", "difficulty")
except:
self.game.config_service.local_config.add_section("settings")
self.game.config_service.local_config.set("settings", "difficulty", "normal")
self.game.config_service.write_local_config()
self.game.configService.localConfig.add_section("settings")
self.game.configService.localConfig.set("settings", "difficulty", "normal")
self.game.configService.write_local_config()
def play_game(self):
"""Main game loop demonstrating new systems."""
self.game.speak(f"Starting game on {self.difficulty} difficulty")
self.game.play_bgm("sounds/game_music.ogg")
start_time = pygame.time.get_ticks()
# Game loop
running = True
while running:
# Update game state
self.player_x += random.uniform(-0.2, 0.2)
current_time = pygame.time.get_ticks()
# Update playtime stats
playtime_hours = (current_time - start_time) / 3600000.0 # Convert to hours
self.player_stats.set_stat("playtime", playtime_hours, level_only=True)
# Handle input
for event in pygame.event.get():
@@ -685,26 +836,148 @@ class MyGame:
if event.key == pygame.K_ESCAPE:
running = False
elif event.key == pygame.K_SPACE:
self.game.scoreboard.increase_score()
self.game.speak(f"Score: {self.game.scoreboard.get_score()}")
# Combat example
self.player_attack()
elif event.key == pygame.K_s:
# Quick save
self.quick_save()
elif event.key == pygame.K_l:
# Quick load
self.quick_load()
# Add some random sounds
if random.random() < 0.05:
sounds = self.game.sound.get_sounds()
if "ambient" in sounds:
sg.play_random_positional(sounds, "ambient",
self.player_x, self.player_x + random.uniform(-10, 10))
# Update game systems
self.update_combat()
self.update_player_stats()
# Random events
if random.random() < 0.01:
self.random_encounter()
pygame.time.delay(50)
# Game over
position = self.game.scoreboard.check_high_score()
if position:
self.game.speak(f"New high score! Position {position}")
self.game.scoreboard.add_high_score()
# Game over - update total stats
total_playtime = self.player_stats.get_stat("playtime", from_total=True)
self.game.speak(f"Session ended. Total playtime: {total_playtime:.1f} hours")
return "menu"
def player_attack(self):
"""Handle player combat."""
if self.player_weapon.can_attack():
damage = self.player_weapon.attack()
self.game.speak("Attack!")
# Simulate hitting enemies
if self.enemies and self.player_weapon.is_attack_active():
enemy = self.enemies[0] # Attack first enemy
if self.player_weapon.can_hit_target(
enemy.position, (self.player_x, self.player_y),
facing_right=True, target_id=enemy.id
):
actual_damage = self.player_weapon.hit_target(enemy.id)
self.player_stats.update_stat("damage_dealt", actual_damage)
# Remove enemy if defeated
if enemy.take_damage(actual_damage):
self.enemies.remove(enemy)
self.player_stats.update_stat("kills", 1)
self.player_stats.update_stat("exp", 25)
self.game.speak("Enemy defeated!")
def update_combat(self):
"""Update combat systems."""
# Update projectiles
for projectile in self.projectiles[:]:
if not projectile.update():
self.projectiles.remove(projectile)
continue
# Check enemy collisions
for enemy in self.enemies[:]:
if projectile.check_collision(enemy.position, enemy.size):
damage = projectile.hit()
if enemy.take_damage(damage):
self.enemies.remove(enemy)
self.player_stats.update_stat("kills", 1)
self.projectiles.remove(projectile)
break
def update_player_stats(self):
"""Handle player progression."""
exp = self.player_stats.get_stat("exp")
level = self.player_stats.get_stat("level")
# Level up check
exp_needed = level * 100
if exp >= exp_needed:
self.player_stats.set_stat("level", level + 1)
self.player_stats.set_stat("exp", exp - exp_needed)
self.player_stats.set_stat("hp", 100) # Full heal on level up
self.game.speak(f"Level up! Now level {level + 1}!")
def random_encounter(self):
"""Create random encounters."""
self.game.speak("An enemy appears!")
# Add enemy logic here
self.player_stats.update_stat("encounters", 1)
def quick_save(self):
"""Create a quick save."""
try:
save_name = f"Quick Save - Level {self.player_stats.get_stat('level')}"
self.create_complete_save(save_name)
self.game.speak("Game saved!")
except Exception as e:
self.game.speak(f"Save failed: {e}")
def quick_load(self):
"""Load the most recent save."""
try:
saves = self.save_manager.get_save_files()
if saves:
self.load_complete_save(saves[0])
self.game.speak("Game loaded!")
else:
self.game.speak("No saves found!")
except Exception as e:
self.game.speak(f"Load failed: {e}")
def create_complete_save(self, save_name):
"""Create comprehensive save with all systems."""
complete_state = {
"player_stats": self.player_stats.to_dict(),
"weapon": self.player_weapon.to_dict(),
"player_position": (self.player_x, self.player_y),
"current_area": self.current_area,
"difficulty": self.difficulty,
"enemies": [enemy.to_dict() for enemy in self.enemies],
"projectiles": [proj.to_dict() for proj in self.projectiles]
}
metadata = {
"display_name": save_name,
"level": self.player_stats.get_stat("level"),
"location": self.current_area,
"playtime": f"{self.player_stats.get_stat('playtime', from_total=True):.1f}h"
}
return self.save_manager.create_save(complete_state, metadata)
def load_complete_save(self, save_path):
"""Load comprehensive save restoring all systems."""
data, metadata = self.save_manager.load_save(save_path)
# Restore all systems
self.player_stats = sg.StatTracker.from_dict(data["player_stats"])
self.player_weapon = sg.Weapon.from_dict(data["weapon"])
self.player_x, self.player_y = data["player_position"]
self.current_area = data["current_area"]
self.difficulty = data["difficulty"]
# Restore dynamic objects (implementation depends on your enemy/projectile classes)
# self.enemies = [Enemy.from_dict(e) for e in data["enemies"]]
# self.projectiles = [sg.Projectile.from_dict(p) for p in data["projectiles"]]
def settings(self):
options = ["easy", "normal", "hard", "back"]
current = options.index(self.difficulty) if self.difficulty in options else 1
@@ -779,6 +1052,18 @@ if __name__ == "__main__":
- Game title is used to determine configuration directory paths
- Services are interconnected, so proper initialization ensures correct operation
6. **New game systems**:
- Use StatTracker for comprehensive statistics with level/total separation
- Implement SaveManager for reliable save/load with metadata
- Leverage Combat systems for professional weapon and projectile mechanics
- Combine all systems for rich, full-featured games
7. **Performance considerations**:
- Reset level stats regularly to prevent memory bloat
- Clean up old saves periodically using SaveManager methods
- Remove inactive projectiles from update loops
- Use weapon cooldowns to prevent spam attacks
## Troubleshooting

View File

@@ -72,6 +72,11 @@ from .utils import (
generate_tone
)
# Import new game systems
from .stat_tracker import StatTracker
from .save_manager import SaveManager
from .combat import Weapon, Projectile
__version__ = '2.0.0'
@@ -118,6 +123,9 @@ __all__ = [
# Game class
'Game',
# Game Systems
'StatTracker', 'SaveManager', 'Weapon', 'Projectile',
# Utils
'check_for_updates', 'get_version_tuple', 'check_compatibility',
'sanitize_filename', 'lerp', 'smooth_step', 'distance_2d',

637
combat.py Normal file
View File

@@ -0,0 +1,637 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Combat system for Storm Games.
Provides weapon and projectile systems with cooldowns, hit detection,
stat bonuses, and flexible configuration options.
"""
import time
import math
from typing import Dict, Any, Optional, Tuple, Union, Callable, List
import copy
class Weapon:
"""Generic weapon system with cooldown, range, and stat bonuses.
Features:
- Cooldown-based attack system
- Hit detection with attack duration
- Configurable stat bonuses (speed, jump, etc.)
- Serialization support for save/load
- Factory methods for common weapon types
- Sound integration (optional)
Example usage:
# Create a sword
sword = Weapon("Iron Sword", damage=10, range_value=2, cooldown=500)
# Add stat bonuses
sword.stat_bonuses = {"speed": 1.1, "critical_chance": 0.05}
# Use in combat
if sword.can_attack():
if sword.can_hit_target(enemy_pos, player_pos, facing_right):
damage_dealt = sword.attack()
"""
def __init__(
self,
name: str,
damage: Union[int, float],
range_value: Union[int, float],
attack_sound: Optional[str] = None,
hit_sound: Optional[str] = None,
cooldown: int = 500,
attack_duration: int = 200,
stat_bonuses: Optional[Dict[str, Union[int, float]]] = None
):
"""Initialize weapon.
Args:
name: Display name of the weapon
damage: Base damage value
range_value: Attack range
attack_sound: Sound key for attack (optional)
hit_sound: Sound key for successful hit (optional)
cooldown: Milliseconds between attacks
attack_duration: Milliseconds the attack window stays open
stat_bonuses: Dictionary of stat multipliers/bonuses
"""
self.name = name
self.damage = damage
self.range_value = range_value
self.attack_sound = attack_sound
self.hit_sound = hit_sound
self.cooldown = cooldown
self.attack_duration = attack_duration
self.stat_bonuses = stat_bonuses or {}
# Attack state tracking
self.last_attack_time = 0
self.attack_start_time = 0
self.is_attacking = False
self.hit_targets = set() # Track what we've hit during current attack
def can_attack(self) -> bool:
"""Check if weapon can currently attack (not on cooldown).
Returns:
True if weapon can attack, False if on cooldown
"""
current_time = time.time() * 1000 # Convert to milliseconds
return current_time - self.last_attack_time >= self.cooldown
def attack(self) -> Union[int, float]:
"""Initiate an attack.
Returns:
Damage value if attack is successful, 0 if on cooldown
"""
if not self.can_attack():
return 0
current_time = time.time() * 1000
self.last_attack_time = current_time
self.attack_start_time = current_time
self.is_attacking = True
self.hit_targets.clear()
return self.damage
def is_attack_active(self) -> bool:
"""Check if the attack window is currently active.
Returns:
True if within attack duration, False otherwise
"""
if not self.is_attacking:
return False
current_time = time.time() * 1000
elapsed = current_time - self.attack_start_time
if elapsed > self.attack_duration:
self.is_attacking = False
return False
return True
def can_hit_target(
self,
target_pos: Union[Tuple[float, float], float],
attacker_pos: Union[Tuple[float, float], float],
facing_right: bool = True,
target_id: Any = None
) -> bool:
"""Check if target is within range and can be hit.
Args:
target_pos: Position of target (x, y) or just x for 1D
attacker_pos: Position of attacker (x, y) or just x for 1D
facing_right: Direction attacker is facing (for 1D games)
target_id: Optional identifier to prevent multiple hits
Returns:
True if target can be hit, False otherwise
"""
# Check if attack is active
if not self.is_attack_active():
return False
# Check if we've already hit this target
if target_id is not None and target_id in self.hit_targets:
return False
# Calculate distance
if isinstance(target_pos, (tuple, list)) and isinstance(attacker_pos, (tuple, list)):
# 2D distance calculation
distance = math.sqrt(
(target_pos[0] - attacker_pos[0])**2 +
(target_pos[1] - attacker_pos[1])**2
)
else:
# 1D distance calculation
target_x = target_pos if isinstance(target_pos, (int, float)) else target_pos[0]
attacker_x = attacker_pos if isinstance(attacker_pos, (int, float)) else attacker_pos[0]
distance = abs(target_x - attacker_x)
# For 1D, also check direction
if facing_right and target_x <= attacker_x:
return False
elif not facing_right and target_x >= attacker_x:
return False
return distance <= self.range_value
def hit_target(self, target_id: Any = None) -> Union[int, float]:
"""Mark a target as hit and return damage.
Args:
target_id: Optional identifier for the target
Returns:
Damage value if hit is valid, 0 if target already hit
"""
if target_id is not None:
if target_id in self.hit_targets:
return 0
self.hit_targets.add(target_id)
return self.damage
def apply_stat_bonuses(self, base_stats: Dict[str, Union[int, float]]) -> Dict[str, Union[int, float]]:
"""Apply weapon's stat bonuses to base stats.
Args:
base_stats: Dictionary of base stat values
Returns:
Dictionary with stat bonuses applied
"""
modified_stats = base_stats.copy()
for stat_name, bonus in self.stat_bonuses.items():
if stat_name in modified_stats:
if isinstance(bonus, (int, float)) and bonus > 0:
if bonus > 1:
# Multiplicative bonus (e.g., 1.2 = 20% increase)
modified_stats[stat_name] *= bonus
else:
# Additive bonus (e.g., 0.05 = +5%)
modified_stats[stat_name] += bonus
return modified_stats
def to_dict(self) -> Dict[str, Any]:
"""Serialize weapon to dictionary for saving.
Returns:
Dictionary representation of weapon
"""
return {
"name": self.name,
"damage": self.damage,
"range_value": self.range_value,
"attack_sound": self.attack_sound,
"hit_sound": self.hit_sound,
"cooldown": self.cooldown,
"attack_duration": self.attack_duration,
"stat_bonuses": self.stat_bonuses.copy()
}
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> 'Weapon':
"""Create weapon from dictionary data.
Args:
data: Dictionary containing weapon data
Returns:
New Weapon instance
"""
return cls(
name=data.get("name", "Unknown Weapon"),
damage=data.get("damage", 1),
range_value=data.get("range_value", 1),
attack_sound=data.get("attack_sound"),
hit_sound=data.get("hit_sound"),
cooldown=data.get("cooldown", 500),
attack_duration=data.get("attack_duration", 200),
stat_bonuses=data.get("stat_bonuses", {})
)
@classmethod
def create_from_config(cls, config_dict: Dict[str, Any]) -> 'Weapon':
"""Create weapon from configuration dictionary.
Args:
config_dict: Configuration dictionary
Returns:
New Weapon instance
"""
return cls.from_dict(config_dict)
# Factory methods for common weapon types
@classmethod
def create_sword(cls, name: str = "Sword", damage: Union[int, float] = 10) -> 'Weapon':
"""Create a standard sword weapon.
Args:
name: Name of the sword
damage: Damage value
Returns:
Sword weapon instance
"""
return cls(
name=name,
damage=damage,
range_value=2,
attack_sound="sword_swing",
hit_sound="sword_hit",
cooldown=800,
attack_duration=300
)
@classmethod
def create_dagger(cls, name: str = "Dagger", damage: Union[int, float] = 6) -> 'Weapon':
"""Create a fast dagger weapon.
Args:
name: Name of the dagger
damage: Damage value
Returns:
Dagger weapon instance
"""
return cls(
name=name,
damage=damage,
range_value=1,
attack_sound="dagger_stab",
hit_sound="dagger_hit",
cooldown=400,
attack_duration=150,
stat_bonuses={"speed": 1.1}
)
@classmethod
def create_staff(cls, name: str = "Staff", damage: Union[int, float] = 8) -> 'Weapon':
"""Create a magical staff weapon.
Args:
name: Name of the staff
damage: Damage value
Returns:
Staff weapon instance
"""
return cls(
name=name,
damage=damage,
range_value=3,
attack_sound="staff_cast",
hit_sound="magic_hit",
cooldown=1200,
attack_duration=400
)
def __str__(self) -> str:
"""String representation of weapon."""
return f"{self.name} (Damage: {self.damage}, Range: {self.range_value})"
def __repr__(self) -> str:
"""Detailed string representation."""
return f"Weapon(name='{self.name}', damage={self.damage}, range={self.range_value})"
class Projectile:
"""Generic projectile system for ranged combat.
Features:
- Position-based movement with direction vectors
- Range limiting and collision detection
- Configurable damage, speed, and behavior
- 1D and 2D movement support
- Hit callbacks for custom effects
Example usage:
# Create an arrow
arrow = Projectile("arrow", (10, 5), (1, 0), speed=0.3, damage=8)
# Update in game loop
while arrow.is_active():
arrow.update()
# Check collision with target
if arrow.check_collision(target_pos, target_size):
damage = arrow.hit()
break
"""
def __init__(
self,
projectile_type: str,
start_pos: Union[Tuple[float, float], float],
direction: Union[Tuple[float, float], float],
speed: float = 0.2,
damage: Union[int, float] = 5,
max_range: Union[int, float] = 12,
on_hit_callback: Optional[Callable] = None
):
"""Initialize projectile.
Args:
projectile_type: Type identifier for the projectile
start_pos: Starting position (x, y) or just x for 1D
direction: Direction vector (dx, dy) or just direction for 1D
speed: Movement speed per update
damage: Damage value
max_range: Maximum travel distance
on_hit_callback: Optional function called when projectile hits
"""
self.projectile_type = projectile_type
self.damage = damage
self.max_range = max_range
self.speed = speed
self.on_hit_callback = on_hit_callback
# Position tracking
if isinstance(start_pos, (tuple, list)):
self.position = list(start_pos)
self.start_position = list(start_pos)
self.is_2d = True
else:
self.position = float(start_pos)
self.start_position = float(start_pos)
self.is_2d = False
# Direction handling
if self.is_2d and isinstance(direction, (tuple, list)):
self.direction = list(direction)
# Normalize direction vector
magnitude = math.sqrt(direction[0]**2 + direction[1]**2)
if magnitude > 0:
self.direction = [direction[0]/magnitude, direction[1]/magnitude]
elif not self.is_2d:
self.direction = float(direction)
else:
raise ValueError("Direction format must match position format (1D or 2D)")
# State tracking
self.active = True
self.has_hit = False
self.distance_traveled = 0.0
def update(self) -> bool:
"""Update projectile position.
Returns:
True if projectile is still active, False if it should be removed
"""
if not self.active:
return False
# Move projectile
if self.is_2d:
old_pos = self.position.copy()
self.position[0] += self.direction[0] * self.speed
self.position[1] += self.direction[1] * self.speed
# Calculate distance traveled
dx = self.position[0] - old_pos[0]
dy = self.position[1] - old_pos[1]
self.distance_traveled += math.sqrt(dx**2 + dy**2)
else:
old_pos = self.position
self.position += self.direction * self.speed
self.distance_traveled += abs(self.position - old_pos)
# Check if projectile has exceeded range
if self.distance_traveled >= self.max_range:
self.active = False
return self.active
def check_collision(
self,
target_pos: Union[Tuple[float, float], float],
target_size: Union[Tuple[float, float], float] = 1.0
) -> bool:
"""Check if projectile collides with a target.
Args:
target_pos: Target position (x, y) or just x for 1D
target_size: Target size (width, height) or just width for 1D
Returns:
True if collision detected, False otherwise
"""
if not self.active or self.has_hit:
return False
if self.is_2d and isinstance(target_pos, (tuple, list)):
# 2D collision detection (simple rectangle/circle)
target_width = target_size[0] if isinstance(target_size, (tuple, list)) else target_size
target_height = target_size[1] if isinstance(target_size, (tuple, list)) else target_size
# Simple bounding box collision
distance_x = abs(self.position[0] - target_pos[0])
distance_y = abs(self.position[1] - target_pos[1])
return distance_x <= target_width / 2 and distance_y <= target_height / 2
else:
# 1D collision detection
target_x = target_pos if isinstance(target_pos, (int, float)) else target_pos[0]
proj_x = self.position if isinstance(self.position, (int, float)) else self.position[0]
target_width = target_size if isinstance(target_size, (int, float)) else target_size[0]
return abs(proj_x - target_x) <= target_width / 2
def hit(self) -> Union[int, float]:
"""Mark projectile as having hit a target.
Returns:
Damage value of the projectile
"""
if self.has_hit:
return 0
self.has_hit = True
self.active = False
# Call hit callback if provided
if self.on_hit_callback:
try:
self.on_hit_callback(self)
except Exception as e:
print(f"Warning: Hit callback failed: {e}")
return self.damage
def is_active(self) -> bool:
"""Check if projectile is still active.
Returns:
True if projectile is active, False otherwise
"""
return self.active and not self.has_hit
def get_position(self) -> Union[Tuple[float, float], float]:
"""Get current position of projectile.
Returns:
Current position (x, y) or just x for 1D
"""
if self.is_2d:
return tuple(self.position)
else:
return self.position
def get_distance_traveled(self) -> float:
"""Get total distance traveled by projectile.
Returns:
Distance traveled from start position
"""
return self.distance_traveled
def to_dict(self) -> Dict[str, Any]:
"""Serialize projectile to dictionary.
Returns:
Dictionary representation of projectile
"""
return {
"projectile_type": self.projectile_type,
"position": self.position,
"start_position": self.start_position,
"direction": self.direction,
"speed": self.speed,
"damage": self.damage,
"max_range": self.max_range,
"distance_traveled": self.distance_traveled,
"active": self.active,
"has_hit": self.has_hit,
"is_2d": self.is_2d
}
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> 'Projectile':
"""Create projectile from dictionary data.
Args:
data: Dictionary containing projectile data
Returns:
New Projectile instance
"""
projectile = cls(
projectile_type=data.get("projectile_type", "projectile"),
start_pos=data.get("start_position", (0, 0)),
direction=data.get("direction", (1, 0)),
speed=data.get("speed", 0.2),
damage=data.get("damage", 5),
max_range=data.get("max_range", 12)
)
# Restore state
projectile.position = data.get("position", projectile.position)
projectile.distance_traveled = data.get("distance_traveled", 0.0)
projectile.active = data.get("active", True)
projectile.has_hit = data.get("has_hit", False)
return projectile
# Factory methods for common projectile types
@classmethod
def create_arrow(cls, start_pos: Union[Tuple[float, float], float], direction: Union[Tuple[float, float], float]) -> 'Projectile':
"""Create a standard arrow projectile.
Args:
start_pos: Starting position
direction: Direction vector
Returns:
Arrow projectile instance
"""
return cls(
projectile_type="arrow",
start_pos=start_pos,
direction=direction,
speed=0.4,
damage=7,
max_range=15
)
@classmethod
def create_fireball(cls, start_pos: Union[Tuple[float, float], float], direction: Union[Tuple[float, float], float]) -> 'Projectile':
"""Create a fireball projectile with area damage.
Args:
start_pos: Starting position
direction: Direction vector
Returns:
Fireball projectile instance
"""
return cls(
projectile_type="fireball",
start_pos=start_pos,
direction=direction,
speed=0.3,
damage=12,
max_range=20
)
@classmethod
def create_bullet(cls, start_pos: Union[Tuple[float, float], float], direction: Union[Tuple[float, float], float]) -> 'Projectile':
"""Create a fast bullet projectile.
Args:
start_pos: Starting position
direction: Direction vector
Returns:
Bullet projectile instance
"""
return cls(
projectile_type="bullet",
start_pos=start_pos,
direction=direction,
speed=0.8,
damage=5,
max_range=25
)
def __str__(self) -> str:
"""String representation of projectile."""
return f"{self.projectile_type} at {self.position} (Damage: {self.damage})"
def __repr__(self) -> str:
"""Detailed string representation."""
return f"Projectile(type='{self.projectile_type}', pos={self.position}, damage={self.damage})"

313
save_manager.py Normal file
View File

@@ -0,0 +1,313 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Save/Load management system for Storm Games.
Provides atomic save operations with XDG-compliant paths,
corruption detection, and automatic cleanup.
"""
import os
import pickle
import tempfile
import shutil
import time
from pathlib import Path
from typing import Any, Dict, List, Optional, Union
from .services import PathService
class SaveManager:
"""Generic save/load manager for game state persistence.
Features:
- XDG-compliant save directories
- Atomic file operations using temporary files
- Pickle-based serialization with version tracking
- Automatic cleanup of old saves
- Corruption detection and recovery
- Comprehensive error handling
Example usage:
# Initialize for a game
save_manager = SaveManager("my-awesome-game")
# Save game state with metadata
game_state = {"level": 5, "score": 1000, "inventory": ["sword", "potion"]}
metadata = {"display_name": "Boss Level", "level": 5}
save_manager.create_save(game_state, metadata)
# Load a save
save_files = save_manager.get_save_files()
if save_files:
game_state, metadata = save_manager.load_save(save_files[0])
"""
def __init__(self, game_name: Optional[str] = None, max_saves: int = 10):
"""Initialize SaveManager.
Args:
game_name: Name of the game for save directory. If None, uses PathService
max_saves: Maximum number of saves to keep (older saves auto-deleted)
"""
self.max_saves = max_saves
self.version = "1.0" # Save format version
# Get or initialize path service
self.path_service = PathService.get_instance()
if game_name:
# Initialize with specific game name
self.path_service.initialize(game_name)
# Ensure we have a valid game path
if not self.path_service.gamePath:
raise ValueError("Game path not initialized. Either provide game_name or initialize PathService first.")
self.save_dir = Path(self.path_service.gamePath) / "saves"
# Create saves directory if it doesn't exist
self.save_dir.mkdir(parents=True, exist_ok=True)
def create_save(self, save_data: Any, metadata: Optional[Dict[str, Any]] = None) -> str:
"""Create a new save file with the given data.
Args:
save_data: Any pickle-serializable object to save
metadata: Optional metadata dictionary for display purposes
Returns:
Path to the created save file
Raises:
Exception: If save operation fails
"""
if metadata is None:
metadata = {}
# Generate filename with timestamp
timestamp = int(time.time())
display_name = metadata.get("display_name", f"Save {timestamp}")
safe_name = self._sanitize_filename(display_name)
filename = f"{timestamp}_{safe_name}.save"
save_path = self.save_dir / filename
# Prepare save structure
save_structure = {
"version": self.version,
"timestamp": timestamp,
"metadata": metadata,
"data": save_data
}
# Atomic save operation using temporary file
with tempfile.NamedTemporaryFile(mode='wb', dir=self.save_dir, delete=False) as temp_file:
try:
pickle.dump(save_structure, temp_file)
temp_file.flush()
os.fsync(temp_file.fileno()) # Force write to disk
# Atomically move temporary file to final location
shutil.move(temp_file.name, save_path)
# Clean up old saves if we exceed max_saves
self._cleanup_old_saves()
return str(save_path)
except Exception as e:
# Clean up temporary file on error
try:
os.unlink(temp_file.name)
except:
pass
raise Exception(f"Failed to create save: {e}")
def load_save(self, filepath: Union[str, Path]) -> tuple[Any, Dict[str, Any]]:
"""Load save data from file.
Args:
filepath: Path to the save file
Returns:
Tuple of (save_data, metadata)
Raises:
Exception: If load operation fails or file is corrupted
"""
filepath = Path(filepath)
if not filepath.exists():
raise FileNotFoundError(f"Save file not found: {filepath}")
try:
with open(filepath, 'rb') as save_file:
save_structure = pickle.load(save_file)
# Validate save structure
if not isinstance(save_structure, dict):
raise ValueError("Invalid save file format")
required_fields = ["version", "timestamp", "data"]
for field in required_fields:
if field not in save_structure:
raise ValueError(f"Save file missing required field: {field}")
# Extract data
save_data = save_structure["data"]
metadata = save_structure.get("metadata", {})
return save_data, metadata
except Exception as e:
# Log corruption and attempt cleanup
print(f"Warning: Corrupted save file detected: {filepath} - {e}")
self._handle_corrupted_save(filepath)
raise Exception(f"Failed to load save: {e}")
def get_save_files(self) -> List[Path]:
"""Get list of save files sorted by creation time (newest first).
Returns:
List of Path objects for save files
"""
if not self.save_dir.exists():
return []
save_files = []
for file_path in self.save_dir.glob("*.save"):
try:
# Validate file can be opened
with open(file_path, 'rb') as f:
save_structure = pickle.load(f)
if isinstance(save_structure, dict) and "timestamp" in save_structure:
save_files.append((file_path, save_structure["timestamp"]))
except:
# Skip corrupted files
print(f"Warning: Skipping corrupted save file: {file_path}")
continue
# Sort by timestamp (newest first)
save_files.sort(key=lambda x: x[1], reverse=True)
return [file_path for file_path, _ in save_files]
def get_save_info(self, filepath: Union[str, Path]) -> Dict[str, Any]:
"""Get metadata information from a save file without loading the full data.
Args:
filepath: Path to the save file
Returns:
Dictionary with save information (metadata + timestamp)
"""
filepath = Path(filepath)
try:
with open(filepath, 'rb') as save_file:
save_structure = pickle.load(save_file)
return {
"timestamp": save_structure.get("timestamp", 0),
"metadata": save_structure.get("metadata", {}),
"version": save_structure.get("version", "unknown"),
"filepath": str(filepath),
"filename": filepath.name
}
except Exception as e:
return {
"timestamp": 0,
"metadata": {"display_name": "Corrupted Save"},
"version": "unknown",
"filepath": str(filepath),
"filename": filepath.name,
"error": str(e)
}
def has_saves(self) -> bool:
"""Check if any save files exist.
Returns:
True if save files exist, False otherwise
"""
return len(self.get_save_files()) > 0
def delete_save(self, filepath: Union[str, Path]) -> bool:
"""Delete a specific save file.
Args:
filepath: Path to the save file to delete
Returns:
True if file was deleted, False if it didn't exist
"""
filepath = Path(filepath)
try:
if filepath.exists():
filepath.unlink()
return True
return False
except Exception as e:
print(f"Warning: Failed to delete save file {filepath}: {e}")
return False
def cleanup_all_saves(self) -> int:
"""Delete all save files.
Returns:
Number of files deleted
"""
save_files = self.get_save_files()
deleted_count = 0
for save_file in save_files:
if self.delete_save(save_file):
deleted_count += 1
return deleted_count
def _cleanup_old_saves(self) -> None:
"""Remove old save files if we exceed max_saves limit."""
save_files = self.get_save_files()
if len(save_files) > self.max_saves:
# Delete oldest saves
for save_file in save_files[self.max_saves:]:
self.delete_save(save_file)
def _handle_corrupted_save(self, filepath: Path) -> None:
"""Handle a corrupted save file by moving it to a backup location."""
try:
backup_dir = self.save_dir / "corrupted"
backup_dir.mkdir(exist_ok=True)
backup_path = backup_dir / f"{filepath.name}.corrupted"
shutil.move(str(filepath), str(backup_path))
print(f"Moved corrupted save to: {backup_path}")
except Exception as e:
print(f"Failed to move corrupted save: {e}")
# As last resort, try to delete it
try:
filepath.unlink()
print(f"Deleted corrupted save: {filepath}")
except:
print(f"Could not clean up corrupted save: {filepath}")
def _sanitize_filename(self, filename: str) -> str:
"""Sanitize filename for cross-platform compatibility."""
# Remove invalid characters
invalid_chars = '<>:"/\\|?*'
for char in invalid_chars:
filename = filename.replace(char, "_")
# Limit length and handle edge cases
filename = filename.strip()
if not filename:
filename = "save"
# Truncate if too long
if len(filename) > 100:
filename = filename[:100]
return filename

237
stat_tracker.py Normal file
View File

@@ -0,0 +1,237 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Statistics tracking system for Storm Games.
Provides flexible stat tracking with separate level and total counters.
Supports any pickle-serializable data type including nested structures.
"""
from typing import Dict, Any, Union, Optional
import copy
class StatTracker:
"""Flexible statistics tracking system.
Tracks statistics with separate level and total counters, supporting
any data type that can be added or assigned.
Example usage:
# Initialize with default stats
stats = StatTracker({"kills": 0, "deaths": 0, "time_played": 0.0})
# Update stats during gameplay
stats.update_stat("kills", 1) # Increment kills
stats.update_stat("time_played", 1.5) # Add time
# Reset level stats for new level
stats.reset_level()
# Get current stats
level_kills = stats.level["kills"]
total_kills = stats.total["kills"]
"""
def __init__(self, default_stats: Optional[Dict[str, Any]] = None):
"""Initialize stat tracker with optional default statistics.
Args:
default_stats: Dictionary of default stat definitions.
If None, uses empty dict for maximum flexibility.
"""
if default_stats is None:
default_stats = {}
# Deep copy to prevent shared references
self.total = copy.deepcopy(default_stats)
self.level = copy.deepcopy(default_stats)
def update_stat(self, stat_name: str, value: Any) -> None:
"""Update a statistic by adding the value to both level and total.
Args:
stat_name: Name of the statistic to update
value: Value to add to the statistic
Note:
For numeric types, this performs addition.
For other types, behavior depends on the type's __add__ method.
If the stat doesn't exist, it will be created with the given value.
"""
if stat_name in self.level:
try:
self.level[stat_name] += value
except TypeError:
# Handle types that don't support += (assign directly)
self.level[stat_name] = value
else:
self.level[stat_name] = value
if stat_name in self.total:
try:
self.total[stat_name] += value
except TypeError:
# Handle types that don't support += (assign directly)
self.total[stat_name] = value
else:
self.total[stat_name] = value
def set_stat(self, stat_name: str, value: Any, level_only: bool = False) -> None:
"""Set a statistic to a specific value.
Args:
stat_name: Name of the statistic to set
value: Value to set
level_only: If True, only update level stats (not total)
"""
self.level[stat_name] = value
if not level_only:
self.total[stat_name] = value
def get_stat(self, stat_name: str, from_total: bool = False) -> Any:
"""Get the current value of a statistic.
Args:
stat_name: Name of the statistic to retrieve
from_total: If True, get from total stats, otherwise from level stats
Returns:
The current value of the statistic, or None if it doesn't exist
"""
source = self.total if from_total else self.level
return source.get(stat_name)
def reset_level(self) -> None:
"""Reset all level statistics to their initial values.
Preserves the structure but resets values to what they were
when the StatTracker was initialized.
"""
# Reset to initial state based on current total structure
for stat_name in self.level:
if isinstance(self.level[stat_name], (int, float)):
self.level[stat_name] = 0 if isinstance(self.level[stat_name], int) else 0.0
elif isinstance(self.level[stat_name], str):
self.level[stat_name] = ""
elif isinstance(self.level[stat_name], list):
self.level[stat_name] = []
elif isinstance(self.level[stat_name], dict):
self.level[stat_name] = {}
else:
# For other types, try to create a new instance or set to None
try:
self.level[stat_name] = type(self.level[stat_name])()
except:
self.level[stat_name] = None
def add_stat(self, stat_name: str, initial_value: Any = 0) -> None:
"""Add a new statistic to both level and total tracking.
Args:
stat_name: Name of the new statistic
initial_value: Initial value for the statistic
"""
self.level[stat_name] = copy.deepcopy(initial_value)
self.total[stat_name] = copy.deepcopy(initial_value)
def remove_stat(self, stat_name: str) -> bool:
"""Remove a statistic from both level and total tracking.
Args:
stat_name: Name of the statistic to remove
Returns:
True if the statistic was removed, False if it didn't exist
"""
removed = False
if stat_name in self.level:
del self.level[stat_name]
removed = True
if stat_name in self.total:
del self.total[stat_name]
removed = True
return removed
def get_all_stats(self, include_level: bool = True, include_total: bool = True) -> Dict[str, Any]:
"""Get dictionary of all statistics.
Args:
include_level: Include level statistics in result
include_total: Include total statistics in result
Returns:
Dictionary containing requested statistics
"""
result = {}
if include_level:
result["level"] = copy.deepcopy(self.level)
if include_total:
result["total"] = copy.deepcopy(self.total)
return result
def merge_stats(self, other_tracker: 'StatTracker') -> None:
"""Merge statistics from another StatTracker instance.
Args:
other_tracker: Another StatTracker to merge stats from
Note:
For numeric types, values are added together.
For other types, behavior depends on the type's __add__ method.
If addition fails, the other tracker's value is used.
"""
# Merge total stats
for stat_name, value in other_tracker.total.items():
if stat_name in self.total:
try:
self.total[stat_name] += value
except TypeError:
self.total[stat_name] = copy.deepcopy(value)
else:
self.total[stat_name] = copy.deepcopy(value)
# Merge level stats
for stat_name, value in other_tracker.level.items():
if stat_name in self.level:
try:
self.level[stat_name] += value
except TypeError:
self.level[stat_name] = copy.deepcopy(value)
else:
self.level[stat_name] = copy.deepcopy(value)
def to_dict(self) -> Dict[str, Any]:
"""Convert StatTracker to dictionary for serialization.
Returns:
Dictionary representation of all stats
"""
return {
"level": copy.deepcopy(self.level),
"total": copy.deepcopy(self.total)
}
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> 'StatTracker':
"""Create StatTracker from dictionary.
Args:
data: Dictionary containing level and total stats
Returns:
New StatTracker instance with loaded data
"""
tracker = cls()
if "level" in data:
tracker.level = copy.deepcopy(data["level"])
if "total" in data:
tracker.total = copy.deepcopy(data["total"])
return tracker
def __str__(self) -> str:
"""String representation of current stats."""
return f"StatTracker(level={self.level}, total={self.total})"
def __repr__(self) -> str:
"""Detailed string representation."""
return self.__str__()

View File

@@ -25,13 +25,62 @@ from .speech import Speech
from .scoreboard import Scoreboard
class Game:
"""Central class to manage all game systems."""
"""Central class to manage all game systems.
The Game class provides a unified interface to all libstormgames functionality,
including the new v2.0+ systems (StatTracker, SaveManager, Combat).
Example usage with new systems:
```python
import libstormgames as sg
# Initialize game with all systems
game = sg.Game("My RPG").initialize()
# Set up new game systems
stats = sg.StatTracker({"level": 1, "exp": 0, "kills": 0})
save_manager = sg.SaveManager("my-rpg") # Uses game's PathService
weapon = sg.Weapon.create_sword("Iron Sword", damage=15)
# Use in game loop
stats.update_stat("kills", 1)
game.speak(f"Level {stats.get_stat('level')} warrior!")
# Save complete game state
game_state = {
"stats": stats.to_dict(),
"weapon": weapon.to_dict(),
"progress": {"area": "forest", "time": 1200}
}
save_manager.create_save(game_state, {"display_name": "Forest Adventure"})
# Clean exit
game.exit()
```
Integration with services:
The Game class automatically initializes and connects all services:
- PathService: Manages XDG-compliant file paths
- ConfigService: Handles game configuration
- VolumeService: Controls audio volume settings
New systems like SaveManager automatically use the Game's PathService
for consistent directory structure.
"""
def __init__(self, title):
"""Initialize a new game.
"""Initialize a new game with all core services.
Args:
title (str): Title of the game
title (str): Title of the game, used for configuration paths
Note:
The game title is used to create XDG-compliant directories:
- Linux: ~/.config/storm-games/game-title/
- Windows: %APPDATA%/storm-games/game-title/
- macOS: ~/Library/Application Support/storm-games/game-title/
All new systems (SaveManager, etc.) will use these paths automatically.
"""
self.title = title