9 Commits

Author SHA1 Message Date
Xuwznln
4c86816920 Update docs 2025-10-01 18:32:03 +08:00
Xuwznln
349511e00f Bump version: 0.0.3 → 0.0.4 2025-10-01 18:07:21 +08:00
Xuwznln
cbfae640d1 Bump version: 0.0.2 → 0.0.3 2025-10-01 18:00:14 +08:00
Xuwznln
de3fa68fa6 recover version 2025-10-01 18:00:10 +08:00
Xuwznln
78395349b2 Modify pypi package name 2025-10-01 17:59:35 +08:00
Xuwznln
2c19419ce8 Bump version: 0.0.1 → 0.0.2 2025-10-01 17:43:15 +08:00
Xuwznln
b9dcb259e3 Update ci 2025-10-01 17:28:57 +08:00
Xuwznln
1a8a0249c1 Add tests and docs 2025-10-01 17:23:30 +08:00
Xuwznln
a9fc374d31 Update ci 2025-10-01 17:07:31 +08:00
50 changed files with 2969 additions and 1140 deletions

10
.bumpversion.cfg Normal file
View File

@@ -0,0 +1,10 @@
[bumpversion]
current_version = 0.0.4
commit = True
tag = True
tag_name = v{new_version}
message = Bump version: {current_version} → {new_version}
[bumpversion:file:elevator_saga/__init__.py]
search = __version__ = "{current_version}"
replace = __version__ = "{new_version}"

View File

@@ -18,13 +18,13 @@ updates:
commit-message:
prefix: "deps"
include: "scope"
# GitHub Actions
- package-ecosystem: "github-actions"
directory: "/"
schedule:
interval: "weekly"
day: "monday"
day: "monday"
time: "06:00"
open-pull-requests-limit: 5
reviewers:
@@ -35,4 +35,3 @@ updates:
commit-message:
prefix: "ci"
include: "scope"

View File

@@ -1,203 +1,175 @@
name: CI
# This workflow will install Python dependencies, run tests and lint with a variety of Python versions
# For more information see: https://docs.github.com/en/actions/automating-builds-and-tests/building-and-testing-python
name: Python package
on:
push:
branches: [ main, develop ]
branches: ["main", "dev"]
pull_request:
branches: [ main, develop ]
schedule:
# Run tests daily at 6 AM UTC
- cron: '0 6 * * *'
branches: ["main", "dev"]
jobs:
test:
name: Test Python ${{ matrix.python-version }} on ${{ matrix.os }}
runs-on: ${{ matrix.os }}
strategy:
fail-fast: false
matrix:
os: [ubuntu-latest]
python-version: ['3.12']
steps:
- uses: actions/checkout@v4
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v4
with:
python-version: ${{ matrix.python-version }}
- name: Cache pip dependencies
uses: actions/cache@v3
with:
path: ~/.cache/pip
key: ${{ runner.os }}-pip-${{ hashFiles('**/pyproject.toml', '**/setup.py') }}
restore-keys: |
${{ runner.os }}-pip-
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -e .[dev]
- name: Check dependencies
run: |
python run_all_tests.py --check-deps
- name: Run linting
run: |
black --check elevator_saga tests
isort --check-only elevator_saga tests
- name: Run type checking
run: |
mypy elevator_saga
- name: Run tests with coverage
run: |
python -m pytest --cov=elevator_saga --cov-report=xml --cov-report=term-missing
- name: Upload coverage to Codecov
if: matrix.python-version == '3.12' && matrix.os == 'ubuntu-latest'
uses: codecov/codecov-action@v3
with:
file: ./coverage.xml
flags: unittests
name: codecov-umbrella
fail_ci_if_error: false
test-examples:
name: Test examples
# Step 1: Code formatting and pre-commit validation (fast failure)
code-format:
name: Code formatting and pre-commit validation
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.12'
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -e .[dev]
- name: Run example tests
run: |
python run_all_tests.py --type examples
build:
name: Build and check package
steps:
- uses: actions/checkout@v5
- name: Set up Python
uses: actions/setup-python@v6
with:
python-version: "3.10" # Use minimum version for consistency
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -e .[dev]
- name: Run pre-commit hooks
uses: pre-commit/action@v3.0.1
with:
extra_args: --all-files
# Step 2: Basic build and test with minimum Python version (3.10)
basic-build:
name: Basic build (Python 3.10, Ubuntu)
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.12'
- name: Install build dependencies
run: |
python -m pip install --upgrade pip
pip install build twine check-manifest
- name: Check manifest
run: check-manifest
- name: Build package
run: python -m build
- name: Check package
run: twine check dist/*
- name: Upload build artifacts
uses: actions/upload-artifact@v3
with:
name: dist
path: dist/
needs: [code-format] # Only run after code formatting passes
steps:
- uses: actions/checkout@v5
- name: Set up Python 3.10
uses: actions/setup-python@v6
with:
python-version: "3.10"
- name: Cache pip dependencies
uses: actions/cache@v4
with:
path: ~/.cache/pip
key: ubuntu-pip-3.10-${{ hashFiles('**/pyproject.toml') }}
restore-keys: |
ubuntu-pip-3.10-
- name: Install dependencies
run: |
python -m pip install --upgrade pip
python -m pip install pytest
pip install -e .[dev]
- name: Test with pytest
run: |
pytest -v
- name: Verify documentation builds
run: |
pip install -e .[docs]
cd docs
make html
# Step 3: Security scan
security:
name: Security scan
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.12'
- name: Install security tools
run: |
python -m pip install --upgrade pip
pip install bandit safety
- name: Run bandit security scan
run: bandit -r elevator_saga/ -f json -o bandit-report.json
continue-on-error: true
- name: Run safety security scan
run: safety check --json --output safety-report.json
continue-on-error: true
- name: Upload security reports
uses: actions/upload-artifact@v3
with:
name: security-reports
path: |
bandit-report.json
safety-report.json
if: always()
needs: [basic-build] # Run in parallel with other tests after basic build
docs:
name: Build documentation
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.12'
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -e .[dev][docs]
# 为将来的文档构建预留
- name: Check documentation
run: |
echo "Documentation build placeholder"
# sphinx-build -b html docs docs/_build/html
performance:
name: Performance benchmarks
runs-on: ubuntu-latest
if: github.event_name == 'push' && github.ref == 'refs/heads/main'
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.12'
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -e .[dev]
pip install pytest-benchmark
# 为将来的性能测试预留
- name: Run benchmarks
run: |
echo "Performance benchmarks placeholder"
# python -m pytest tests/benchmarks/ --benchmark-json=benchmark.json
- uses: actions/checkout@v5
- name: Set up Python
uses: actions/setup-python@v6
with:
python-version: "3.10"
- name: Run Safety CLI to check for vulnerabilities
uses: pyupio/safety-action@v1
with:
api-key: ${{ secrets.SAFETY_CHECK }}
output-format: json
args: --detailed-output --output-format json
continue-on-error: true
- name: Upload security reports
uses: actions/upload-artifact@v4
with:
name: security-reports
path: |
safety-report.json
if: always()
# Step 4: Package build check
package-build:
name: Package build check
runs-on: ubuntu-latest
needs: [basic-build] # Run in parallel with other checks
steps:
- uses: actions/checkout@v5
- name: Set up Python
uses: actions/setup-python@v6
with:
python-version: "3.10" # Use minimum version for consistency
- name: Install build dependencies
run: |
python -m pip install --upgrade pip
pip install build twine
- name: Build package
run: python -m build
- name: Check package
run: twine check dist/*
- name: Upload build artifacts
uses: actions/upload-artifact@v4
with:
name: dist
path: dist/
# Step 5: Full matrix build (only after all basic checks pass)
full-matrix-build:
name: Test Python ${{ matrix.python-version }} on ${{ matrix.os }}
runs-on: ${{ matrix.os }}
needs: [security, package-build] # Wait for all prerequisite checks
strategy:
fail-fast: false
matrix:
os: [ubuntu-latest, windows-latest, macos-latest]
python-version: ["3.10", "3.11", "3.12", "3.13"]
exclude:
# Skip the combination we already tested in basic-build
- os: ubuntu-latest
python-version: "3.10"
steps:
- uses: actions/checkout@v5
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v6
with:
python-version: ${{ matrix.python-version }}
- name: Cache pip dependencies
uses: actions/cache@v4
with:
path: ~/.cache/pip
key: ${{ runner.os }}-pip-${{ matrix.python-version }}-${{ hashFiles('**/pyproject.toml') }}
restore-keys: |
${{ runner.os }}-pip-${{ matrix.python-version }}-
${{ runner.os }}-pip-
- name: Install dependencies
run: |
python -m pip install --upgrade pip
python -m pip install pytest
pip install -e .[dev]
- name: Test with pytest
run: |
pytest

View File

@@ -1,269 +0,0 @@
name: Code Quality
on:
push:
branches: [ main, develop ]
pull_request:
branches: [ main, develop ]
schedule:
# Run weekly code quality checks
- cron: '0 2 * * 1'
jobs:
lint:
name: Linting and formatting
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.11'
- name: Cache pip dependencies
uses: actions/cache@v3
with:
path: ~/.cache/pip
key: ${{ runner.os }}-pip-lint-${{ hashFiles('**/pyproject.toml') }}
restore-keys: |
${{ runner.os }}-pip-lint-
${{ runner.os }}-pip-
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -e .
pip install flake8 pylint
- name: Run Black formatting check
run: |
black --check --diff elevator_saga tests
- name: Run isort import sorting check
run: |
isort --check-only --diff elevator_saga tests
- name: Run flake8
run: |
flake8 elevator_saga tests --count --select=E9,F63,F7,F82 --show-source --statistics
flake8 elevator_saga tests --count --exit-zero --max-complexity=10 --max-line-length=88 --statistics
- name: Run pylint
run: |
pylint elevator_saga --exit-zero --output-format=parseable --reports=no | tee pylint-report.txt
- name: Upload lint reports
uses: actions/upload-artifact@v3
with:
name: lint-reports
path: |
pylint-report.txt
if: always()
type-check:
name: Type checking
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.11'
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -e .
- name: Run mypy type checking
run: |
mypy elevator_saga --html-report mypy-report --txt-report mypy-report
- name: Upload type check reports
uses: actions/upload-artifact@v3
with:
name: type-check-reports
path: mypy-report/
if: always()
complexity:
name: Code complexity analysis
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.11'
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install radon xenon
- name: Run cyclomatic complexity check
run: |
radon cc elevator_saga --min B --total-average
- name: Run maintainability index
run: |
radon mi elevator_saga --min B
- name: Run complexity with xenon
run: |
xenon --max-absolute B --max-modules B --max-average A elevator_saga
continue-on-error: true
dependencies:
name: Dependency analysis
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.11'
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -e .
pip install pip-audit pipdeptree
- name: Generate dependency tree
run: |
pipdeptree --freeze > requirements-freeze.txt
pipdeptree --graph-output png > dependency-graph.png
- name: Check for known vulnerabilities
run: |
pip-audit --format=json --output=vulnerability-report.json
continue-on-error: true
- name: Upload dependency reports
uses: actions/upload-artifact@v3
with:
name: dependency-reports
path: |
requirements-freeze.txt
dependency-graph.png
vulnerability-report.json
if: always()
documentation:
name: Documentation quality
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.11'
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -e .
pip install pydocstyle interrogate
- name: Check docstring style
run: |
pydocstyle elevator_saga --count --explain --source
continue-on-error: true
- name: Check docstring coverage
run: |
interrogate elevator_saga --ignore-init-method --ignore-magic --ignore-module --ignore-nested-functions --fail-under=70
continue-on-error: true
pre-commit:
name: Pre-commit hooks
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.11'
- name: Install pre-commit
run: |
python -m pip install --upgrade pip
pip install pre-commit
- name: Cache pre-commit
uses: actions/cache@v3
with:
path: ~/.cache/pre-commit
key: pre-commit-${{ hashFiles('.pre-commit-config.yaml') }}
- name: Run pre-commit hooks
run: |
pre-commit run --all-files --show-diff-on-failure
continue-on-error: true
summary:
name: Quality summary
runs-on: ubuntu-latest
needs: [lint, type-check, complexity, dependencies, documentation, pre-commit]
if: always()
steps:
- name: Create quality report
run: |
echo "## 📊 Code Quality Report" >> $GITHUB_STEP_SUMMARY
echo "| Check | Status |" >> $GITHUB_STEP_SUMMARY
echo "|-------|--------|" >> $GITHUB_STEP_SUMMARY
# Format results
if [ "${{ needs.lint.result }}" = "success" ]; then
echo "| Linting | ✅ Passed |" >> $GITHUB_STEP_SUMMARY
else
echo "| Linting | ❌ Failed |" >> $GITHUB_STEP_SUMMARY
fi
if [ "${{ needs.type-check.result }}" = "success" ]; then
echo "| Type Check | ✅ Passed |" >> $GITHUB_STEP_SUMMARY
else
echo "| Type Check | ❌ Failed |" >> $GITHUB_STEP_SUMMARY
fi
if [ "${{ needs.complexity.result }}" = "success" ]; then
echo "| Complexity | ✅ Passed |" >> $GITHUB_STEP_SUMMARY
else
echo "| Complexity | ⚠️ Warning |" >> $GITHUB_STEP_SUMMARY
fi
if [ "${{ needs.dependencies.result }}" = "success" ]; then
echo "| Dependencies | ✅ Secure |" >> $GITHUB_STEP_SUMMARY
else
echo "| Dependencies | ⚠️ Check needed |" >> $GITHUB_STEP_SUMMARY
fi
if [ "${{ needs.documentation.result }}" = "success" ]; then
echo "| Documentation | ✅ Good |" >> $GITHUB_STEP_SUMMARY
else
echo "| Documentation | ⚠️ Needs improvement |" >> $GITHUB_STEP_SUMMARY
fi
if [ "${{ needs.pre-commit.result }}" = "success" ]; then
echo "| Pre-commit | ✅ Passed |" >> $GITHUB_STEP_SUMMARY
else
echo "| Pre-commit | ⚠️ Some hooks failed |" >> $GITHUB_STEP_SUMMARY
fi
echo "" >> $GITHUB_STEP_SUMMARY
echo "📈 **Overall Quality:** ${{ needs.lint.result == 'success' && needs.type-check.result == 'success' && 'Good' || 'Needs attention' }}" >> $GITHUB_STEP_SUMMARY

View File

@@ -1,34 +0,0 @@
name: Dependabot Auto-merge
on:
pull_request:
types: [opened, synchronize, reopened]
jobs:
dependabot:
name: Auto-merge Dependabot PRs
runs-on: ubuntu-latest
if: github.actor == 'dependabot[bot]'
steps:
- name: Dependabot metadata
id: metadata
uses: dependabot/fetch-metadata@v1
with:
github-token: "${{ secrets.GITHUB_TOKEN }}"
- name: Enable auto-merge for Dependabot PRs
if: steps.metadata.outputs.update-type == 'version-update:semver-patch' || steps.metadata.outputs.update-type == 'version-update:semver-minor'
run: gh pr merge --auto --merge "$PR_URL"
env:
PR_URL: ${{ github.event.pull_request.html_url }}
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
- name: Comment on major version updates
if: steps.metadata.outputs.update-type == 'version-update:semver-major'
run: |
gh pr comment "$PR_URL" --body "🚨 **Major version update detected!** Please review this PR carefully before merging."
env:
PR_URL: ${{ github.event.pull_request.html_url }}
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}

83
.github/workflows/docs.yml vendored Normal file
View File

@@ -0,0 +1,83 @@
name: Build and Deploy Documentation
on:
push:
branches: [main]
pull_request:
branches: [main]
workflow_dispatch:
inputs:
branch:
description: "要部署文档的分支"
required: false
default: "main"
type: string
deploy_to_pages:
description: "是否部署到 GitHub Pages"
required: false
default: true
type: boolean
# Sets permissions of the GITHUB_TOKEN to allow deployment to GitHub Pages
permissions:
contents: read
pages: write
id-token: write
# Allow only one concurrent deployment, skipping runs queued between the run in-progress and latest queued.
# However, do NOT cancel in-progress runs as we want to allow these production deployments to complete.
concurrency:
group: "pages"
cancel-in-progress: false
jobs:
# Build documentation
build:
runs-on: ubuntu-latest
steps:
- name: Checkout
uses: actions/checkout@v5
with:
ref: ${{ github.event.inputs.branch || github.ref }}
- name: Set up Python
uses: actions/setup-python@v6
with:
python-version: "3.10"
- name: Install dependencies
run: |
python -m pip install --upgrade pip
# Install package in development mode to get version info
pip install -e .
# Install documentation dependencies
pip install -e .[docs]
- name: Setup Pages
id: pages
uses: actions/configure-pages@v5
if: github.ref == 'refs/heads/main' || (github.event_name == 'workflow_dispatch' && github.event.inputs.deploy_to_pages == 'true')
- name: Build Sphinx documentation
run: |
cd docs
make html
- name: Upload artifact
uses: actions/upload-pages-artifact@v4
if: github.ref == 'refs/heads/main' || (github.event_name == 'workflow_dispatch' && github.event.inputs.deploy_to_pages == 'true')
with:
path: docs/_build/html
# Deploy to GitHub Pages
deploy:
if: github.ref == 'refs/heads/main' || (github.event_name == 'workflow_dispatch' && github.event.inputs.deploy_to_pages == 'true')
environment:
name: github-pages
url: ${{ steps.deployment.outputs.page_url }}
runs-on: ubuntu-latest
needs: build
steps:
- name: Deploy to GitHub Pages
id: deployment
uses: actions/deploy-pages@v4

View File

@@ -1,189 +1,258 @@
name: Publish to PyPI
# This workflow will upload a Python Package to PyPI when a release is created
# For more information see: https://docs.github.com/en/actions/automating-builds-and-tests/building-and-testing-python#publishing-to-package-registries
# This workflow uses actions that are not certified by GitHub.
# They are provided by a third-party and are governed by
# separate terms of service, privacy policy, and support
# documentation.
name: Upload PyPI package
on:
release:
types: [published]
types: [published, edited]
workflow_dispatch:
inputs:
test_pypi:
description: 'Publish to Test PyPI instead of PyPI'
description: "Publish to Test PyPI instead of PyPI"
required: false
default: 'false'
default: false
type: boolean
permissions:
contents: read
jobs:
test:
name: Run tests before publish
# Step 1: Code formatting and pre-commit validation (fast failure)
code-format:
name: Code formatting and pre-commit validation
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.12'
# - name: Install dependencies
# run: |
# python -m pip install --upgrade pip
# pip install -e .
# - name: Run comprehensive tests
# run: |
# python -m pytest --cov=msgcenterpy --cov-fail-under=80
# - name: Run linting
# run: |
# black --check msgcenterpy tests
# isort --check-only msgcenterpy tests
# mypy msgcenterpy
build:
name: Build package
steps:
- uses: actions/checkout@v5
- name: Set up Python
uses: actions/setup-python@v6
with:
python-version: "3.10" # Use minimum version for consistency
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -e .[dev]
- name: Run pre-commit hooks
uses: pre-commit/action@v3.0.1
with:
extra_args: --all-files
# Step 2: Basic build and test with minimum Python version (3.10)
basic-build:
name: Basic build (Python 3.10, Ubuntu)
runs-on: ubuntu-latest
needs: test
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.12'
# - name: Install build dependencies
# run: |
# python -m pip install --upgrade pip
# pip install build twine check-manifest
# - name: Verify version consistency
# run: |
# # 检查版本号一致性
# VERSION=$(python -c "import elevator_saga; print(elevator_saga.__version__)" 2>/dev/null || echo "unknown")
# TAG_VERSION="${GITHUB_REF#refs/tags/v}"
# if [ "$GITHUB_EVENT_NAME" = "release" ]; then
# if [ "$VERSION" != "$TAG_VERSION" ]; then
# echo "Version mismatch: package=$VERSION, tag=$TAG_VERSION"
# exit 1
# fi
# fi
# - name: Check manifest
# run: check-manifest
- name: Install build dependencies
run: |
python -m pip install --upgrade pip
pip install build twine check-manifest
- name: Build package
run: |
python -m build
# - name: Check package
# run: |
# twine check dist/*
- name: Upload build artifacts
uses: actions/upload-artifact@v4
with:
name: dist-${{ github.run_number }}
path: dist/
retention-days: 30
needs: [code-format] # Only run after code formatting passes
publish-test:
name: Publish to Test PyPI
steps:
- uses: actions/checkout@v5
- name: Set up Python 3.10
uses: actions/setup-python@v6
with:
python-version: "3.10"
- name: Cache pip dependencies
uses: actions/cache@v4
with:
path: ~/.cache/pip
key: ubuntu-pip-3.10-${{ hashFiles('**/pyproject.toml') }}
restore-keys: |
ubuntu-pip-3.10-
- name: Install dependencies
run: |
python -m pip install --upgrade pip
python -m pip install pytest
pip install -e .[dev]
- name: Test with pytest
run: |
pytest -v
# Step 3: Security scan
security:
name: Security scan
runs-on: ubuntu-latest
needs: build
if: github.event.inputs.test_pypi == 'true' || (github.event_name == 'release' && github.event.release.prerelease)
environment:
name: test-pypi
url: https://test.pypi.org/p/elevator-saga
steps:
- name: Download build artifacts
uses: actions/download-artifact@v3
with:
name: dist-${{ github.run_number }}
path: dist/
- name: Publish to Test PyPI
uses: pypa/gh-action-pypi-publish@release/v1
with:
repository-url: https://test.pypi.org/legacy/
password: ${{ secrets.TEST_PYPI_API_TOKEN }}
verbose: true
needs: [basic-build] # Run in parallel with other tests after basic build
publish-pypi:
steps:
- uses: actions/checkout@v5
- name: Run Safety CLI to check for vulnerabilities
uses: pyupio/safety-action@v1
with:
api-key: ${{ secrets.SAFETY_CHECK }}
output-format: json
args: --detailed-output --output-format json
continue-on-error: true
- name: Upload security reports
uses: actions/upload-artifact@v4
with:
name: security-reports
path: |
safety-report.json
if: always()
release-build:
name: Build release distributions
runs-on: ubuntu-latest
needs: [basic-build]
steps:
- uses: actions/checkout@v5
- name: Set up Python
uses: actions/setup-python@v6
with:
python-version: "3.10" # Use minimum version for consistency
- name: Install build dependencies
run: |
python -m pip install --upgrade pip
python -m pip install build twine
- name: Verify version consistency
if: github.event_name == 'release' && (github.event.action == 'published' || (github.event.action == 'edited' && !github.event.release.prerelease))
run: |
# Install package first
pip install -e .
# Get package version (fail fast if not available)
VERSION=$(python -c "import elevator_saga; print(elevator_saga.__version__)")
# Handle both v0.0.3 and 0.0.3 tag formats
RAW_TAG="${GITHUB_REF#refs/tags/}"
if [[ "$RAW_TAG" == v* ]]; then
TAG_VERSION="${RAW_TAG#v}"
else
TAG_VERSION="$RAW_TAG"
fi
echo "Package version: $VERSION"
echo "Tag version: $TAG_VERSION"
if [ "$VERSION" != "$TAG_VERSION" ]; then
echo "❌ Version mismatch: package=$VERSION, tag=$TAG_VERSION"
echo "Please ensure the package version matches the git tag"
exit 1
fi
echo "✅ Version verification passed: $VERSION"
- name: Build release distributions
run: |
python -m build
- name: Check package
run: |
twine check dist/*
- name: Upload distributions
uses: actions/upload-artifact@v4
with:
name: release-dists
path: dist/
pypi-publish:
name: Publish to PyPI
runs-on: ubuntu-latest
needs: build
if: github.event_name == 'release' && !github.event.release.prerelease && github.event.inputs.test_pypi != 'true'
environment:
name: pypi
url: https://pypi.org/p/elevator-saga
needs:
- release-build
if: github.event_name == 'release' && !github.event.release.prerelease && github.event.inputs.test_pypi != 'true' && (github.event.action == 'published' || github.event.action == 'edited')
permissions:
# IMPORTANT: this permission is mandatory for trusted publishing
id-token: write
# Note: For enhanced security, consider configuring deployment environments
# in your GitHub repository settings with protection rules
steps:
- name: Download build artifacts
uses: actions/download-artifact@v3
with:
name: dist-${{ github.run_number }}
path: dist/
- name: Publish to PyPI
uses: pypa/gh-action-pypi-publish@release/v1
with:
password: ${{ secrets.PYPI_API_TOKEN }}
verbose: true
- name: Retrieve release distributions
uses: actions/download-artifact@v5
with:
name: release-dists
path: dist/
- name: Publish release distributions to PyPI
uses: pypa/gh-action-pypi-publish@release/v1
test-pypi-publish:
name: Publish to Test PyPI
runs-on: ubuntu-latest
needs:
- release-build
if: github.event.inputs.test_pypi == 'true' || (github.event_name == 'release' && github.event.release.prerelease && (github.event.action == 'published' || github.event.action == 'edited'))
permissions:
# IMPORTANT: this permission is mandatory for trusted publishing
id-token: write
# Note: For enhanced security, consider configuring deployment environments
# in your GitHub repository settings with protection rules
steps:
- name: Retrieve release distributions
uses: actions/download-artifact@v5
with:
name: release-dists
path: dist/
- name: Publish release distributions to Test PyPI
uses: pypa/gh-action-pypi-publish@release/v1
with:
repository-url: https://test.pypi.org/legacy/
create-github-release-assets:
name: Add assets to GitHub release
runs-on: ubuntu-latest
needs: build
if: github.event_name == 'release'
needs: release-build
if: github.event_name == 'release' && (github.event.action == 'published' || github.event.action == 'edited')
permissions:
contents: write # Need write access to upload release assets
steps:
- name: Download build artifacts
uses: actions/download-artifact@v3
with:
name: dist-${{ github.run_number }}
path: dist/
- name: Upload release assets
uses: softprops/action-gh-release@v1
with:
files: dist/*
env:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
- name: Retrieve release distributions
uses: actions/download-artifact@v5
with:
name: release-dists
path: dist/
- name: Upload release assets
uses: softprops/action-gh-release@v2
with:
files: dist/*
env:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
post-publish:
name: Post-publish tasks
runs-on: ubuntu-latest
needs: [publish-pypi, publish-test]
if: always() && (needs.publish-pypi.result == 'success' || needs.publish-test.result == 'success')
steps:
- uses: actions/checkout@v4
- name: Create deployment summary
run: |
echo "## 🚀 Deployment Summary" >> $GITHUB_STEP_SUMMARY
echo "| Item | Status |" >> $GITHUB_STEP_SUMMARY
echo "|------|--------|" >> $GITHUB_STEP_SUMMARY
if [ "${{ needs.publish-pypi.result }}" = "success" ]; then
echo "| PyPI | ✅ Published |" >> $GITHUB_STEP_SUMMARY
elif [ "${{ needs.publish-test.result }}" = "success" ]; then
echo "| Test PyPI | ✅ Published |" >> $GITHUB_STEP_SUMMARY
fi
echo "| GitHub Release | ✅ Assets uploaded |" >> $GITHUB_STEP_SUMMARY
echo "| Version | ${{ github.event.release.tag_name || 'test' }} |" >> $GITHUB_STEP_SUMMARY
# 为将来的通知预留
- name: Notify team
run: |
echo "Package published successfully!"
# 可以添加 Slack、Discord 等通知
needs: [pypi-publish, test-pypi-publish]
if: always() && (needs.pypi-publish.result == 'success' || needs.test-pypi-publish.result == 'success')
steps:
- uses: actions/checkout@v5
- name: Create deployment summary
run: |
echo "## Deployment Summary" >> $GITHUB_STEP_SUMMARY
echo "| Item | Status |" >> $GITHUB_STEP_SUMMARY
echo "|------|--------|" >> $GITHUB_STEP_SUMMARY
if [ "${{ needs.pypi-publish.result }}" = "success" ]; then
echo "| PyPI | Published |" >> $GITHUB_STEP_SUMMARY
elif [ "${{ needs.test-pypi-publish.result }}" = "success" ]; then
echo "| Test PyPI | Published |" >> $GITHUB_STEP_SUMMARY
fi
echo "| GitHub Release | Assets uploaded |" >> $GITHUB_STEP_SUMMARY
echo "| Version | ${{ github.event.release.tag_name || 'test' }} |" >> $GITHUB_STEP_SUMMARY

65
.gitignore vendored
View File

@@ -1,5 +1,60 @@
.vscode
.idea
__pycache__
.mypy_cache
elevator_saga.egg-info
# ================================
# Python-related files
# ================================
# Compiled Python files
__pycache__/
*.py[cod]
*$py.class
# Distribution / packaging
.Python
build/
dist/
*.egg-info/
.installed.cfg
*.egg
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py,cover
.hypothesis/
.pytest_cache/
cover/
# Type checking
.mypy_cache/
# Documentation
docs/_build/
docs/_static/
docs/_templates/
docs/_static/
# ================================
# IDE and Editor files
# ================================
# PyCharm
.idea/
# Visual Studio Code
.vscode/
.cursor/
.cursorignore
pyrightconfig.json
# ================================
# Operating System files
# ================================
# macOS
.DS_Store

View File

@@ -5,33 +5,34 @@ repos:
hooks:
- id: black
language_version: python3
args: ["--line-length=88"]
args: ["--line-length=120"]
# Import sorting
- repo: https://github.com/pycqa/isort
rev: 5.13.2
hooks:
- id: isort
args: ["--profile", "black", "--multi-line", "3"]
args:
["--profile", "black", "--multi-line", "3", "--line-length", "120"]
# Linting
- repo: https://github.com/pycqa/flake8
rev: 7.0.0
hooks:
- id: flake8
args:
- "--max-line-length=88"
- "--extend-ignore=E203,W503,F401"
- "--exclude=build,dist,.eggs"
args:
- "--max-line-length=200" # Allow longer lines after black formatting
- "--extend-ignore=E203,W503,F401,E402,E721,F841"
- "--exclude=build,dist,__pycache__,.mypy_cache,.pytest_cache,htmlcov,.idea,.vscode,docs/_build,elevatorpy.egg-info,elevator_saga.egg-info,.eggs"
# Type checking
- repo: https://github.com/pre-commit/mirrors-mypy
rev: v1.8.0
hooks:
- id: mypy
additional_dependencies: [types-PyYAML]
args: ["--ignore-missing-imports", "--scripts-are-modules"]
exclude: "^(tests/|examples/)"
additional_dependencies: [types-PyYAML, types-jsonschema]
args: ["--ignore-missing-imports", "--disable-error-code=unused-ignore"]
files: "^(elevator_saga/)" # Check both source code
# General pre-commit hooks
- repo: https://github.com/pre-commit/pre-commit-hooks
@@ -45,18 +46,16 @@ repos:
- id: check-json
- id: check-toml
- id: check-xml
# Security
- id: check-merge-conflict
- id: check-case-conflict
- id: check-symlinks
- id: check-added-large-files
args: ["--maxkb=1000"]
# Python specific
- id: check-ast
- id: check-builtin-literals
- id: check-docstring-first
- id: debug-statements
- id: name-tests-test
args: ["--django"]
@@ -67,47 +66,17 @@ repos:
hooks:
- id: bandit
args: ["-c", "pyproject.toml"]
additional_dependencies: ["bandit[toml]"]
additional_dependencies: ["bandit[toml]", "pbr"]
exclude: "^tests/"
# Documentation
- repo: https://github.com/pycqa/pydocstyle
rev: 6.3.0
hooks:
- id: pydocstyle
args: ["--convention=google"]
exclude: "^(tests/|examples/)"
# YAML/JSON formatting
- repo: https://github.com/pre-commit/mirrors-prettier
rev: v4.0.0-alpha.8
hooks:
- id: prettier
types_or: [yaml, json, markdown]
exclude: "^(.github/)"
# Spell checking (optional)
- repo: https://github.com/codespell-project/codespell
rev: v2.2.6
hooks:
- id: codespell
args: ["--write-changes"]
exclude: "^(.*\\.po|.*\\.pot|CHANGELOG\\.md)$"
exclude: "^(build/|dist/|__pycache__/|\\.mypy_cache/|\\.pytest_cache/|htmlcov/|\\.idea/|\\.vscode/|docs/_build/|elevatorpy\\.egg-info/|elevator_saga\\.egg-info/)"
# Global settings
default_stages: [commit, push]
default_stages: [pre-commit, pre-push]
fail_fast: false
# CI settings
ci:
autofix_commit_msg: |
[pre-commit.ci] auto fixes from pre-commit.com hooks
for more information, see https://pre-commit.ci
autofix_prs: true
autoupdate_branch: ''
autoupdate_commit_msg: '[pre-commit.ci] pre-commit autoupdate'
autoupdate_schedule: weekly
skip: []
submodules: false

21
LICENSE Normal file
View File

@@ -0,0 +1,21 @@
MIT License
Copyright (c) 2025 Elevator Saga Team
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

70
README.md Normal file
View File

@@ -0,0 +1,70 @@
# Elevator Saga
<div align="center">
[![PyPI version](https://badge.fury.io/py/elevator-py.svg)](https://badge.fury.io/py/elevator-py)
[![Python versions](https://img.shields.io/pypi/pyversions/elevator-py.svg)](https://pypi.org/project/elevator-py/)
[![Build Status](https://github.com/ZGCA-Forge/Elevator/actions/workflows/ci.yml/badge.svg)](https://github.com/ZGCA-Forge/Elevator/actions)
[![Documentation](https://img.shields.io/badge/docs-GitHub%20Pages-brightgreen)](https://zgca-forge.github.io/Elevator/)
[![GitHub stars](https://img.shields.io/github/stars/ZGCA-Forge/Elevator.svg?style=social&label=Star)](https://github.com/ZGCA-Forge/Elevator)
[![GitHub forks](https://img.shields.io/github/forks/ZGCA-Forge/Elevator.svg?style=social&label=Fork)](https://github.com/ZGCA-Forge/Elevator/fork)
[![GitHub issues](https://img.shields.io/github/issues/ZGCA-Forge/Elevator.svg)](https://github.com/ZGCA-Forge/Elevator/issues)
[![License](https://img.shields.io/badge/license-MIT-blue.svg)](https://github.com/ZGCA-Forge/Elevator/blob/main/LICENSE)
</div>
---
Elevator Saga is a Python implementation of an elevator [simulation game](https://play.elevatorsaga.com/) with a event-driven architecture Design and optimize elevator control algorithms to efficiently transport passengers in buildings.
### Features
- 🏢 **Realistic Simulation**: Physics-based elevator movement with acceleration, deceleration, and realistic timing
## Installation
### Basic Installation
```bash
pip install elevator-py
```
## Quick Start
### Running the Game
```bash
# Start the backend simulator (Terminal #1)
python -m elevator_saga.server.simulator
```
```bash
# Start your own client (Terminal #2)
# Example:
python -m elevator_saga.client_examples.bus_example
```
## Documentation
For detailed documentation, please visit: [https://zgca-forge.github.io/Elevator/](https://zgca-forge.github.io/Elevator/)
## Contributing
Contributions are welcome! Please feel free to submit a Pull Request.
## Star History
[![Star History Chart](https://api.star-history.com/svg?repos=ZGCA-Forge/Elevator&type=Date)](https://star-history.com/#ZGCA-Forge/Elevator&Date)
## License
This project is licensed under MIT License - see the [LICENSE](LICENSE) file for details.
---
<div align="center">
Made with ❤️ by the ZGCA-Forge Team
</div>

3
docs/.gitignore vendored Normal file
View File

@@ -0,0 +1,3 @@
_build/
_static/
_templates/

20
docs/Makefile Normal file
View File

@@ -0,0 +1,20 @@
# Minimal makefile for Sphinx documentation
#
# You can set these variables from the command line, and also
# from the environment for the first two.
SPHINXOPTS ?=
SPHINXBUILD ?= sphinx-build
SOURCEDIR = .
BUILDDIR = _build
# Put it first so that "make" without argument is like "make help".
help:
@$(SPHINXBUILD) -M help "$(SOURCEDIR)" "$(BUILDDIR)" $(SPHINXOPTS) $(O)
.PHONY: help Makefile
# Catch-all target: route all unknown targets to Sphinx using the new
# "make mode" option. $(O) is meant as a shortcut for $(SPHINXOPTS).
%: Makefile
@$(SPHINXBUILD) -M $@ "$(SOURCEDIR)" "$(BUILDDIR)" $(SPHINXOPTS) $(O)

61
docs/README.md Normal file
View File

@@ -0,0 +1,61 @@
# Elevator Saga Documentation
This directory contains the Sphinx documentation for Elevator Saga.
## Building the Documentation
### Install Dependencies
```bash
pip install -r requirements.txt
```
### Build HTML Documentation
```bash
make html
```
The generated HTML documentation will be in `_build/html/`.
### View Documentation
Open `_build/html/index.html` in your browser:
```bash
# On Linux
xdg-open _build/html/index.html
# On macOS
open _build/html/index.html
# On Windows
start _build/html/index.html
```
### Other Build Formats
```bash
make latexpdf # Build PDF documentation
make epub # Build EPUB documentation
make clean # Clean build directory
```
## Documentation Structure
- `index.rst` - Main documentation index
- `models.rst` - Data models documentation
- `client.rst` - Client architecture and proxy models
- `communication.rst` - HTTP communication protocol
- `events.rst` - Event system and tick-based simulation
- `api/modules.rst` - Auto-generated API reference
## Contributing
When adding new documentation:
1. Create `.rst` files for new topics
2. Add them to the `toctree` in `index.rst`
3. Follow reStructuredText syntax
4. Build locally to verify formatting
5. Submit PR with documentation changes

39
docs/api/modules.rst Normal file
View File

@@ -0,0 +1,39 @@
API Reference
=============
.. toctree::
:maxdepth: 4
Core Modules
------------
.. automodule:: elevator_saga.core.models
:members:
:undoc-members:
:show-inheritance:
Client Modules
--------------
.. automodule:: elevator_saga.client.api_client
:members:
:undoc-members:
:show-inheritance:
.. automodule:: elevator_saga.client.proxy_models
:members:
:undoc-members:
:show-inheritance:
.. automodule:: elevator_saga.client.base_controller
:members:
:undoc-members:
:show-inheritance:
Server Modules
--------------
.. automodule:: elevator_saga.server.simulator
:members:
:undoc-members:
:show-inheritance:

351
docs/client.rst Normal file
View File

@@ -0,0 +1,351 @@
Client Architecture and Proxy Models
====================================
The Elevator Saga client provides a powerful abstraction layer that allows you to interact with the simulation using dynamic proxy objects. This architecture provides type-safe, read-only access to simulation state while enabling elevator control commands.
Overview
--------
The client architecture consists of three main components:
1. **Proxy Models** (``proxy_models.py``): Dynamic proxies that provide transparent access to server state
2. **API Client** (``api_client.py``): HTTP client for communicating with the server
3. **Base Controller** (``base_controller.py``): Abstract base class for implementing control algorithms
Proxy Models
------------
Proxy models in ``elevator_saga/client/proxy_models.py`` provide a clever way to access remote state as if it were local. They inherit from the data models but override attribute access to fetch fresh data from the server.
How Proxy Models Work
~~~~~~~~~~~~~~~~~~~~~~
The proxy pattern implementation uses Python's ``__getattribute__`` magic method to intercept attribute access:
1. When you access an attribute (e.g., ``elevator.current_floor``), the proxy intercepts the call
2. The proxy fetches the latest state from the server via API client
3. The proxy returns the requested attribute from the fresh state
4. All accesses are **read-only** to maintain consistency
This design ensures you always work with the most up-to-date simulation state without manual refresh calls.
ProxyElevator
~~~~~~~~~~~~~
Dynamic proxy for ``ElevatorState`` that provides access to elevator properties and control methods:
.. code-block:: python
class ProxyElevator(ElevatorState):
"""
Dynamic proxy for elevator state
Provides complete type-safe access and control methods
"""
def __init__(self, elevator_id: int, api_client: ElevatorAPIClient):
self._elevator_id = elevator_id
self._api_client = api_client
self._init_ok = True
def go_to_floor(self, floor: int, immediate: bool = False) -> bool:
"""Command elevator to go to specified floor"""
return self._api_client.go_to_floor(self._elevator_id, floor, immediate)
**Accessible Properties** (from ElevatorState):
- ``id``: Elevator identifier
- ``current_floor``: Current floor number
- ``current_floor_float``: Precise position (e.g., 2.5)
- ``target_floor``: Destination floor
- ``position``: Full Position object
- ``passengers``: List of passenger IDs on board
- ``max_capacity``: Maximum passenger capacity
- ``run_status``: Current ElevatorStatus
- ``target_floor_direction``: Direction to target (UP/DOWN/STOPPED)
- ``last_tick_direction``: Previous movement direction
- ``is_idle``: Whether stopped
- ``is_full``: Whether at capacity
- ``is_running``: Whether in motion
- ``pressed_floors``: Destination floors of current passengers
- ``load_factor``: Current load (0.0 to 1.0)
- ``indicators``: Up/down indicator lights
**Control Method**:
- ``go_to_floor(floor, immediate=False)``: Send elevator to floor
- ``immediate=True``: Change target immediately
- ``immediate=False``: Queue as next target after current destination
**Example Usage**:
.. code-block:: python
# Access elevator state
print(f"Elevator {elevator.id} at floor {elevator.current_floor}")
print(f"Direction: {elevator.target_floor_direction.value}")
print(f"Passengers: {len(elevator.passengers)}/{elevator.max_capacity}")
# Check status
if elevator.is_idle:
print("Elevator is idle")
elif elevator.is_full:
print("Elevator is full!")
# Control elevator
if elevator.current_floor == 0:
elevator.go_to_floor(5) # Send to floor 5
ProxyFloor
~~~~~~~~~~
Dynamic proxy for ``FloorState`` that provides access to floor information:
.. code-block:: python
class ProxyFloor(FloorState):
"""
Dynamic proxy for floor state
Provides read-only access to floor information
"""
def __init__(self, floor_id: int, api_client: ElevatorAPIClient):
self._floor_id = floor_id
self._api_client = api_client
self._init_ok = True
**Accessible Properties** (from FloorState):
- ``floor``: Floor number
- ``up_queue``: List of passenger IDs waiting to go up
- ``down_queue``: List of passenger IDs waiting to go down
- ``has_waiting_passengers``: Whether any passengers are waiting
- ``total_waiting``: Total number of waiting passengers
**Example Usage**:
.. code-block:: python
floor = floors[0]
print(f"Floor {floor.floor}")
print(f"Waiting to go up: {len(floor.up_queue)} passengers")
print(f"Waiting to go down: {len(floor.down_queue)} passengers")
if floor.has_waiting_passengers:
print(f"Total waiting: {floor.total_waiting}")
ProxyPassenger
~~~~~~~~~~~~~~
Dynamic proxy for ``PassengerInfo`` that provides access to passenger information:
.. code-block:: python
class ProxyPassenger(PassengerInfo):
"""
Dynamic proxy for passenger information
Provides read-only access to passenger data
"""
def __init__(self, passenger_id: int, api_client: ElevatorAPIClient):
self._passenger_id = passenger_id
self._api_client = api_client
self._init_ok = True
**Accessible Properties** (from PassengerInfo):
- ``id``: Passenger identifier
- ``origin``: Starting floor
- ``destination``: Target floor
- ``arrive_tick``: When passenger appeared
- ``pickup_tick``: When passenger boarded (0 if waiting)
- ``dropoff_tick``: When passenger reached destination (0 if in transit)
- ``elevator_id``: Current elevator ID (None if waiting)
- ``status``: Current PassengerStatus
- ``wait_time``: Ticks waited before boarding
- ``system_time``: Total ticks in system
- ``travel_direction``: UP or DOWN
**Example Usage**:
.. code-block:: python
print(f"Passenger {passenger.id}")
print(f"From floor {passenger.origin} to {passenger.destination}")
print(f"Status: {passenger.status.value}")
if passenger.status == PassengerStatus.IN_ELEVATOR:
print(f"In elevator {passenger.elevator_id}")
print(f"Waited {passenger.wait_time} ticks")
Read-Only Protection
~~~~~~~~~~~~~~~~~~~~
All proxy models are **read-only**. Attempting to modify attributes will raise an error:
.. code-block:: python
elevator.current_floor = 5 # ❌ Raises AttributeError
elevator.passengers.append(123) # ❌ Raises AttributeError
This ensures that:
1. Client cannot corrupt server state
2. All state changes go through proper API commands
3. State consistency is maintained
Implementation Details
~~~~~~~~~~~~~~~~~~~~~~
The proxy implementation uses a clever pattern with ``_init_ok`` flag:
.. code-block:: python
class ProxyElevator(ElevatorState):
_init_ok = False
def __init__(self, elevator_id: int, api_client: ElevatorAPIClient):
self._elevator_id = elevator_id
self._api_client = api_client
self._init_ok = True # Enable proxy behavior
def __getattribute__(self, name: str) -> Any:
# During initialization, use normal attribute access
if not name.startswith("_") and self._init_ok and name not in self.__class__.__dict__:
# Try to find as a method of this class
try:
self_attr = object.__getattribute__(self, name)
if callable(self_attr):
return object.__getattribute__(self, name)
except AttributeError:
pass
# Fetch fresh state and return attribute
elevator_state = self._get_elevator_state()
return elevator_state.__getattribute__(name)
else:
return object.__getattribute__(self, name)
def __setattr__(self, name: str, value: Any) -> None:
# Allow setting during initialization only
if not self._init_ok:
object.__setattr__(self, name, value)
else:
raise AttributeError(f"Cannot modify read-only attribute '{name}'")
This design:
1. Allows normal initialization of internal fields (``_elevator_id``, ``_api_client``)
2. Intercepts access to data attributes after initialization
3. Preserves access to class methods (like ``go_to_floor``)
4. Blocks all attribute modifications after initialization
Base Controller
---------------
The ``ElevatorController`` class in ``base_controller.py`` provides the framework for implementing control algorithms:
.. code-block:: python
from elevator_saga.client.base_controller import ElevatorController
from elevator_saga.client.proxy_models import ProxyElevator, ProxyFloor, ProxyPassenger
from typing import List
class MyController(ElevatorController):
def __init__(self):
super().__init__("http://127.0.0.1:8000", auto_run=True)
def on_init(self, elevators: List[ProxyElevator], floors: List[ProxyFloor]) -> None:
"""Called once at start with all elevators and floors"""
print(f"Initialized with {len(elevators)} elevators")
def on_passenger_call(self, passenger: ProxyPassenger, floor: ProxyFloor, direction: str) -> None:
"""Called when a passenger presses a button"""
print(f"Passenger {passenger.id} at floor {floor.floor} going {direction}")
def on_elevator_stopped(self, elevator: ProxyElevator, floor: ProxyFloor) -> None:
"""Called when elevator stops at a floor"""
print(f"Elevator {elevator.id} stopped at floor {floor.floor}")
# Implement your dispatch logic here
def on_elevator_idle(self, elevator: ProxyElevator) -> None:
"""Called when elevator becomes idle"""
# Send idle elevator somewhere useful
elevator.go_to_floor(0)
The controller provides these event handlers:
- ``on_init(elevators, floors)``: Initialization
- ``on_event_execute_start(tick, events, elevators, floors)``: Before processing tick events
- ``on_event_execute_end(tick, events, elevators, floors)``: After processing tick events
- ``on_passenger_call(passenger, floor, direction)``: Button press
- ``on_elevator_stopped(elevator, floor)``: Elevator arrival
- ``on_elevator_idle(elevator)``: Elevator becomes idle
- ``on_passenger_board(elevator, passenger)``: Passenger boards
- ``on_passenger_alight(elevator, passenger, floor)``: Passenger alights
- ``on_elevator_passing_floor(elevator, floor, direction)``: Elevator passes floor
- ``on_elevator_approaching(elevator, floor, direction)``: Elevator about to arrive
Complete Example
----------------
Here's a simple controller that sends idle elevators to the ground floor:
.. code-block:: python
#!/usr/bin/env python3
from typing import List
from elevator_saga.client.base_controller import ElevatorController
from elevator_saga.client.proxy_models import ProxyElevator, ProxyFloor, ProxyPassenger
class SimpleController(ElevatorController):
def __init__(self):
super().__init__("http://127.0.0.1:8000", auto_run=True)
self.pending_calls = []
def on_init(self, elevators: List[ProxyElevator], floors: List[ProxyFloor]) -> None:
print(f"Controlling {len(elevators)} elevators in {len(floors)}-floor building")
def on_passenger_call(self, passenger: ProxyPassenger, floor: ProxyFloor, direction: str) -> None:
print(f"Call from floor {floor.floor}, direction {direction}")
self.pending_calls.append((floor.floor, direction))
# Dispatch nearest idle elevator
self._dispatch_to_call(floor.floor)
def on_elevator_idle(self, elevator: ProxyElevator) -> None:
if self.pending_calls:
floor, direction = self.pending_calls.pop(0)
elevator.go_to_floor(floor)
else:
# No calls, return to ground floor
elevator.go_to_floor(0)
def on_elevator_stopped(self, elevator: ProxyElevator, floor: ProxyFloor) -> None:
print(f"Elevator {elevator.id} at floor {floor.floor}")
print(f" Passengers on board: {len(elevator.passengers)}")
print(f" Waiting at floor: {floor.total_waiting}")
def _dispatch_to_call(self, floor: int) -> None:
# Find nearest idle elevator and send it
# (Simplified - real implementation would be more sophisticated)
pass
if __name__ == "__main__":
controller = SimpleController()
controller.start()
Benefits of Proxy Architecture
-------------------------------
1. **Type Safety**: IDE autocomplete and type checking work perfectly
2. **Always Fresh**: No need to manually refresh state
3. **Clean API**: Access remote state as if it were local
4. **Read-Only Safety**: Cannot accidentally corrupt server state
5. **Separation of Concerns**: State management handled by proxies, logic in controller
6. **Testability**: Can mock API client for unit tests
Next Steps
----------
- See :doc:`communication` for details on the HTTP API
- See :doc:`events` for understanding the event-driven simulation
- Check ``client_examples/bus_example.py`` for a complete implementation

527
docs/communication.rst Normal file
View File

@@ -0,0 +1,527 @@
HTTP Communication Architecture
================================
Elevator Saga uses a **client-server architecture** with HTTP-based communication. The server manages the simulation state and physics, while clients send commands and query state over HTTP.
Architecture Overview
---------------------
.. code-block:: text
┌─────────────────┐ ┌─────────────────┐
│ Client │ │ Server │
│ (Controller) │ │ (Simulator) │
├─────────────────┤ ├─────────────────┤
│ base_controller │◄──── HTTP ────────►│ simulator.py │
│ proxy_models │ │ Flask Routes │
│ api_client │ │ │
└─────────────────┘ └─────────────────┘
│ │
│ GET /api/state │
│ POST /api/step │
│ POST /api/elevators/:id/go_to_floor │
│ │
└───────────────────────────────────────┘
The separation provides:
- **Modularity**: Server and client can be developed independently
- **Multiple Clients**: Multiple controllers can compete/cooperate
- **Language Flexibility**: Clients can be written in any language
- **Network Deployment**: Server and client can run on different machines
Server Side: Simulator
----------------------
The server is implemented in ``elevator_saga/server/simulator.py`` using **Flask** as the HTTP framework.
API Endpoints
~~~~~~~~~~~~~
**GET /api/state**
Returns complete simulation state:
.. code-block:: python
@app.route("/api/state", methods=["GET"])
def get_state() -> Response:
try:
state = simulation.get_state()
return json_response(state)
except Exception as e:
return json_response({"error": str(e)}, 500)
Response format:
.. code-block:: json
{
"tick": 42,
"elevators": [
{
"id": 0,
"position": {"current_floor": 2, "target_floor": 5, "floor_up_position": 3},
"passengers": [101, 102],
"max_capacity": 10,
"run_status": "constant_speed",
"..."
}
],
"floors": [
{"floor": 0, "up_queue": [103], "down_queue": []},
"..."
],
"passengers": {
"101": {"id": 101, "origin": 0, "destination": 5, "..."}
},
"metrics": {
"done": 50,
"total": 100,
"avg_wait": 15.2,
"p95_wait": 30.0,
"avg_system": 25.5,
"p95_system": 45.0
}
}
**POST /api/step**
Advances simulation by specified number of ticks:
.. code-block:: python
@app.route("/api/step", methods=["POST"])
def step_simulation() -> Response:
try:
data = request.get_json() or {}
ticks = data.get("ticks", 1)
events = simulation.step(ticks)
return json_response({
"tick": simulation.tick,
"events": events,
})
except Exception as e:
return json_response({"error": str(e)}, 500)
Request body:
.. code-block:: json
{"ticks": 1}
Response:
.. code-block:: json
{
"tick": 43,
"events": [
{
"tick": 43,
"type": "stopped_at_floor",
"data": {"elevator": 0, "floor": 5, "reason": "move_reached"}
}
]
}
**POST /api/elevators/:id/go_to_floor**
Commands an elevator to go to a floor:
.. code-block:: python
@app.route("/api/elevators/<int:elevator_id>/go_to_floor", methods=["POST"])
def elevator_go_to_floor(elevator_id: int) -> Response:
try:
data = request.get_json() or {}
floor = data["floor"]
immediate = data.get("immediate", False)
simulation.elevator_go_to_floor(elevator_id, floor, immediate)
return json_response({"success": True})
except Exception as e:
return json_response({"error": str(e)}, 500)
Request body:
.. code-block:: json
{"floor": 5, "immediate": false}
- ``immediate=false``: Set as next target after current destination
- ``immediate=true``: Change target immediately (cancels current target)
**POST /api/reset**
Resets simulation to initial state:
.. code-block:: python
@app.route("/api/reset", methods=["POST"])
def reset_simulation() -> Response:
try:
simulation.reset()
return json_response({"success": True})
except Exception as e:
return json_response({"error": str(e)}, 500)
**POST /api/traffic/next**
Loads next traffic scenario:
.. code-block:: python
@app.route("/api/traffic/next", methods=["POST"])
def next_traffic_round() -> Response:
try:
full_reset = request.get_json().get("full_reset", False)
success = simulation.next_traffic_round(full_reset)
if success:
return json_response({"success": True})
else:
return json_response({"success": False, "error": "No more scenarios"}, 400)
except Exception as e:
return json_response({"error": str(e)}, 500)
**GET /api/traffic/info**
Gets current traffic scenario information:
.. code-block:: python
@app.route("/api/traffic/info", methods=["GET"])
def get_traffic_info() -> Response:
try:
info = simulation.get_traffic_info()
return json_response(info)
except Exception as e:
return json_response({"error": str(e)}, 500)
Response:
.. code-block:: json
{
"current_index": 0,
"total_files": 5,
"max_tick": 1000
}
Client Side: API Client
-----------------------
The client is implemented in ``elevator_saga/client/api_client.py`` using Python's built-in ``urllib`` library.
ElevatorAPIClient Class
~~~~~~~~~~~~~~~~~~~~~~~
.. code-block:: python
class ElevatorAPIClient:
"""Unified elevator API client"""
def __init__(self, base_url: str):
self.base_url = base_url.rstrip("/")
# Caching fields
self._cached_state: Optional[SimulationState] = None
self._cached_tick: int = -1
self._tick_processed: bool = False
State Caching Strategy
~~~~~~~~~~~~~~~~~~~~~~~
The API client implements **smart caching** to reduce HTTP requests:
.. code-block:: python
def get_state(self, force_reload: bool = False) -> SimulationState:
"""Get simulation state with caching"""
# Return cached state if valid
if not force_reload and self._cached_state is not None and not self._tick_processed:
return self._cached_state
# Fetch fresh state
response_data = self._send_get_request("/api/state")
# ... parse and create SimulationState ...
# Update cache
self._cached_state = simulation_state
self._cached_tick = simulation_state.tick
self._tick_processed = False # Mark as fresh
return simulation_state
def mark_tick_processed(self) -> None:
"""Mark current tick as processed, invalidating cache"""
self._tick_processed = True
**Cache Behavior**:
1. First ``get_state()`` call in a tick fetches from server
2. Subsequent calls within same tick return cached data
3. After ``step()`` is called, cache is invalidated
4. Next ``get_state()`` fetches fresh data
This provides:
- **Performance**: Minimize HTTP requests
- **Consistency**: All operations in a tick see same state
- **Freshness**: New tick always gets new state
Core API Methods
~~~~~~~~~~~~~~~~
**get_state(force_reload=False)**
Fetches current simulation state:
.. code-block:: python
def get_state(self, force_reload: bool = False) -> SimulationState:
if not force_reload and self._cached_state is not None and not self._tick_processed:
return self._cached_state
response_data = self._send_get_request("/api/state")
# Parse response into data models
elevators = [ElevatorState.from_dict(e) for e in response_data["elevators"]]
floors = [FloorState.from_dict(f) for f in response_data["floors"]]
# ... handle passengers and metrics ...
simulation_state = SimulationState(
tick=response_data["tick"],
elevators=elevators,
floors=floors,
passengers=passengers,
metrics=metrics,
events=[]
)
# Update cache
self._cached_state = simulation_state
self._cached_tick = simulation_state.tick
self._tick_processed = False
return simulation_state
**step(ticks=1)**
Advances simulation:
.. code-block:: python
def step(self, ticks: int = 1) -> StepResponse:
response_data = self._send_post_request("/api/step", {"ticks": ticks})
# Parse events
events = []
for event_data in response_data["events"]:
event_dict = event_data.copy()
if "type" in event_dict:
event_dict["type"] = EventType(event_dict["type"])
events.append(SimulationEvent.from_dict(event_dict))
return StepResponse(
success=True,
tick=response_data["tick"],
events=events
)
**go_to_floor(elevator_id, floor, immediate=False)**
Sends elevator to floor:
.. code-block:: python
def go_to_floor(self, elevator_id: int, floor: int, immediate: bool = False) -> bool:
command = GoToFloorCommand(
elevator_id=elevator_id,
floor=floor,
immediate=immediate
)
try:
response = self.send_elevator_command(command)
return response
except Exception as e:
debug_log(f"Go to floor failed: {e}")
return False
HTTP Request Implementation
~~~~~~~~~~~~~~~~~~~~~~~~~~~
The client uses ``urllib.request`` for HTTP communication:
**GET Request**:
.. code-block:: python
def _send_get_request(self, endpoint: str) -> Dict[str, Any]:
url = f"{self.base_url}{endpoint}"
try:
with urllib.request.urlopen(url, timeout=60) as response:
data = json.loads(response.read().decode("utf-8"))
return data
except urllib.error.URLError as e:
raise RuntimeError(f"GET {url} failed: {e}")
**POST Request**:
.. code-block:: python
def _send_post_request(self, endpoint: str, data: Dict[str, Any]) -> Dict[str, Any]:
url = f"{self.base_url}{endpoint}"
request_body = json.dumps(data).encode("utf-8")
req = urllib.request.Request(
url,
data=request_body,
headers={"Content-Type": "application/json"}
)
try:
with urllib.request.urlopen(req, timeout=600) as response:
response_data = json.loads(response.read().decode("utf-8"))
return response_data
except urllib.error.URLError as e:
raise RuntimeError(f"POST {url} failed: {e}")
Communication Flow
------------------
Typical communication sequence during one tick:
.. code-block:: text
Client Server
│ │
│ 1. GET /api/state │
├─────────────────────────────────────►│
│ ◄── SimulationState (cached) │
│ │
│ 2. Analyze state, make decisions │
│ │
│ 3. POST /api/elevators/0/go_to_floor│
├─────────────────────────────────────►│
│ ◄── {"success": true} │
│ │
│ 4. GET /api/state (from cache) │
│ No HTTP request! │
│ │
│ 5. POST /api/step │
├─────────────────────────────────────►│
│ Server processes tick │
│ - Moves elevators │
│ - Boards/alights passengers │
│ - Generates events │
│ ◄── {tick: 43, events: [...]} │
│ │
│ 6. Process events │
│ Cache invalidated │
│ │
│ 7. GET /api/state (fetches fresh) │
├─────────────────────────────────────►│
│ ◄── SimulationState │
│ │
└──────────────────────────────────────┘
Error Handling
--------------
Both client and server implement robust error handling:
**Server Side**:
.. code-block:: python
@app.route("/api/step", methods=["POST"])
def step_simulation() -> Response:
try:
# ... process request ...
return json_response(result)
except Exception as e:
return json_response({"error": str(e)}, 500)
**Client Side**:
.. code-block:: python
def go_to_floor(self, elevator_id: int, floor: int, immediate: bool = False) -> bool:
try:
response = self.send_elevator_command(command)
return response
except Exception as e:
debug_log(f"Go to floor failed: {e}")
return False
Thread Safety
-------------
The simulator uses a lock to ensure thread-safe access:
.. code-block:: python
class ElevatorSimulation:
def __init__(self, ...):
self.lock = threading.Lock()
def step(self, num_ticks: int = 1) -> List[SimulationEvent]:
with self.lock:
# ... process ticks ...
def get_state(self) -> SimulationStateResponse:
with self.lock:
# ... return state ...
This allows Flask to handle concurrent requests safely.
**Batch Commands**:
.. code-block:: python
# ❌ Bad - sequential commands
elevator1.go_to_floor(5)
time.sleep(0.1) # Wait for response
elevator2.go_to_floor(3)
# ✅ Good - issue commands quickly
elevator1.go_to_floor(5)
elevator2.go_to_floor(3)
# All commands received before next tick
**Cache Awareness**:
Use ``mark_tick_processed()`` to explicitly invalidate cache if needed, but normally the framework handles this automatically.
Testing the API
---------------
You can test the API directly using curl:
.. code-block:: bash
# Get state
curl http://127.0.0.1:8000/api/state
# Step simulation
curl -X POST http://127.0.0.1:8000/api/step \
-H "Content-Type: application/json" \
-d '{"ticks": 1}'
# Send elevator to floor
curl -X POST http://127.0.0.1:8000/api/elevators/0/go_to_floor \
-H "Content-Type: application/json" \
-d '{"floor": 5, "immediate": false}'
# Reset simulation
curl -X POST http://127.0.0.1:8000/api/reset \
-H "Content-Type: application/json" \
-d '{}'
Next Steps
----------
- See :doc:`events` for understanding how events drive the simulation
- See :doc:`client` for using the API through proxy models
- Check the source code for complete implementation details

59
docs/conf.py Normal file
View File

@@ -0,0 +1,59 @@
# Configuration file for the Sphinx documentation builder.
#
# For the full list of built-in configuration values, see the documentation:
# https://www.sphinx-doc.org/en/master/usage/configuration.html
# -- Project information -----------------------------------------------------
# https://www.sphinx-doc.org/en/master/usage/configuration.html#project-information
project = "Elevator Saga"
copyright = "2025, ZGCA-Forge Team"
author = "ZGCA-Forge Team"
release = "0.1.0"
# -- General configuration ---------------------------------------------------
# https://www.sphinx-doc.org/en/master/usage/configuration.html#general-configuration
extensions = [
"sphinx.ext.autodoc",
"sphinx.ext.napoleon",
"sphinx.ext.viewcode",
"sphinx.ext.intersphinx",
"sphinx.ext.todo",
"sphinx_rtd_theme",
]
templates_path = ["_templates"]
exclude_patterns = ["_build", "Thumbs.db", ".DS_Store"]
# -- Options for HTML output -------------------------------------------------
# https://www.sphinx-doc.org/en/master/usage/configuration.html#options-for-html-output
html_theme = "sphinx_rtd_theme"
html_static_path = ["_static"]
# -- Extension configuration -------------------------------------------------
# Napoleon settings
napoleon_google_docstring = True
napoleon_numpy_docstring = True
napoleon_include_init_with_doc = True
napoleon_include_private_with_doc = False
napoleon_include_special_with_doc = True
napoleon_use_admonition_for_examples = False
napoleon_use_admonition_for_notes = False
napoleon_use_admonition_for_references = False
napoleon_use_ivar = False
napoleon_use_param = True
napoleon_use_rtype = True
napoleon_preprocess_types = False
napoleon_type_aliases = None
napoleon_attr_annotations = True
# Intersphinx mapping
intersphinx_mapping = {
"python": ("https://docs.python.org/3", None),
}
# Todo extension
todo_include_todos = True

530
docs/events.rst Normal file
View File

@@ -0,0 +1,530 @@
Event-Driven Simulation and Tick-Based Execution
================================================
Elevator Saga uses an **event-driven, tick-based** discrete simulation model. The simulation progresses in discrete time steps (ticks), and events are generated to notify the controller about state changes.
Simulation Overview
-------------------
.. code-block:: text
┌──────────────────────────────────────────────────────────┐
│ Simulation Loop │
│ │
│ Tick N │
│ 1. Update elevator status (START_UP → CONSTANT_SPEED) │
│ 2. Process arrivals (new passengers) │
│ 3. Move elevators (physics simulation) │
│ 4. Process stops (boarding/alighting) │
│ 5. Generate events │
│ │
│ Events sent to client → Client processes → Commands │
│ │
│ Tick N+1 │
│ (repeat...) │
└──────────────────────────────────────────────────────────┘
Tick-Based Execution
--------------------
What is a Tick?
~~~~~~~~~~~~~~~
A **tick** is the fundamental unit of simulation time. Each tick represents one discrete time step where:
1. Physics is updated (elevators move)
2. State changes occur (passengers board/alight)
3. Events are generated
4. Controller receives events and makes decisions
Think of it like frames in a video game - the simulation updates at discrete intervals.
Tick Processing Flow
~~~~~~~~~~~~~~~~~~~~
In ``simulator.py``, the ``step()`` method processes ticks:
.. code-block:: python
def step(self, num_ticks: int = 1) -> List[SimulationEvent]:
"""Process one or more simulation ticks"""
with self.lock:
new_events = []
for _ in range(num_ticks):
self.state.tick += 1
tick_events = self._process_tick()
new_events.extend(tick_events)
# Force complete passengers if max duration reached
if self.tick >= self.max_duration_ticks:
completed_count = self.force_complete_remaining_passengers()
return new_events
Each ``_process_tick()`` executes the four-phase cycle:
.. code-block:: python
def _process_tick(self) -> List[SimulationEvent]:
"""Process one simulation tick"""
events_start = len(self.state.events)
# Phase 1: Update elevator status
self._update_elevator_status()
# Phase 2: Add new passengers from traffic queue
self._process_arrivals()
# Phase 3: Move elevators
self._move_elevators()
# Phase 4: Process elevator stops and passenger boarding/alighting
self._process_elevator_stops()
# Return events generated this tick
return self.state.events[events_start:]
Elevator State Machine
-----------------------
Elevators transition through states each tick:
.. code-block:: text
STOPPED ──(target set)──► START_UP ──(1 tick)──► CONSTANT_SPEED
(near target)
START_DOWN
(1 tick)
(arrived) STOPPED
State Transitions
~~~~~~~~~~~~~~~~~
**Phase 1: Update Elevator Status** (``_update_elevator_status()``):
.. code-block:: python
def _update_elevator_status(self) -> None:
"""Update elevator operational state"""
for elevator in self.elevators:
# If no direction, check for next target
if elevator.target_floor_direction == Direction.STOPPED:
if elevator.next_target_floor is not None:
self._set_elevator_target_floor(elevator, elevator.next_target_floor)
self._process_passenger_in()
elevator.next_target_floor = None
else:
continue
# Transition state machine
if elevator.run_status == ElevatorStatus.STOPPED:
# Start acceleration
elevator.run_status = ElevatorStatus.START_UP
elif elevator.run_status == ElevatorStatus.START_UP:
# Switch to constant speed after 1 tick
elevator.run_status = ElevatorStatus.CONSTANT_SPEED
**Important Notes**:
- ``START_UP`` = acceleration (not direction!)
- ``START_DOWN`` = deceleration (not direction!)
- Actual movement direction is ``target_floor_direction`` (UP/DOWN)
- State transitions happen **before** movement
Movement Physics
----------------
Speed by State
~~~~~~~~~~~~~~
Elevators move at different speeds depending on their state:
.. code-block:: python
def _move_elevators(self) -> None:
"""Move all elevators towards their destinations"""
for elevator in self.elevators:
# Determine speed based on state
movement_speed = 0
if elevator.run_status == ElevatorStatus.START_UP:
movement_speed = 1 # Accelerating: 0.1 floors/tick
elif elevator.run_status == ElevatorStatus.START_DOWN:
movement_speed = 1 # Decelerating: 0.1 floors/tick
elif elevator.run_status == ElevatorStatus.CONSTANT_SPEED:
movement_speed = 2 # Full speed: 0.2 floors/tick
if movement_speed == 0:
continue
# Apply movement in appropriate direction
if elevator.target_floor_direction == Direction.UP:
new_floor = elevator.position.floor_up_position_add(movement_speed)
elif elevator.target_floor_direction == Direction.DOWN:
new_floor = elevator.position.floor_up_position_add(-movement_speed)
Position System
~~~~~~~~~~~~~~~
Positions use a **10-unit sub-floor** system:
- ``current_floor = 2, floor_up_position = 0`` → exactly at floor 2
- ``current_floor = 2, floor_up_position = 5`` → halfway between floors 2 and 3
- ``current_floor = 2, floor_up_position = 10`` → advances to ``current_floor = 3, floor_up_position = 0``
This granularity allows smooth movement and precise deceleration timing.
Deceleration Logic
~~~~~~~~~~~~~~~~~~
Elevators must decelerate before stopping:
.. code-block:: python
def _should_start_deceleration(self, elevator: ElevatorState) -> bool:
"""Check if should start decelerating"""
distance = self._calculate_distance_to_target(elevator)
return distance == 1 # Start deceleration 1 position unit before target
# In _move_elevators():
if elevator.run_status == ElevatorStatus.CONSTANT_SPEED:
if self._should_start_deceleration(elevator):
elevator.run_status = ElevatorStatus.START_DOWN
This ensures elevators don't overshoot their target floor.
Event System
------------
Event Types
~~~~~~~~~~~
The simulation generates 8 types of events defined in ``EventType`` enum:
.. code-block:: python
class EventType(Enum):
UP_BUTTON_PRESSED = "up_button_pressed"
DOWN_BUTTON_PRESSED = "down_button_pressed"
PASSING_FLOOR = "passing_floor"
STOPPED_AT_FLOOR = "stopped_at_floor"
ELEVATOR_APPROACHING = "elevator_approaching"
IDLE = "idle"
PASSENGER_BOARD = "passenger_board"
PASSENGER_ALIGHT = "passenger_alight"
Event Generation
~~~~~~~~~~~~~~~~
Events are generated during tick processing:
**Passenger Arrival**:
.. code-block:: python
def _process_arrivals(self) -> None:
"""Process new passenger arrivals"""
while self.traffic_queue and self.traffic_queue[0].tick <= self.tick:
traffic_entry = self.traffic_queue.pop(0)
passenger = PassengerInfo(
id=traffic_entry.id,
origin=traffic_entry.origin,
destination=traffic_entry.destination,
arrive_tick=self.tick,
)
self.passengers[passenger.id] = passenger
if passenger.destination > passenger.origin:
self.floors[passenger.origin].up_queue.append(passenger.id)
# Generate UP_BUTTON_PRESSED event
self._emit_event(
EventType.UP_BUTTON_PRESSED,
{"floor": passenger.origin, "passenger": passenger.id}
)
else:
self.floors[passenger.origin].down_queue.append(passenger.id)
# Generate DOWN_BUTTON_PRESSED event
self._emit_event(
EventType.DOWN_BUTTON_PRESSED,
{"floor": passenger.origin, "passenger": passenger.id}
)
**Elevator Movement**:
.. code-block:: python
def _move_elevators(self) -> None:
for elevator in self.elevators:
# ... movement logic ...
# Passing a floor
if old_floor != new_floor and new_floor != target_floor:
self._emit_event(
EventType.PASSING_FLOOR,
{
"elevator": elevator.id,
"floor": new_floor,
"direction": elevator.target_floor_direction.value
}
)
# About to arrive (during deceleration)
if self._near_next_stop(elevator):
self._emit_event(
EventType.ELEVATOR_APPROACHING,
{
"elevator": elevator.id,
"floor": elevator.target_floor,
"direction": elevator.target_floor_direction.value
}
)
# Arrived at target
if target_floor == new_floor and elevator.position.floor_up_position == 0:
elevator.run_status = ElevatorStatus.STOPPED
self._emit_event(
EventType.STOPPED_AT_FLOOR,
{
"elevator": elevator.id,
"floor": new_floor,
"reason": "move_reached"
}
)
**Boarding and Alighting**:
.. code-block:: python
def _process_elevator_stops(self) -> None:
for elevator in self.elevators:
if elevator.run_status != ElevatorStatus.STOPPED:
continue
current_floor = elevator.current_floor
# Passengers alight
passengers_to_remove = []
for passenger_id in elevator.passengers:
passenger = self.passengers[passenger_id]
if passenger.destination == current_floor:
passenger.dropoff_tick = self.tick
passengers_to_remove.append(passenger_id)
for passenger_id in passengers_to_remove:
elevator.passengers.remove(passenger_id)
self._emit_event(
EventType.PASSENGER_ALIGHT,
{"elevator": elevator.id, "floor": current_floor, "passenger": passenger_id}
)
**Idle Detection**:
.. code-block:: python
# If elevator stopped with no direction, it's idle
if elevator.last_tick_direction == Direction.STOPPED:
self._emit_event(
EventType.IDLE,
{"elevator": elevator.id, "floor": current_floor}
)
Event Processing in Controller
-------------------------------
The ``ElevatorController`` base class automatically routes events to handler methods:
.. code-block:: python
class ElevatorController(ABC):
def _execute_events(self, events: List[SimulationEvent]) -> None:
"""Process events and route to handlers"""
for event in events:
if event.type == EventType.UP_BUTTON_PRESSED:
passenger_id = event.data["passenger"]
floor = self.floors[event.data["floor"]]
passenger = ProxyPassenger(passenger_id, self.api_client)
self.on_passenger_call(passenger, floor, "up")
elif event.type == EventType.DOWN_BUTTON_PRESSED:
passenger_id = event.data["passenger"]
floor = self.floors[event.data["floor"]]
passenger = ProxyPassenger(passenger_id, self.api_client)
self.on_passenger_call(passenger, floor, "down")
elif event.type == EventType.STOPPED_AT_FLOOR:
elevator = self.elevators[event.data["elevator"]]
floor = self.floors[event.data["floor"]]
self.on_elevator_stopped(elevator, floor)
elif event.type == EventType.IDLE:
elevator = self.elevators[event.data["elevator"]]
self.on_elevator_idle(elevator)
# ... other event types ...
Control Flow: Bus Example
--------------------------
The ``bus_example.py`` demonstrates a simple "bus route" algorithm:
.. code-block:: python
class ElevatorBusExampleController(ElevatorController):
def __init__(self):
super().__init__("http://127.0.0.1:8000", True)
self.all_passengers = []
self.max_floor = 0
def on_init(self, elevators: List[ProxyElevator], floors: List[ProxyFloor]) -> None:
"""Initialize elevators to starting positions"""
self.max_floor = floors[-1].floor
self.floors = floors
for i, elevator in enumerate(elevators):
# Distribute elevators evenly across floors
target_floor = (i * (len(floors) - 1)) // len(elevators)
elevator.go_to_floor(target_floor, immediate=True)
def on_event_execute_start(self, tick: int, events: List[SimulationEvent],
elevators: List[ProxyElevator], floors: List[ProxyFloor]) -> None:
"""Print state before processing events"""
print(f"Tick {tick}: Processing {len(events)} events {[e.type.value for e in events]}")
for elevator in elevators:
print(
f"\t{elevator.id}[{elevator.target_floor_direction.value},"
f"{elevator.current_floor_float}/{elevator.target_floor}]"
+ "👦" * len(elevator.passengers),
end=""
)
print()
def on_elevator_stopped(self, elevator: ProxyElevator, floor: ProxyFloor) -> None:
"""Implement bus route logic"""
print(f"🛑 Elevator E{elevator.id} stopped at F{floor.floor}")
# Bus algorithm: go up to top, then down to bottom, repeat
if elevator.last_tick_direction == Direction.UP and elevator.current_floor == self.max_floor:
# At top, start going down
elevator.go_to_floor(elevator.current_floor - 1)
elif elevator.last_tick_direction == Direction.DOWN and elevator.current_floor == 0:
# At bottom, start going up
elevator.go_to_floor(elevator.current_floor + 1)
elif elevator.last_tick_direction == Direction.UP:
# Continue upward
elevator.go_to_floor(elevator.current_floor + 1)
elif elevator.last_tick_direction == Direction.DOWN:
# Continue downward
elevator.go_to_floor(elevator.current_floor - 1)
def on_elevator_idle(self, elevator: ProxyElevator) -> None:
"""Send idle elevator to floor 1"""
elevator.go_to_floor(1)
Execution Sequence
~~~~~~~~~~~~~~~~~~
Here's what happens in a typical tick:
.. code-block:: text
Server: Tick 42
Phase 1: Update status
- Elevator 0: STOPPED → START_UP (has target)
Phase 2: Process arrivals
- Passenger 101 arrives at floor 0, going to floor 5
- Event: UP_BUTTON_PRESSED
Phase 3: Move elevators
- Elevator 0: floor 2.0 → 2.1 (accelerating)
Phase 4: Process stops
- (no stops this tick)
Events: [UP_BUTTON_PRESSED, PASSING_FLOOR]
Client: Receive events
on_event_execute_start(tick=42, events=[...])
- Print "Tick 42: Processing 2 events"
_execute_events():
- UP_BUTTON_PRESSED → on_passenger_call()
→ Controller decides which elevator to send
- PASSING_FLOOR → on_elevator_passing_floor()
on_event_execute_end(tick=42, events=[...])
Client: Send commands
- elevator.go_to_floor(0) → POST /api/elevators/0/go_to_floor
Client: Step simulation
- POST /api/step → Server processes tick 43
Key Timing Concepts
-------------------
Immediate vs. Queued
~~~~~~~~~~~~~~~~~~~~
.. code-block:: python
# Queued (default): Wait until current target reached
elevator.go_to_floor(5, immediate=False)
# ← Sets elevator.next_target_floor = 5
# ← Processed when current_floor == target_floor
# Immediate: Change target right away
elevator.go_to_floor(5, immediate=True)
# ← Sets elevator.position.target_floor = 5 immediately
# ← May interrupt current journey
Use ``immediate=True`` for emergency redirects, ``immediate=False`` (default) for normal operation.
Performance Metrics
-------------------
Metrics are calculated from passenger data:
.. code-block:: python
def _calculate_metrics(self) -> MetricsResponse:
"""Calculate performance metrics"""
completed = [p for p in self.state.passengers.values()
if p.status == PassengerStatus.COMPLETED]
wait_times = [float(p.wait_time) for p in completed]
system_times = [float(p.system_time) for p in completed]
return MetricsResponse(
done=len(completed),
total=len(self.state.passengers),
avg_wait=sum(wait_times) / len(wait_times) if wait_times else 0,
p95_wait=percentile(wait_times, 95),
avg_system=sum(system_times) / len(system_times) if system_times else 0,
p95_system=percentile(system_times, 95),
)
Key metrics:
- **Wait time**: ``pickup_tick - arrive_tick`` (how long passenger waited)
- **System time**: ``dropoff_tick - arrive_tick`` (total time in system)
- **P95**: 95th percentile (worst-case for most passengers)
Summary
-------
The event-driven, tick-based architecture provides:
- **Deterministic**: Same inputs always produce same results
- **Testable**: Easy to create test scenarios with traffic files
- **Debuggable**: Clear event trail shows what happened when
- **Flexible**: Easy to implement different dispatch algorithms
- **Scalable**: Can simulate large buildings and many passengers
Next Steps
----------
- Study ``bus_example.py`` for a complete working example
- Implement your own controller by extending ``ElevatorController``
- Experiment with different dispatch algorithms
- Analyze performance metrics to optimize your approach

106
docs/index.rst Normal file
View File

@@ -0,0 +1,106 @@
Welcome to Elevator Saga's Documentation!
==========================================
.. image:: https://badge.fury.io/py/elevator-py.svg
:target: https://badge.fury.io/py/elevator-py
:alt: PyPI version
.. image:: https://img.shields.io/pypi/pyversions/elevator-py.svg
:target: https://pypi.org/project/elevator-py/
:alt: Python versions
.. image:: https://github.com/ZGCA-Forge/Elevator/actions/workflows/ci.yml/badge.svg
:target: https://github.com/ZGCA-Forge/Elevator/actions
:alt: Build Status
.. image:: https://img.shields.io/github/stars/ZGCA-Forge/Elevator.svg?style=social&label=Star
:target: https://github.com/ZGCA-Forge/Elevator
:alt: GitHub stars
Elevator Saga is a Python implementation of an elevator `simulation game <https://play.elevatorsaga.com/>`_ with an event-driven architecture. Design and optimize elevator control algorithms to efficiently transport passengers in buildings.
Features
--------
🏢 **Realistic Simulation**: Physics-based elevator movement with acceleration, deceleration, and realistic timing
🎮 **Event-Driven Architecture**: React to various events such as button presses, elevator arrivals, and passenger boarding
🔌 **Client-Server Model**: Separate simulation server from control logic for clean architecture
📊 **Performance Metrics**: Track wait times, system times, and completion rates
🎯 **Flexible Control**: Implement your own algorithms using a simple controller interface
Installation
------------
Basic Installation
~~~~~~~~~~~~~~~~~~
.. code-block:: bash
pip install elevator-py
Quick Start
-----------
Running the Simulation
~~~~~~~~~~~~~~~~~~~~~~
**Terminal #1: Start the backend simulator**
.. code-block:: bash
python -m elevator_saga.server.simulator
**Terminal #2: Start your controller**
.. code-block:: bash
python -m elevator_saga.client_examples.bus_example
Architecture Overview
---------------------
Elevator Saga follows a **client-server architecture**:
- **Server** (`simulator.py`): Manages the simulation state, physics, and event generation
- **Client** (`base_controller.py`): Implements control algorithms and reacts to events
- **Communication** (`api_client.py`): HTTP-based API for state queries and commands
- **Data Models** (`models.py`): Unified data structures shared between client and server
Contents
--------
.. toctree::
:maxdepth: 2
:caption: User Guide
models
client
communication
events
.. toctree::
:maxdepth: 1
:caption: API Reference
api/modules
Contributing
------------
Contributions are welcome! Please feel free to submit a Pull Request.
License
-------
This project is licensed under MIT License - see the LICENSE file for details.
Indices and tables
==================
* :ref:`genindex`
* :ref:`modindex`
* :ref:`search`

347
docs/models.rst Normal file
View File

@@ -0,0 +1,347 @@
Data Models
===========
The Elevator Saga system uses a unified data model architecture defined in ``elevator_saga/core/models.py``. These models ensure type consistency and serialization between the client and server components.
Overview
--------
All data models inherit from ``SerializableModel``, which provides:
- **to_dict()**: Convert model to dictionary
- **to_json()**: Convert model to JSON string
- **from_dict()**: Create model instance from dictionary
- **from_json()**: Create model instance from JSON string
This unified serialization approach ensures seamless data exchange over HTTP between client and server.
Core Enumerations
-----------------
Direction
~~~~~~~~~
Represents the direction of elevator movement or passenger travel:
.. code-block:: python
class Direction(Enum):
UP = "up" # Moving upward
DOWN = "down" # Moving downward
STOPPED = "stopped" # Not moving
ElevatorStatus
~~~~~~~~~~~~~~
Represents the elevator's operational state in the state machine:
.. code-block:: python
class ElevatorStatus(Enum):
START_UP = "start_up" # Acceleration phase
START_DOWN = "start_down" # Deceleration phase
CONSTANT_SPEED = "constant_speed" # Constant speed phase
STOPPED = "stopped" # Stopped at floor
**Important**: ``START_UP`` and ``START_DOWN`` refer to **acceleration/deceleration states**, not movement direction. The actual movement direction is determined by the ``target_floor_direction`` property.
State Machine Transition:
.. code-block:: text
STOPPED → START_UP → CONSTANT_SPEED → START_DOWN → STOPPED
1 tick 1 tick N ticks 1 tick
PassengerStatus
~~~~~~~~~~~~~~~
Represents the passenger's current state:
.. code-block:: python
class PassengerStatus(Enum):
WAITING = "waiting" # Waiting at origin floor
IN_ELEVATOR = "in_elevator" # Inside an elevator
COMPLETED = "completed" # Reached destination
CANCELLED = "cancelled" # Cancelled (unused)
EventType
~~~~~~~~~
Defines all possible simulation events:
.. code-block:: python
class EventType(Enum):
UP_BUTTON_PRESSED = "up_button_pressed"
DOWN_BUTTON_PRESSED = "down_button_pressed"
PASSING_FLOOR = "passing_floor"
STOPPED_AT_FLOOR = "stopped_at_floor"
ELEVATOR_APPROACHING = "elevator_approaching"
IDLE = "idle"
PASSENGER_BOARD = "passenger_board"
PASSENGER_ALIGHT = "passenger_alight"
Core Data Models
----------------
Position
~~~~~~~~
Represents elevator position with sub-floor granularity:
.. code-block:: python
@dataclass
class Position(SerializableModel):
current_floor: int = 0 # Current floor number
target_floor: int = 0 # Target floor number
floor_up_position: int = 0 # Position within floor (0-9)
- **floor_up_position**: Represents position between floors with 10 units per floor
- **current_floor_float**: Returns floating-point floor position (e.g., 2.5 = halfway between floors 2 and 3)
Example:
.. code-block:: python
position = Position(current_floor=2, floor_up_position=5)
print(position.current_floor_float) # 2.5
ElevatorState
~~~~~~~~~~~~~
Complete state information for an elevator:
.. code-block:: python
@dataclass
class ElevatorState(SerializableModel):
id: int
position: Position
next_target_floor: Optional[int] = None
passengers: List[int] = [] # Passenger IDs
max_capacity: int = 10
speed_pre_tick: float = 0.5
run_status: ElevatorStatus = ElevatorStatus.STOPPED
last_tick_direction: Direction = Direction.STOPPED
indicators: ElevatorIndicators = field(default_factory=ElevatorIndicators)
passenger_destinations: Dict[int, int] = {} # passenger_id -> floor
energy_consumed: float = 0.0
last_update_tick: int = 0
Key Properties:
- ``current_floor``: Integer floor number
- ``current_floor_float``: Precise position including sub-floor
- ``target_floor``: Destination floor
- ``target_floor_direction``: Direction to target (UP/DOWN/STOPPED)
- ``is_idle``: Whether elevator is stopped
- ``is_full``: Whether elevator is at capacity
- ``is_running``: Whether elevator is in motion
- ``pressed_floors``: List of destination floors for current passengers
- ``load_factor``: Current load as fraction of capacity (0.0 to 1.0)
FloorState
~~~~~~~~~~
State information for a building floor:
.. code-block:: python
@dataclass
class FloorState(SerializableModel):
floor: int
up_queue: List[int] = [] # Passenger IDs waiting to go up
down_queue: List[int] = [] # Passenger IDs waiting to go down
Properties:
- ``has_waiting_passengers``: Whether any passengers are waiting
- ``total_waiting``: Total number of waiting passengers
PassengerInfo
~~~~~~~~~~~~~
Complete information about a passenger:
.. code-block:: python
@dataclass
class PassengerInfo(SerializableModel):
id: int
origin: int # Starting floor
destination: int # Target floor
arrive_tick: int # When passenger appeared
pickup_tick: int = 0 # When passenger boarded elevator
dropoff_tick: int = 0 # When passenger reached destination
elevator_id: Optional[int] = None
Properties:
- ``status``: Current PassengerStatus
- ``wait_time``: Ticks waited before boarding
- ``system_time``: Total ticks in system (arrive to dropoff)
- ``travel_direction``: UP/DOWN based on origin and destination
SimulationState
~~~~~~~~~~~~~~~
Complete state of the simulation:
.. code-block:: python
@dataclass
class SimulationState(SerializableModel):
tick: int
elevators: List[ElevatorState]
floors: List[FloorState]
passengers: Dict[int, PassengerInfo]
metrics: PerformanceMetrics
events: List[SimulationEvent]
Helper Methods:
- ``get_elevator_by_id(id)``: Find elevator by ID
- ``get_floor_by_number(number)``: Find floor by number
- ``get_passengers_by_status(status)``: Filter passengers by status
- ``add_event(type, data)``: Add new event to queue
Traffic and Configuration
-------------------------
TrafficEntry
~~~~~~~~~~~~
Defines a single passenger arrival:
.. code-block:: python
@dataclass
class TrafficEntry(SerializableModel):
id: int
origin: int
destination: int
tick: int # When passenger arrives
TrafficPattern
~~~~~~~~~~~~~~
Collection of traffic entries defining a test scenario:
.. code-block:: python
@dataclass
class TrafficPattern(SerializableModel):
name: str
description: str
entries: List[TrafficEntry]
metadata: Dict[str, Any]
Properties:
- ``total_passengers``: Number of passengers in pattern
- ``duration``: Tick when last passenger arrives
Performance Metrics
-------------------
PerformanceMetrics
~~~~~~~~~~~~~~~~~~
Tracks simulation performance:
.. code-block:: python
@dataclass
class PerformanceMetrics(SerializableModel):
completed_passengers: int = 0
total_passengers: int = 0
average_wait_time: float = 0.0
p95_wait_time: float = 0.0 # 95th percentile
average_system_time: float = 0.0
p95_system_time: float = 0.0 # 95th percentile
Properties:
- ``completion_rate``: Fraction of passengers completed (0.0 to 1.0)
API Models
----------
The models also include HTTP API request/response structures:
- ``APIRequest``: Base request with ID and timestamp
- ``APIResponse``: Base response with success flag
- ``StepRequest/StepResponse``: Advance simulation time
- ``StateRequest``: Query simulation state
- ``ElevatorCommand``: Send command to elevator
- ``GoToFloorCommand``: Specific command to move elevator
Example Usage
-------------
Creating a Simulation State
~~~~~~~~~~~~~~~~~~~~~~~~~~~~
.. code-block:: python
from elevator_saga.core.models import (
create_empty_simulation_state,
ElevatorState,
Position,
)
# Create a building with 3 elevators, 10 floors, capacity 8
state = create_empty_simulation_state(
elevators=3,
floors=10,
max_capacity=8
)
# Access elevator state
elevator = state.elevators[0]
print(f"Elevator {elevator.id} at floor {elevator.current_floor}")
Working with Traffic Patterns
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
.. code-block:: python
from elevator_saga.core.models import (
create_simple_traffic_pattern,
TrafficPattern,
)
# Create traffic pattern: (origin, destination, tick)
pattern = create_simple_traffic_pattern(
name="morning_rush",
passengers=[
(0, 5, 10), # Floor 0→5 at tick 10
(0, 8, 15), # Floor 0→8 at tick 15
(2, 0, 20), # Floor 2→0 at tick 20
]
)
print(f"Pattern has {pattern.total_passengers} passengers")
print(f"Duration: {pattern.duration} ticks")
Serialization
~~~~~~~~~~~~~
All models support JSON serialization:
.. code-block:: python
# Serialize to JSON
elevator = state.elevators[0]
json_str = elevator.to_json()
# Deserialize from JSON
restored = ElevatorState.from_json(json_str)
# Or use dictionaries
data = elevator.to_dict()
restored = ElevatorState.from_dict(data)
This enables seamless transmission over HTTP between client and server.

2
docs/requirements.txt Normal file
View File

@@ -0,0 +1,2 @@
sphinx>=5.0.0
sphinx-rtd-theme>=1.0.0

View File

@@ -6,5 +6,5 @@ A Python implementation of the Elevator Saga game with event-driven architecture
realistic elevator dispatch algorithm development and testing.
"""
__version__ = "1.0.0"
__version__ = "0.0.4"
__author__ = "ZGCA Team"

View File

@@ -134,12 +134,14 @@ class ElevatorAPIClient:
def send_elevator_command(self, command: Union[GoToFloorCommand]) -> bool:
"""发送电梯命令"""
endpoint = self._get_elevator_endpoint(command)
debug_log(f"Sending elevator command: {command.command_type} to elevator {command.elevator_id} To:F{command.floor}")
debug_log(
f"Sending elevator command: {command.command_type} to elevator {command.elevator_id} To:F{command.floor}"
)
response_data = self._send_post_request(endpoint, command.parameters)
if response_data.get("success"):
return response_data["success"]
return bool(response_data["success"])
else:
raise RuntimeError(f"Command failed: {response_data.get('error_message')}")
@@ -168,7 +170,7 @@ class ElevatorAPIClient:
try:
with urllib.request.urlopen(url, timeout=60) as response:
data = json.loads(response.read().decode("utf-8"))
data: Dict[str, Any] = json.loads(response.read().decode("utf-8"))
# debug_log(f"GET {url} -> {response.status}")
return data
except urllib.error.URLError as e:
@@ -178,7 +180,7 @@ class ElevatorAPIClient:
"""重置模拟"""
try:
response_data = self._send_post_request("/api/reset", {})
success = response_data.get("success", False)
success = bool(response_data.get("success", False))
if success:
# 清空缓存,因为状态已重置
self._cached_state = None
@@ -190,11 +192,11 @@ class ElevatorAPIClient:
debug_log(f"Reset failed: {e}")
return False
def next_traffic_round(self, full_reset = False) -> bool:
def next_traffic_round(self, full_reset: bool = False) -> bool:
"""切换到下一个流量文件"""
try:
response_data = self._send_post_request("/api/traffic/next", {"full_reset": full_reset})
success = response_data.get("success", False)
success = bool(response_data.get("success", False))
if success:
# 清空缓存,因为流量文件已切换,状态会改变
self._cached_state = None
@@ -230,7 +232,7 @@ class ElevatorAPIClient:
try:
with urllib.request.urlopen(req, timeout=600) as response:
response_data = json.loads(response.read().decode("utf-8"))
response_data: Dict[str, Any] = json.loads(response.read().decode("utf-8"))
# debug_log(f"POST {url} -> {response.status}")
return response_data
except urllib.error.URLError as e:

View File

@@ -44,7 +44,7 @@ class ElevatorController(ABC):
self.api_client = ElevatorAPIClient(server_url)
@abstractmethod
def on_init(self, elevators: List[Any], floors: List[Any]):
def on_init(self, elevators: List[Any], floors: List[Any]) -> None:
"""
算法初始化方法 - 必须由子类实现
@@ -55,7 +55,7 @@ class ElevatorController(ABC):
pass
@abstractmethod
def on_event_execute_start(self, tick: int, events: List[Any], elevators: List[Any], floors: List[Any]):
def on_event_execute_start(self, tick: int, events: List[Any], elevators: List[Any], floors: List[Any]) -> None:
"""
事件执行前的回调 - 必须由子类实现
@@ -68,7 +68,7 @@ class ElevatorController(ABC):
pass
@abstractmethod
def on_event_execute_end(self, tick: int, events: List[Any], elevators: List[Any], floors: List[Any]):
def on_event_execute_end(self, tick: int, events: List[Any], elevators: List[Any], floors: List[Any]) -> None:
"""
事件执行后的回调 - 必须由子类实现
@@ -80,20 +80,20 @@ class ElevatorController(ABC):
"""
pass
def on_start(self):
def on_start(self) -> None:
"""
算法启动前的回调 - 可选实现
"""
print(f"启动 {self.__class__.__name__} 算法")
def on_stop(self):
def on_stop(self) -> None:
"""
算法停止后的回调 - 可选实现
"""
print(f"停止 {self.__class__.__name__} 算法")
@abstractmethod
def on_passenger_call(self, passenger: ProxyPassenger, floor: ProxyFloor, direction: str):
def on_passenger_call(self, passenger: ProxyPassenger, floor: ProxyFloor, direction: str) -> None:
"""
乘客呼叫时的回调 - 可选实现
@@ -104,7 +104,7 @@ class ElevatorController(ABC):
pass
@abstractmethod
def on_elevator_idle(self, elevator: ProxyElevator):
def on_elevator_idle(self, elevator: ProxyElevator) -> None:
"""
电梯空闲时的回调 - 可选实现
@@ -114,7 +114,7 @@ class ElevatorController(ABC):
pass
@abstractmethod
def on_elevator_stopped(self, elevator: ProxyElevator, floor: ProxyFloor):
def on_elevator_stopped(self, elevator: ProxyElevator, floor: ProxyFloor) -> None:
"""
电梯停靠时的回调 - 可选实现
@@ -125,7 +125,7 @@ class ElevatorController(ABC):
pass
@abstractmethod
def on_passenger_board(self, elevator: ProxyElevator, passenger: ProxyPassenger):
def on_passenger_board(self, elevator: ProxyElevator, passenger: ProxyPassenger) -> None:
"""
乘客上梯时的回调 - 可选实现
@@ -136,7 +136,7 @@ class ElevatorController(ABC):
pass
@abstractmethod
def on_passenger_alight(self, elevator: ProxyElevator, passenger: ProxyPassenger, floor: ProxyFloor):
def on_passenger_alight(self, elevator: ProxyElevator, passenger: ProxyPassenger, floor: ProxyFloor) -> None:
"""
乘客下车时的回调 - 可选实现
@@ -148,7 +148,7 @@ class ElevatorController(ABC):
pass
@abstractmethod
def on_elevator_passing_floor(self, elevator: ProxyElevator, floor: ProxyFloor, direction: str):
def on_elevator_passing_floor(self, elevator: ProxyElevator, floor: ProxyFloor, direction: str) -> None:
"""
电梯经过楼层时的回调 - 可选实现
@@ -160,7 +160,7 @@ class ElevatorController(ABC):
pass
@abstractmethod
def on_elevator_approaching(self, elevator: ProxyElevator, floor: ProxyFloor, direction: str):
def on_elevator_approaching(self, elevator: ProxyElevator, floor: ProxyFloor, direction: str) -> None:
"""
电梯即将到达时的回调 - 可选实现
@@ -171,7 +171,7 @@ class ElevatorController(ABC):
"""
pass
def _internal_init(self, elevators: List[Any], floors: List[Any]):
def _internal_init(self, elevators: List[Any], floors: List[Any]) -> None:
"""内部初始化方法"""
self.elevators = elevators
self.floors = floors
@@ -180,7 +180,7 @@ class ElevatorController(ABC):
# 调用用户的初始化方法
self.on_init(elevators, floors)
def start(self):
def start(self) -> None:
"""
启动控制器
"""
@@ -198,12 +198,12 @@ class ElevatorController(ABC):
self.is_running = False
self.on_stop()
def stop(self):
def stop(self) -> None:
"""停止控制器"""
self.is_running = False
print(f"停止 {self.__class__.__name__}")
def on_simulation_complete(self, final_state: Dict[str, Any]):
def on_simulation_complete(self, final_state: Dict[str, Any]) -> None:
"""
模拟完成时的回调 - 可选实现
@@ -212,7 +212,7 @@ class ElevatorController(ABC):
"""
pass
def _run_event_driven_simulation(self):
def _run_event_driven_simulation(self) -> None:
"""运行事件驱动的模拟"""
try:
# 获取初始状态并初始化默认从0开始
@@ -304,7 +304,7 @@ class ElevatorController(ABC):
try:
traffic_info = self.api_client.get_traffic_info()
if traffic_info:
self.current_traffic_max_tick = traffic_info["max_tick"]
self.current_traffic_max_tick = int(traffic_info["max_tick"])
debug_log(f"Updated traffic info - max_tick: {self.current_traffic_max_tick}")
else:
debug_log("Failed to get traffic info")
@@ -313,7 +313,7 @@ class ElevatorController(ABC):
debug_log(f"Error updating traffic info: {e}")
self.current_traffic_max_tick = 0
def _handle_single_event(self, event: SimulationEvent):
def _handle_single_event(self, event: SimulationEvent) -> None:
"""处理单个事件"""
if event.type == EventType.UP_BUTTON_PRESSED:
floor_id = event.data["floor"]
@@ -385,7 +385,7 @@ class ElevatorController(ABC):
floor_proxy = ProxyFloor(floor_id, self.api_client)
self.on_passenger_alight(elevator_proxy, passenger_proxy, floor_proxy)
def _reset_and_reinit(self):
def _reset_and_reinit(self) -> None:
"""重置并重新初始化"""
try:
# 重置服务器状态

View File

@@ -22,6 +22,8 @@ class ProxyFloor(FloorState):
"""获取 FloorState 实例"""
state = self._api_client.get_state()
floor_data = next((f for f in state.floors if f.floor == self._floor_id), None)
if floor_data is None:
raise ValueError(f"Floor {self._floor_id} not found in state")
return floor_data
def __getattribute__(self, name: str) -> Any:
@@ -66,6 +68,8 @@ class ProxyElevator(ElevatorState):
# 获取当前状态
state = self._api_client.get_state()
elevator_data = next((e for e in state.elevators if e.id == self._elevator_id), None)
if elevator_data is None:
raise ValueError(f"Elevator {self._elevator_id} not found in state")
return elevator_data
def __getattribute__(self, name: str) -> Any:
@@ -113,6 +117,8 @@ class ProxyPassenger(PassengerInfo):
"""获取 PassengerInfo 实例"""
state = self._api_client.get_state()
passenger_data = state.passengers.get(self._passenger_id)
if passenger_data is None:
raise ValueError(f"Passenger {self._passenger_id} not found in state")
return passenger_data
def __getattribute__(self, name: str) -> Any:

View File

@@ -3,11 +3,11 @@ from typing import List
from elevator_saga.client.base_controller import ElevatorController
from elevator_saga.client.proxy_models import ProxyElevator, ProxyFloor, ProxyPassenger
from elevator_saga.core.models import SimulationEvent, Direction
from elevator_saga.core.models import Direction, SimulationEvent
class SingleElevatorBusController(ElevatorController):
def __init__(self):
class ElevatorBusExampleController(ElevatorController):
def __init__(self) -> None:
super().__init__("http://127.0.0.1:8000", True)
self.all_passengers: List[ProxyPassenger] = []
self.max_floor = 0
@@ -27,7 +27,11 @@ class SingleElevatorBusController(ElevatorController):
) -> None:
print(f"Tick {tick}: 即将处理 {len(events)} 个事件 {[e.type.value for e in events]}")
for i in elevators:
print(f"\t{i.id}[{i.target_floor_direction.value},{i.current_floor_float}/{i.target_floor}]" + "👦" * len(i.passengers), end="")
print(
f"\t{i.id}[{i.target_floor_direction.value},{i.current_floor_float}/{i.target_floor}]"
+ "👦" * len(i.passengers),
end="",
)
print()
def on_event_execute_end(
@@ -35,7 +39,7 @@ class SingleElevatorBusController(ElevatorController):
) -> None:
pass
def on_passenger_call(self, passenger:ProxyPassenger, floor: ProxyFloor, direction: str) -> None:
def on_passenger_call(self, passenger: ProxyPassenger, floor: ProxyFloor, direction: str) -> None:
self.all_passengers.append(passenger)
pass
@@ -67,6 +71,7 @@ class SingleElevatorBusController(ElevatorController):
def on_elevator_approaching(self, elevator: ProxyElevator, floor: ProxyFloor, direction: str) -> None:
pass
if __name__ == "__main__":
algorithm = SingleElevatorBusController()
algorithm = ElevatorBusExampleController()
algorithm.start()

View File

@@ -7,7 +7,7 @@ from typing import Dict, List
from elevator_saga.client.base_controller import ElevatorController
from elevator_saga.client.proxy_models import ProxyElevator, ProxyFloor, ProxyPassenger
from elevator_saga.core.models import SimulationEvent, Direction
from elevator_saga.core.models import Direction, SimulationEvent
class ElevatorBusController(ElevatorController):
@@ -45,7 +45,11 @@ class ElevatorBusController(ElevatorController):
"""事件执行前的回调"""
print(f"Tick {tick}: 即将处理 {len(events)} 个事件 {[e.type.value for e in events]}")
for i in elevators:
print(f"\t{i.id}[{i.target_floor_direction.value},{i.current_floor_float}/{i.target_floor}]" + "👦" * len(i.passengers), end="")
print(
f"\t{i.id}[{i.target_floor_direction.value},{i.current_floor_float}/{i.target_floor}]"
+ "👦" * len(i.passengers),
end="",
)
print()
def on_event_execute_end(
@@ -55,7 +59,7 @@ class ElevatorBusController(ElevatorController):
# print(f"✅ Tick {tick}: 已处理 {len(events)} 个事件")
pass
def on_passenger_call(self, passenger:ProxyPassenger, floor: ProxyFloor, direction: str) -> None:
def on_passenger_call(self, passenger: ProxyPassenger, floor: ProxyFloor, direction: str) -> None:
"""
乘客呼叫时的回调
公交车模式下电梯已经在循环运行无需特别响应呼叫
@@ -108,9 +112,7 @@ class ElevatorBusController(ElevatorController):
乘客上梯时的回调
打印乘客上梯信息
"""
print(
f" 乘客{passenger.id} E{elevator.id}⬆️ F{elevator.current_floor} -> F{passenger.destination}"
)
print(f" 乘客{passenger.id} E{elevator.id}⬆️ F{elevator.current_floor} -> F{passenger.destination}")
def on_passenger_alight(self, elevator: ProxyElevator, passenger: ProxyPassenger, floor: ProxyFloor) -> None:
"""
@@ -137,6 +139,7 @@ class ElevatorBusController(ElevatorController):
elevator.go_to_floor(elevator.target_floor + 1, immediate=True)
print(f" 不让0号电梯上行停站设定新目标楼层 {elevator.target_floor + 1}")
if __name__ == "__main__":
algorithm = ElevatorBusController(debug=True)
algorithm.start()

View File

@@ -85,7 +85,6 @@ class SerializableModel:
setattr(instance, k, v.__class__(value))
return instance
@classmethod
def from_json(cls: Type[T], json_str: str) -> T:
"""从JSON字符串创建实例"""
@@ -113,10 +112,10 @@ class Position(SerializableModel):
floor_up_position: int = 0
@property
def current_floor_float(self):
def current_floor_float(self) -> float:
return self.current_floor + self.floor_up_position / 10
def floor_up_position_add(self, num: int):
def floor_up_position_add(self, num: int) -> int:
self.floor_up_position += num
# 处理向上楼层跨越
@@ -139,7 +138,7 @@ class ElevatorIndicators(SerializableModel):
up: bool = False
down: bool = False
def set_direction(self, direction: Direction):
def set_direction(self, direction: Direction) -> None:
"""根据方向设置指示灯"""
if direction == Direction.UP:
self.up = True
@@ -202,13 +201,13 @@ class ElevatorState(SerializableModel):
id: int
position: Position
next_target_floor: Optional[int] = None
passengers: List[int] = field(default_factory=list) # type: ignore[reportUnknownVariableType] 乘客ID列表
passengers: List[int] = field(default_factory=list) # 乘客ID列表
max_capacity: int = 10
speed_pre_tick: float = 0.5
run_status: ElevatorStatus = ElevatorStatus.STOPPED
last_tick_direction: Direction = Direction.STOPPED
indicators: ElevatorIndicators = field(default_factory=ElevatorIndicators)
passenger_destinations: Dict[int, int] = field(default_factory=dict) # type: ignore[reportUnknownVariableType] 乘客ID -> 目的地楼层映射
passenger_destinations: Dict[int, int] = field(default_factory=dict) # 乘客ID -> 目的地楼层映射
energy_consumed: float = 0.0
last_update_tick: int = 0
@@ -223,7 +222,7 @@ class ElevatorState(SerializableModel):
def current_floor_float(self) -> float:
"""当前楼层"""
if isinstance(self.position, dict):
self.position = Position.from_dict(self.position)
self.position = Position.from_dict(self.position) # type: ignore[arg-type]
return self.position.current_floor_float
@property
@@ -269,7 +268,7 @@ class ElevatorState(SerializableModel):
"""按下的楼层(基于当前乘客的目的地动态计算)"""
return sorted(list(set(self.passenger_destinations.values())))
def clear_destinations(self):
def clear_destinations(self) -> None:
"""清空目标队列"""
self.next_target_floor = None
@@ -279,8 +278,8 @@ class FloorState(SerializableModel):
"""楼层状态"""
floor: int
up_queue: List[int] = field(default_factory=list) # type: ignore[reportUnknownVariableType] 等待上行的乘客ID
down_queue: List[int] = field(default_factory=list) # type: ignore[reportUnknownVariableType] 等待下行的乘客ID
up_queue: List[int] = field(default_factory=list) # 等待上行的乘客ID
down_queue: List[int] = field(default_factory=list) # 等待下行的乘客ID
@property
def has_waiting_passengers(self) -> bool:
@@ -292,7 +291,7 @@ class FloorState(SerializableModel):
"""总等待人数"""
return len(self.up_queue) + len(self.down_queue)
def add_waiting_passenger(self, passenger_id: int, direction: Direction):
def add_waiting_passenger(self, passenger_id: int, direction: Direction) -> None:
"""添加等待乘客"""
if direction == Direction.UP:
if passenger_id not in self.up_queue:
@@ -321,7 +320,7 @@ class SimulationEvent(SerializableModel):
data: Dict[str, Any]
timestamp: Optional[str] = None
def __post_init__(self):
def __post_init__(self) -> None:
if self.timestamp is None:
self.timestamp = datetime.now().isoformat()
@@ -360,9 +359,9 @@ class SimulationState(SerializableModel):
tick: int
elevators: List[ElevatorState]
floors: List[FloorState]
passengers: Dict[int, PassengerInfo] = field(default_factory=dict) # type: ignore[reportUnknownVariableType]
passengers: Dict[int, PassengerInfo] = field(default_factory=dict)
metrics: PerformanceMetrics = field(default_factory=PerformanceMetrics)
events: List[SimulationEvent] = field(default_factory=list) # type: ignore[reportUnknownVariableType]
events: List[SimulationEvent] = field(default_factory=list)
def get_elevator_by_id(self, elevator_id: int) -> Optional[ElevatorState]:
"""根据ID获取电梯"""
@@ -382,7 +381,7 @@ class SimulationState(SerializableModel):
"""根据状态获取乘客"""
return [p for p in self.passengers.values() if p.status == status]
def add_event(self, event_type: EventType, data: Dict[str, Any]):
def add_event(self, event_type: EventType, data: Dict[str, Any]) -> None:
"""添加事件"""
event = SimulationEvent(tick=self.tick, type=event_type, data=data)
self.events.append(event)
@@ -422,7 +421,7 @@ class StepResponse(SerializableModel):
success: bool
tick: int
events: List[SimulationEvent] = field(default_factory=list) # type: ignore[reportUnknownVariableType]
events: List[SimulationEvent] = field(default_factory=list)
request_id: Optional[str] = None
error_message: Optional[str] = None
timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
@@ -443,7 +442,7 @@ class ElevatorCommand(SerializableModel):
elevator_id: int
command_type: str # "go_to_floor", "stop"
parameters: Dict[str, Any] = field(default_factory=dict) # type: ignore[reportUnknownVariableType]
parameters: Dict[str, Any] = field(default_factory=dict)
request_id: str = field(default_factory=lambda: str(uuid.uuid4()))
timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
@@ -494,7 +493,7 @@ class TrafficPattern(SerializableModel):
entries: List[TrafficEntry] = field(default_factory=list)
metadata: Dict[str, Any] = field(default_factory=dict)
def add_entry(self, entry: TrafficEntry):
def add_entry(self, entry: TrafficEntry) -> None:
"""添加流量条目"""
self.entries.append(entry)

View File

@@ -7,7 +7,7 @@ from typing import Dict, List
from elevator_saga.client.base_controller import ElevatorController
from elevator_saga.client.proxy_models import ProxyElevator, ProxyFloor, ProxyPassenger
from elevator_saga.core.models import SimulationEvent, Direction
from elevator_saga.core.models import Direction, SimulationEvent
class ElevatorBusController(ElevatorController):
@@ -45,7 +45,11 @@ class ElevatorBusController(ElevatorController):
"""事件执行前的回调"""
print(f"Tick {tick}: 即将处理 {len(events)} 个事件 {[e.type.value for e in events]}")
for i in elevators:
print(f"\t{i.id}[{i.target_floor_direction.value},{i.current_floor_float}/{i.target_floor}]" + "👦" * len(i.passengers), end="")
print(
f"\t{i.id}[{i.target_floor_direction.value},{i.current_floor_float}/{i.target_floor}]"
+ "👦" * len(i.passengers),
end="",
)
print()
def on_event_execute_end(
@@ -54,7 +58,7 @@ class ElevatorBusController(ElevatorController):
"""事件执行后的回调"""
pass
def on_passenger_call(self, passenger:ProxyPassenger, floor: ProxyFloor, direction: str) -> None:
def on_passenger_call(self, passenger: ProxyPassenger, floor: ProxyFloor, direction: str) -> None:
"""
乘客呼叫时的回调
公交车模式下,电梯已经在循环运行,无需特别响应呼叫
@@ -107,9 +111,7 @@ class ElevatorBusController(ElevatorController):
乘客上梯时的回调
打印乘客上梯信息
"""
print(
f" 乘客{passenger.id} E{elevator.id}⬆️ F{elevator.current_floor} -> F{passenger.destination}"
)
print(f" 乘客{passenger.id} E{elevator.id}⬆️ F{elevator.current_floor} -> F{passenger.destination}")
def on_passenger_alight(self, elevator: ProxyElevator, passenger: ProxyPassenger, floor: ProxyFloor) -> None:
"""
@@ -136,6 +138,7 @@ class ElevatorBusController(ElevatorController):
elevator.go_to_floor(elevator.target_floor + 1, immediate=True)
print(f" 不让0号电梯上行停站设定新目标楼层 {elevator.target_floor + 1}")
if __name__ == "__main__":
algorithm = ElevatorBusController(debug=True)
algorithm.start()

View File

@@ -33,13 +33,13 @@ from elevator_saga.core.models import (
_SERVER_DEBUG_MODE = False
def set_server_debug_mode(enabled: bool):
def set_server_debug_mode(enabled: bool) -> None:
"""Enable or disable server debug logging"""
global _SERVER_DEBUG_MODE
globals()["_SERVER_DEBUG_MODE"] = enabled
def server_debug_log(message: str):
def server_debug_log(message: str) -> None:
"""Print server debug message if debug mode is enabled"""
if _SERVER_DEBUG_MODE:
print(f"[SERVER-DEBUG] {message}", flush=True)
@@ -360,9 +360,7 @@ class ElevatorSimulation:
destination=traffic_entry.destination,
arrive_tick=self.tick,
)
assert (
traffic_entry.origin != traffic_entry.destination
), f"乘客{passenger.id}目的地和起始地{traffic_entry.origin}重复"
assert traffic_entry.origin != traffic_entry.destination, f"乘客{passenger.id}目的地和起始地{traffic_entry.origin}重复"
self.passengers[passenger.id] = passenger
server_debug_log(f"乘客 {passenger.id:4} 创建 | {passenger}")
if passenger.destination > passenger.origin:
@@ -434,7 +432,9 @@ class ElevatorSimulation:
if target_floor == new_floor and elevator.position.floor_up_position == 0:
elevator.run_status = ElevatorStatus.STOPPED
# 刚进入Stopped状态可以通过last_direction识别
self._emit_event(EventType.STOPPED_AT_FLOOR, {"elevator": elevator.id, "floor": new_floor, "reason": "move_reached"})
self._emit_event(
EventType.STOPPED_AT_FLOOR, {"elevator": elevator.id, "floor": new_floor, "reason": "move_reached"}
)
# elevator.energy_consumed += abs(direction * elevator.speed_pre_tick) * 0.5
def _process_elevator_stops(self) -> None:
@@ -471,7 +471,7 @@ class ElevatorSimulation:
self._set_elevator_target_floor(elevator, elevator.next_target_floor)
elevator.next_target_floor = None
def _set_elevator_target_floor(self, elevator: ElevatorState, floor: int):
def _set_elevator_target_floor(self, elevator: ElevatorState, floor: int) -> None:
"""
同一个tick内提示
[SERVER-DEBUG] 电梯 E0 下一目的地设定为 F1
@@ -492,9 +492,7 @@ class ElevatorSimulation:
server_debug_log(f"电梯 E{elevator.id} 被设定为减速")
if elevator.current_floor != floor or elevator.position.floor_up_position != 0:
old_status = elevator.run_status.value
server_debug_log(
f"电梯{elevator.id} 状态:{old_status}->{elevator.run_status.value}"
)
server_debug_log(f"电梯{elevator.id} 状态:{old_status}->{elevator.run_status.value}")
def _calculate_distance_to_target(self, elevator: ElevatorState) -> float:
"""计算到目标楼层的距离以floor_up_position为单位"""

View File

@@ -455,4 +455,4 @@
"tick": 196
}
]
}
}

View File

@@ -179,4 +179,4 @@
"tick": 74
}
]
}
}

View File

@@ -9,7 +9,7 @@ import math
import os.path
import random
from pathlib import Path
from typing import Any, Dict, List, Optional
from typing import Any, Callable, Dict, List, Optional
# 建筑规模配置
BUILDING_SCALES = {
@@ -294,9 +294,7 @@ def generate_fire_evacuation_traffic(
# 在10个tick内陆续到达模拟疏散的紧急性
arrival_tick = alarm_tick + random.randint(0, min(10, duration - alarm_tick - 1))
if arrival_tick < duration:
traffic.append(
{"id": passenger_id, "origin": floor, "destination": 0, "tick": arrival_tick} # 疏散到大厅
)
traffic.append({"id": passenger_id, "origin": floor, "destination": 0, "tick": arrival_tick}) # 疏散到大厅
passenger_id += 1
return limit_traffic_count(traffic, max_people)
@@ -738,12 +736,12 @@ def determine_building_scale(floors: int, elevators: int) -> str:
return "large"
def generate_traffic_file(scenario: str, output_file: str, scale: Optional[str] = None, **kwargs) -> int:
def generate_traffic_file(scenario: str, output_file: str, scale: Optional[str] = None, **kwargs: Any) -> int:
"""生成单个流量文件,支持规模化配置"""
if scenario not in TRAFFIC_SCENARIOS:
raise ValueError(f"Unknown scenario: {scenario}. Available: {list(TRAFFIC_SCENARIOS.keys())}")
config = TRAFFIC_SCENARIOS[scenario]
config: Dict[str, Any] = TRAFFIC_SCENARIOS[scenario]
# 确定建筑规模
if scale is None:
@@ -766,6 +764,7 @@ def generate_traffic_file(scenario: str, output_file: str, scale: Optional[str]
scale_params = config["scales"].get(scale, {})
# 合并参数kwargs > scale_params > building_scale_defaults
assert scale is not None # scale should be determined by this point
building_scale = BUILDING_SCALES[scale]
params = {}
@@ -786,9 +785,10 @@ def generate_traffic_file(scenario: str, output_file: str, scale: Optional[str]
# 生成流量数据 - 只传递生成器函数需要的参数
import inspect
generator_signature = inspect.signature(config["generator"])
generator_func: Callable[..., List[Dict[str, Any]]] = config["generator"]
generator_signature = inspect.signature(generator_func)
generator_params = {k: v for k, v in params.items() if k in generator_signature.parameters}
traffic_data = config["generator"](**generator_params)
traffic_data = generator_func(**generator_params)
# 准备building配置
building_config = {
@@ -819,7 +819,7 @@ def generate_scaled_traffic_files(
seed: int = 42,
generate_all_scales: bool = False,
custom_building: Optional[Dict[str, Any]] = None,
):
) -> None:
"""生成按规模分类的流量文件"""
output_path = Path(output_dir)
output_path.mkdir(exist_ok=True)
@@ -848,7 +848,7 @@ def generate_scaled_traffic_files(
def _generate_files_for_scale(
output_path: Path, scale: str, seed: int, custom_building: Optional[Dict[str, Any]] = None
):
) -> None:
"""为指定规模生成所有适合的场景文件"""
building_config = BUILDING_SCALES[scale]
total_passengers = 0
@@ -868,9 +868,10 @@ def _generate_files_for_scale(
print(f"\nGenerating {scale} scale traffic files:")
print(f"Building: {floors} floors, {elevators} elevators, capacity {elevator_capacity}")
for scenario_name, config in TRAFFIC_SCENARIOS.items():
for scenario_name, scenario_config in TRAFFIC_SCENARIOS.items():
# 检查场景是否适合该规模
if scale not in config["suitable_scales"]:
config_dict: Dict[str, Any] = scenario_config
if scale not in config_dict["suitable_scales"]:
continue
filename = f"{scenario_name}.json"
@@ -905,7 +906,7 @@ def generate_all_traffic_files(
elevators: int = 2,
elevator_capacity: int = 8,
seed: int = 42,
):
) -> None:
"""生成所有场景的流量文件 - 保持向后兼容"""
scale = determine_building_scale(floors, elevators)
custom_building = {"floors": floors, "elevators": elevators, "capacity": elevator_capacity}
@@ -913,7 +914,7 @@ def generate_all_traffic_files(
generate_scaled_traffic_files(output_dir=output_dir, scale=scale, seed=seed, custom_building=custom_building)
def main():
def main() -> None:
"""主函数 - 命令行接口"""
import argparse

View File

@@ -407,4 +407,4 @@
"tick": 197
}
]
}
}

View File

@@ -371,4 +371,4 @@
"tick": 197
}
]
}
}

View File

@@ -161,4 +161,4 @@
"tick": 192
}
]
}
}

View File

@@ -491,4 +491,4 @@
"tick": 138
}
]
}
}

View File

@@ -311,4 +311,4 @@
"tick": 49
}
]
}
}

View File

@@ -611,4 +611,4 @@
"tick": 190
}
]
}
}

View File

@@ -587,4 +587,4 @@
"tick": 199
}
]
}
}

View File

@@ -479,4 +479,4 @@
"tick": 196
}
]
}
}

View File

@@ -491,4 +491,4 @@
"tick": 146
}
]
}
}

View File

@@ -1,45 +1,56 @@
[build-system]
requires = ["setuptools>=45", "wheel", "setuptools_scm[toml]>=6.2"]
requires = ["setuptools>=61.0", "wheel"]
build-backend = "setuptools.build_meta"
[project]
name = "elevator-saga"
name = "elevator-py"
dynamic = ["version"]
description = "Python implementation of Elevator Saga game with PyEE event system"
readme = "README_CN.md"
license = {text = "MIT"}
description = "Python implementation of Elevator Saga game with event system"
readme = "README.md"
requires-python = ">=3.10"
license = {file = "LICENSE"}
authors = [
{name = "Elevator Saga Team"},
{name = "ZGCA-Forge Team", email = "zgca@zgca.com"}
]
keywords = ["elevator", "simulation", "game", "event-driven", "optimization"]
classifiers = [
"Development Status :: 4 - Beta",
"Intended Audience :: Developers",
"Intended Audience :: Education",
"Intended Audience :: Education",
"License :: OSI Approved :: MIT License",
"Operating System :: OS Independent",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Programming Language :: Python :: 3.13",
"Topic :: Games/Entertainment :: Simulation",
"Topic :: Software Development :: Libraries :: Python Modules",
]
requires-python = ">=3.12"
dependencies = [
"pyee>=11.0.0",
"numpy>=1.20.0",
"matplotlib>=3.5.0",
"seaborn>=0.11.0",
"pandas>=1.3.0",
"flask",
"flask>=2.0.0",
]
[project.optional-dependencies]
dev = [
"pytest>=6.0",
"pytest-cov",
"black",
"flake8",
"isort",
"mypy",
"pytest>=7.0.0",
"pytest-cov>=4.0.0",
"black>=22.0.0",
"isort>=5.0.0",
"mypy>=1.0.0",
"flake8>=6.0.0",
"pre-commit>=2.20.0",
"bump2version>=1.0.0",
]
docs = [
"sphinx>=5.0.0",
"sphinx-rtd-theme>=1.0.0",
"myst-parser>=0.18.0",
]
all = [
"elevatorpy[dev,docs]",
]
[project.scripts]
@@ -50,66 +61,38 @@ elevator-grader = "elevator_saga.grader.grader:main"
elevator-batch-test = "elevator_saga.grader.batch_runner:main"
[project.urls]
Repository = "https://github.com/yourusername/elevator-saga"
Issues = "https://github.com/yourusername/elevator-saga/issues"
[tool.setuptools_scm]
write_to = "elevator_saga/_version.py"
Homepage = "https://github.com/ZGCA-Forge/Elevator"
Documentation = "https://zgca-forge.github.io/Elevator/"
Repository = "https://github.com/ZGCA-Forge/Elevator"
Issues = "https://github.com/ZGCA-Forge/Elevator/issues"
[tool.setuptools.packages.find]
where = ["."]
include = ["elevator_saga*"]
[tool.setuptools.dynamic]
version = {attr = "elevator_saga.__version__"}
[tool.black]
line-length = 120
target-version = ['py312']
include = '\.pyi?$'
extend-exclude = '''
/(
# directories
\.eggs
| \.git
| \.hg
| \.mypy_cache
| \.tox
| \.venv
| _build
| buck-out
| build
| dist
)/
'''
target-version = ['py310', 'py311', 'py312']
[tool.isort]
profile = "black"
line_length = 120
multi_line_output = 3
include_trailing_comma = true
force_grid_wrap = 0
use_parentheses = true
ensure_newline_before_comments = true
[tool.mypy]
python_version = "3.12"
python_version = "3.10"
warn_return_any = true
warn_unused_configs = true
disallow_untyped_defs = true
disallow_incomplete_defs = true
check_untyped_defs = true
disallow_untyped_decorators = true
no_implicit_optional = true
warn_redundant_casts = true
warn_unused_ignores = true
warn_no_return = true
warn_unreachable = true
strict_equality = true
[[tool.mypy.overrides]]
module = [
"pyee.*",
"matplotlib.*",
"seaborn.*",
"pandas.*",
"numpy.*",
"flask.*",
]
@@ -117,38 +100,20 @@ ignore_missing_imports = true
[tool.pytest.ini_options]
testpaths = ["tests"]
python_files = ["test_*.py", "*_test.py"]
python_classes = ["Test*"]
python_functions = ["test_*"]
addopts = [
"--strict-markers",
"--strict-config",
"--verbose",
]
markers = [
"slow: marks tests as slow (deselect with '-m \"not slow\"')",
"integration: marks tests as integration tests",
python_files = "test_*.py"
python_classes = "Test*"
python_functions = "test_*"
addopts = "-v --tb=short --strict-markers --strict-config -ra --color=yes"
filterwarnings = [
"ignore::DeprecationWarning",
"ignore::PendingDeprecationWarning",
]
[tool.coverage.run]
source = ["elevator_saga"]
omit = [
"*/tests/*",
"*/test_*",
"*/__pycache__/*",
"*/.*",
]
[tool.coverage.report]
exclude_lines = [
"pragma: no cover",
"def __repr__",
"if self.debug:",
"if settings.DEBUG",
"raise AssertionError",
"raise NotImplementedError",
"if 0:",
"if __name__ == .__main__.:",
"class .*\\bProtocol\\):",
"@(abc\\.)?abstractmethod",
]
[tool.bandit]
exclude_dirs = ["tests", "test_*.py", "*_test.py", ".venv", "venv", "build", "dist"]
# B101: assert语句用于类型检查和开发时验证不是安全问题
# B601: shell命令参数化在受控环境中使用
# B310: urllib.urlopen用于连接受控的API服务器URL来源可信
# B311: random模块用于生成电梯流量模拟数据非加密用途
skips = ["B101", "B601", "B310", "B311"]

View File

@@ -1,33 +1,30 @@
{
"include": ["."],
"exclude": [
"build",
"dist",
"__pycache__",
".mypy_cache",
".pytest_cache",
"htmlcov",
".idea",
".vscode",
"docs/_build",
"elevator_saga.egg-info"
],
"pythonVersion": "3.10",
"typeCheckingMode": "strict",
"executionEnvironments": [
{
"root": ".",
"extraPaths": ["elevator_saga"]
}
],
"reportMissingImports": "none",
"reportUnusedImport": "warning",
"reportUnusedVariable": "warning",
"reportUnknownArgumentType": "warning",
"reportUnknownMemberType": "warning",
"reportUnknownVariableType": "warning",
"reportUnknownParameterType": "warning",
"reportPrivateUsage": "warning",
"reportMissingTypeStubs": false
}
"include": ["."],
"exclude": [
"build",
"dist",
"__pycache__",
".mypy_cache",
".pytest_cache",
"htmlcov",
".idea",
".vscode",
"docs/_build",
"elevatorpy.egg-info",
"elevator_saga.egg-info",
".eggs",
"MsgCenterPy"
],
"pythonVersion": "3.10",
"typeCheckingMode": "basic",
"executionEnvironments": [
{
"root": ".",
"extraPaths": ["elevator_saga"]
}
],
"reportMissingImports": "warning",
"reportUnusedImport": "warning",
"reportUnusedVariable": "warning",
"reportMissingTypeStubs": false
}

View File

@@ -1,157 +0,0 @@
#!/usr/bin/env python3
"""
Test runner for Elevator Saga project
"""
import argparse
import subprocess
import sys
import os
from pathlib import Path
def check_dependencies():
"""Check if all dependencies are installed correctly"""
print("🔍 Checking dependencies...")
try:
import elevator_saga
print(f"✅ elevator_saga version: {getattr(elevator_saga, '__version__', 'unknown')}")
# Check main dependencies
dependencies = ["pyee", "numpy", "matplotlib", "seaborn", "pandas", "flask"]
for dep in dependencies:
try:
__import__(dep)
print(f"{dep}: installed")
except ImportError:
print(f"{dep}: missing")
return False
print("✅ All dependencies are correctly installed")
return True
except ImportError as e:
print(f"❌ Error importing elevator_saga: {e}")
return False
def run_unit_tests():
"""Run unit tests"""
print("🧪 Running unit tests...")
# Check if tests directory exists
tests_dir = Path("tests")
if not tests_dir.exists():
print(" No tests directory found, creating basic test structure...")
tests_dir.mkdir()
(tests_dir / "__init__.py").touch()
# Create a basic test file
basic_test = tests_dir / "test_basic.py"
basic_test.write_text(
'''"""Basic tests for elevator_saga"""
import unittest
from elevator_saga.core.models import Direction, SimulationEvent
class TestBasic(unittest.TestCase):
"""Basic functionality tests"""
def test_direction_enum(self):
"""Test Direction enum"""
self.assertEqual(Direction.UP.value, "up")
self.assertEqual(Direction.DOWN.value, "down")
self.assertEqual(Direction.NONE.value, "none")
def test_import(self):
"""Test that main modules can be imported"""
import elevator_saga.client.base_controller
import elevator_saga.core.models
import elevator_saga.server.simulator
if __name__ == '__main__':
unittest.main()
'''
)
# Run pytest if available, otherwise unittest
try:
result = subprocess.run([sys.executable, "-m", "pytest", "tests/", "-v"], capture_output=True, text=True)
print(result.stdout)
if result.stderr:
print(result.stderr)
return result.returncode == 0
except FileNotFoundError:
print("pytest not found, using unittest...")
result = subprocess.run([sys.executable, "-m", "unittest", "discover", "tests"], capture_output=True, text=True)
print(result.stdout)
if result.stderr:
print(result.stderr)
return result.returncode == 0
def run_example_tests():
"""Run example files to ensure they work"""
print("🚀 Running example tests...")
example_files = ["simple_example.py", "test_example.py"]
for example_file in example_files:
if os.path.exists(example_file):
print(f"Testing {example_file}...")
# Just check if the file can be imported without errors
try:
result = subprocess.run(
[sys.executable, "-c", f"import {example_file[:-3]}"], capture_output=True, text=True, timeout=10
)
if result.returncode == 0:
print(f"{example_file}: import successful")
else:
print(f"{example_file}: import failed")
print(result.stderr)
return False
except subprocess.TimeoutExpired:
print(f"{example_file}: timeout (probably waiting for server)")
# This is expected for examples that try to connect to server
print(f"{example_file}: import successful (with server connection)")
except Exception as e:
print(f"{example_file}: error - {e}")
return False
return True
def main():
parser = argparse.ArgumentParser(description="Run tests for Elevator Saga")
parser.add_argument("--check-deps", action="store_true", help="Check dependencies only")
parser.add_argument("--type", choices=["unit", "examples", "all"], default="all", help="Type of tests to run")
args = parser.parse_args()
success = True
if args.check_deps:
success = check_dependencies()
else:
# Always check dependencies first
if not check_dependencies():
print("❌ Dependency check failed")
return 1
if args.type in ["unit", "all"]:
if not run_unit_tests():
success = False
if args.type in ["examples", "all"]:
if not run_example_tests():
success = False
if success:
print("🎉 All tests passed!")
return 0
else:
print("💥 Some tests failed!")
return 1
if __name__ == "__main__":
sys.exit(main())

View File

@@ -1,59 +0,0 @@
#!/usr/bin/env python3
"""
Setup script for Elevator Saga Python Package
"""
from setuptools import setup, find_packages
with open("README_CN.md", "r", encoding="utf-8") as fh:
long_description = fh.read()
setup(
name="elevator-saga",
version="1.0.0",
author="Elevator Saga Team",
description="Python implementation of Elevator Saga game with PyEE event system",
long_description=long_description,
long_description_content_type="text/markdown",
packages=find_packages(),
classifiers=[
"Development Status :: 4 - Beta",
"Intended Audience :: Developers",
"Intended Audience :: Education",
"License :: OSI Approved :: MIT License",
"Operating System :: OS Independent",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.12",
"Topic :: Games/Entertainment :: Simulation",
"Topic :: Software Development :: Libraries :: Python Modules",
],
python_requires=">=3.12",
install_requires=[
"pyee>=11.0.0",
"numpy>=1.20.0",
"matplotlib>=3.5.0",
"seaborn>=0.11.0",
"pandas>=1.3.0",
"flask",
],
extras_require={
"dev": [
"pytest>=6.0",
"pytest-cov",
"black",
"flake8",
"isort",
"mypy",
],
},
entry_points={
"console_scripts": [
"elevator-saga=elevator_saga.cli.main:main",
"elevator-server=elevator_saga.cli.main:server_main",
"elevator-client=elevator_saga.cli.main:client_main",
"elevator-grader=elevator_saga.grader.grader:main",
"elevator-batch-test=elevator_saga.grader.batch_runner:main",
],
},
include_package_data=True,
zip_safe=False,
)

3
tests/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
"""
Elevator Saga Test Suite
"""

73
tests/test_imports.py Normal file
View File

@@ -0,0 +1,73 @@
"""
Test module imports to ensure all modules can be imported successfully
"""
import pytest
def test_import_core_models():
"""Test importing core data models"""
from elevator_saga.core.models import (
Direction,
ElevatorState,
ElevatorStatus,
EventType,
FloorState,
PassengerInfo,
PassengerStatus,
SimulationState,
TrafficEntry,
TrafficPattern,
)
assert Direction is not None
assert ElevatorState is not None
assert ElevatorStatus is not None
assert EventType is not None
assert FloorState is not None
assert PassengerInfo is not None
assert PassengerStatus is not None
assert SimulationState is not None
assert TrafficEntry is not None
assert TrafficPattern is not None
def test_import_client_api():
"""Test importing client API"""
from elevator_saga.client.api_client import ElevatorAPIClient
assert ElevatorAPIClient is not None
def test_import_proxy_models():
"""Test importing proxy models"""
from elevator_saga.client.proxy_models import ProxyElevator, ProxyFloor, ProxyPassenger
assert ProxyElevator is not None
assert ProxyFloor is not None
assert ProxyPassenger is not None
def test_import_base_controller():
"""Test importing base controller"""
from elevator_saga.client.base_controller import ElevatorController
assert ElevatorController is not None
def test_import_simulator():
"""Test importing simulator"""
from elevator_saga.server.simulator import ElevatorSimulation
assert ElevatorSimulation is not None
def test_import_client_example():
"""Test importing client example"""
from elevator_saga.client_examples.bus_example import ElevatorBusExampleController
assert ElevatorBusExampleController is not None
if __name__ == "__main__":
pytest.main([__file__, "-v"])