open_pub/Elevator
1
0
Fork 0
mirror of https://github.com/ZGCA-Forge/Elevator.git synced 2026-02-10 09:45:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

Compare commits

base: open_pub:0.0.3
Branches Tags
open_pub:main open_pub:dependabot/github_actions/actions/download-artifact-7 open_pub:dependabot/github_actions/actions/cache-5 open_pub:dependabot/github_actions/actions/upload-artifact-6 open_pub:dependabot/github_actions/actions/checkout-6
open_pub:0.0.11 open_pub:0.0.10 open_pub:0.0.9 open_pub:0.0.8 open_pub:0.0.7 open_pub:0.0.6 open_pub:0.0.5 open_pub:0.0.4 open_pub:0.0.3 open_pub:0.0.2
..
compare: open_pub:0903e6ebd0e70dde9e459d9f991669da0923d973
Branches Tags
open_pub:dependabot/github_actions/actions/download-artifact-7 open_pub:dependabot/github_actions/actions/cache-5 open_pub:dependabot/github_actions/actions/upload-artifact-6 open_pub:dependabot/github_actions/actions/checkout-6 open_pub:main
open_pub:0.0.11 open_pub:0.0.10 open_pub:0.0.9 open_pub:0.0.8 open_pub:0.0.7 open_pub:0.0.6 open_pub:0.0.5 open_pub:0.0.4 open_pub:0.0.3 open_pub:0.0.2

1 Commits
0.0.3 ... 0903e6ebd0

Author SHA1 Message Date
dependabot[bot]
0903e6ebd0 ci(deps): bump codecov/codecov-action from 3 to 5 
Bumps [codecov/codecov-action](https://github.com/codecov/codecov-action) from 3 to 5.
- [Release notes](https://github.com/codecov/codecov-action/releases)
- [Changelog](https://github.com/codecov/codecov-action/blob/main/CHANGELOG.md)
- [Commits](https://github.com/codecov/codecov-action/compare/v3...v5)

---
updated-dependencies:
- dependency-name: codecov/codecov-action
  dependency-version: '5'
  dependency-type: direct:production
  update-type: version-update:semver-major
...

Signed-off-by: dependabot[bot] <support@github.com>
 2025-09-29 02:45:28 +00:00
50 changed files with 1138 additions and 3095 deletions
Expand all files Collapse all files

10
.bumpversion.cfg
View File

@@ -1,10 +0,0 @@
[bumpversion]
current_version = 0.0.3
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}"

5
.github/dependabot.yml vendored
View File

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

346
.github/workflows/ci.yml vendored
View File

@@ -1,175 +1,203 @@
# This workflow will install Python dependencies, run tests and lint with a variety of Python versions name: CI
# For more information see: https://docs.github.com/en/actions/automating-builds-and-tests/building-and-testing-python
name: Python package
on: on:
push: push:
branches: ["main", "dev"] branches: [ main, develop ]
pull_request: pull_request:
branches: ["main", "dev"] branches: [ main, develop ]
schedule:
# Run tests daily at 6 AM UTC
- cron: '0 6 * * *'
jobs: jobs:
# Step 1: Code formatting and pre-commit validation (fast failure) test:
code-format:
name: Code formatting and pre-commit validation
runs-on: ubuntu-latest
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: [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
needs: [basic-build] # Run in parallel with other tests after basic build
steps:
- 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 }} name: Test Python ${{ matrix.python-version }} on ${{ matrix.os }}
runs-on: ${{ matrix.os }} runs-on: ${{ matrix.os }}
needs: [security, package-build] # Wait for all prerequisite checks
strategy: strategy:
fail-fast: false fail-fast: false
matrix: matrix:
os: [ubuntu-latest, windows-latest, macos-latest] os: [ubuntu-latest]
python-version: ["3.10", "3.11", "3.12", "3.13"] python-version: ['3.12']
exclude:
# Skip the combination we already tested in basic-build
- os: ubuntu-latest
python-version: "3.10"
steps: steps:
- uses: actions/checkout@v5 - 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@v5
with:
file: ./coverage.xml
flags: unittests
name: codecov-umbrella
fail_ci_if_error: false
- name: Set up Python ${{ matrix.python-version }} test-examples:
uses: actions/setup-python@v6 name: Test examples
with: runs-on: ubuntu-latest
python-version: ${{ matrix.python-version }}
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
- name: Cache pip dependencies build:
uses: actions/cache@v4 name: Build and check package
with: runs-on: ubuntu-latest
path: ~/.cache/pip
key: ${{ runner.os }}-pip-${{ matrix.python-version }}-${{ hashFiles('**/pyproject.toml') }} steps:
restore-keys: | - uses: actions/checkout@v4
${{ runner.os }}-pip-${{ matrix.python-version }}-
${{ runner.os }}-pip- - 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/
- name: Install dependencies security:
run: | name: Security scan
python -m pip install --upgrade pip runs-on: ubuntu-latest
python -m pip install pytest
pip install -e .[dev] 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()
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
- name: Test with pytest
run: |
pytest

269
.github/workflows/code-quality.yml vendored Normal file
View File

@@ -0,0 +1,269 @@
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

34
.github/workflows/dependabot.yml vendored Normal file
View File

@@ -0,0 +1,34 @@
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
View File

@@ -1,83 +0,0 @@
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

385
.github/workflows/publish.yml vendored
View File

@@ -1,258 +1,189 @@
# This workflow will upload a Python Package to PyPI when a release is created name: Publish to PyPI
# 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: on:
release: release:
types: [published, edited] types: [published]
workflow_dispatch: workflow_dispatch:
inputs: inputs:
test_pypi: test_pypi:
description: "Publish to Test PyPI instead of PyPI" description: 'Publish to Test PyPI instead of PyPI'
required: false required: false
default: false default: 'false'
type: boolean type: boolean
permissions:
contents: read
jobs: jobs:
# Step 1: Code formatting and pre-commit validation (fast failure) test:
code-format: name: Run tests before publish
name: Code formatting and pre-commit validation
runs-on: ubuntu-latest runs-on: ubuntu-latest
steps: steps:
- uses: actions/checkout@v5 - 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
- name: Set up Python build:
uses: actions/setup-python@v6 name: Build package
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 runs-on: ubuntu-latest
needs: [code-format] # Only run after code formatting passes needs: test
steps: steps:
- uses: actions/checkout@v5 - 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
- name: Set up Python 3.10 publish-test:
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: [basic-build] # Run in parallel with other tests after basic build
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:
- 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: 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 name: Publish to Test PyPI
runs-on: ubuntu-latest runs-on: ubuntu-latest
needs: needs: build
- release-build if: github.event.inputs.test_pypi == 'true' || (github.event_name == 'release' && github.event.release.prerelease)
if: github.event.inputs.test_pypi == 'true' || (github.event_name == 'release' && github.event.release.prerelease && (github.event.action == 'published' || github.event.action == 'edited')) environment:
permissions: name: test-pypi
# IMPORTANT: this permission is mandatory for trusted publishing url: https://test.pypi.org/p/elevator-saga
id-token: write
# Note: For enhanced security, consider configuring deployment environments
# in your GitHub repository settings with protection rules
steps: steps:
- name: Retrieve release distributions - name: Download build artifacts
uses: actions/download-artifact@v5 uses: actions/download-artifact@v3
with: with:
name: release-dists name: dist-${{ github.run_number }}
path: dist/ 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
- name: Publish release distributions to Test PyPI publish-pypi:
uses: pypa/gh-action-pypi-publish@release/v1 name: Publish to PyPI
with: runs-on: ubuntu-latest
repository-url: https://test.pypi.org/legacy/ 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
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
create-github-release-assets: create-github-release-assets:
name: Add assets to GitHub release name: Add assets to GitHub release
runs-on: ubuntu-latest runs-on: ubuntu-latest
needs: release-build needs: build
if: github.event_name == 'release' && (github.event.action == 'published' || github.event.action == 'edited') if: github.event_name == 'release'
permissions:
contents: write # Need write access to upload release assets
steps: steps:
- name: Retrieve release distributions - name: Download build artifacts
uses: actions/download-artifact@v5 uses: actions/download-artifact@v3
with: with:
name: release-dists name: dist-${{ github.run_number }}
path: dist/ path: dist/
- name: Upload release assets - name: Upload release assets
uses: softprops/action-gh-release@v2 uses: softprops/action-gh-release@v1
with: with:
files: dist/* files: dist/*
env: env:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
post-publish: post-publish:
name: Post-publish tasks name: Post-publish tasks
runs-on: ubuntu-latest runs-on: ubuntu-latest
needs: [pypi-publish, test-pypi-publish] needs: [publish-pypi, publish-test]
if: always() && (needs.pypi-publish.result == 'success' || needs.test-pypi-publish.result == 'success') if: always() && (needs.publish-pypi.result == 'success' || needs.publish-test.result == 'success')
steps: steps:
- uses: actions/checkout@v5 - 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 等通知
- 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,60 +1,5 @@
# ================================ .vscode
# Python-related files .idea
# ================================ __pycache__
.mypy_cache
# Compiled Python files elevator_saga.egg-info
__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

61
.pre-commit-config.yaml
View File

@@ -5,34 +5,33 @@ repos:
hooks: hooks:
- id: black - id: black
language_version: python3 language_version: python3
args: ["--line-length=120"] args: ["--line-length=88"]
# Import sorting # Import sorting
- repo: https://github.com/pycqa/isort - repo: https://github.com/pycqa/isort
rev: 5.13.2 rev: 5.13.2
hooks: hooks:
- id: isort - id: isort
args: args: ["--profile", "black", "--multi-line", "3"]
["--profile", "black", "--multi-line", "3", "--line-length", "120"]
# Linting # Linting
- repo: https://github.com/pycqa/flake8 - repo: https://github.com/pycqa/flake8
rev: 7.0.0 rev: 7.0.0
hooks: hooks:
- id: flake8 - id: flake8
args: args:
- "--max-line-length=200" # Allow longer lines after black formatting - "--max-line-length=88"
- "--extend-ignore=E203,W503,F401,E402,E721,F841" - "--extend-ignore=E203,W503,F401"
- "--exclude=build,dist,__pycache__,.mypy_cache,.pytest_cache,htmlcov,.idea,.vscode,docs/_build,elevatorpy.egg-info,elevator_saga.egg-info,.eggs" - "--exclude=build,dist,.eggs"
# Type checking # Type checking
- repo: https://github.com/pre-commit/mirrors-mypy - repo: https://github.com/pre-commit/mirrors-mypy
rev: v1.8.0 rev: v1.8.0
hooks: hooks:
- id: mypy - id: mypy
additional_dependencies: [types-PyYAML, types-jsonschema] additional_dependencies: [types-PyYAML]
args: ["--ignore-missing-imports", "--disable-error-code=unused-ignore"] args: ["--ignore-missing-imports", "--scripts-are-modules"]
files: "^(elevator_saga/)" # Check both source code exclude: "^(tests/|examples/)"
# General pre-commit hooks # General pre-commit hooks
- repo: https://github.com/pre-commit/pre-commit-hooks - repo: https://github.com/pre-commit/pre-commit-hooks
@@ -46,16 +45,18 @@ repos:
- id: check-json - id: check-json
- id: check-toml - id: check-toml
- id: check-xml - id: check-xml
# Security # Security
- id: check-merge-conflict - id: check-merge-conflict
- id: check-case-conflict - id: check-case-conflict
- id: check-symlinks - id: check-symlinks
- id: check-added-large-files - id: check-added-large-files
args: ["--maxkb=1000"] args: ["--maxkb=1000"]
# Python specific # Python specific
- id: check-ast - id: check-ast
- id: check-builtin-literals
- id: check-docstring-first
- id: debug-statements - id: debug-statements
- id: name-tests-test - id: name-tests-test
args: ["--django"] args: ["--django"]
@@ -66,17 +67,47 @@ repos:
hooks: hooks:
- id: bandit - id: bandit
args: ["-c", "pyproject.toml"] args: ["-c", "pyproject.toml"]
additional_dependencies: ["bandit[toml]", "pbr"] additional_dependencies: ["bandit[toml]"]
exclude: "^tests/" 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 # YAML/JSON formatting
- repo: https://github.com/pre-commit/mirrors-prettier - repo: https://github.com/pre-commit/mirrors-prettier
rev: v4.0.0-alpha.8 rev: v4.0.0-alpha.8
hooks: hooks:
- id: prettier - id: prettier
types_or: [yaml, json, markdown] types_or: [yaml, json, markdown]
exclude: "^(build/|dist/|__pycache__/|\\.mypy_cache/|\\.pytest_cache/|htmlcov/|\\.idea/|\\.vscode/|docs/_build/|elevatorpy\\.egg-info/|elevator_saga\\.egg-info/)" 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)$"
# Global settings # Global settings
default_stages: [pre-commit, pre-push] default_stages: [commit, push]
fail_fast: false 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
View File

@@ -1,21 +0,0 @@
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.

84
README.md
View File

@@ -1,84 +0,0 @@
# 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
```
### With Development Dependencies
```bash
pip install elevator-py[dev]
```
### From Source
```bash
git clone https://github.com/ZGCA-Forge/Elevator.git
cd Elevator
pip install -e .[dev]
```
## 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>

0
elevator_saga/client_examples/__init__.py → README_CN.md
View File

3
docs/.gitignore vendored
View File

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

20
docs/Makefile
View File

@@ -1,20 +0,0 @@
# 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
View File

@@ -1,61 +0,0 @@
# 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
View File

@@ -1,39 +0,0 @@
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:

370
docs/client.rst
View File

@@ -1,370 +0,0 @@
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
Performance Considerations
--------------------------
Each attribute access triggers an API call. For better performance:
.. code-block:: python
# ❌ Inefficient - multiple API calls
if elevator.current_floor < elevator.target_floor:
diff = elevator.target_floor - elevator.current_floor
# ✅ Better - store references
current = elevator.current_floor
target = elevator.target_floor
if current < target:
diff = target - current
However, the API client implements **caching within a single tick**, so multiple accesses during event processing are efficient.
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

544
docs/communication.rst
View File

@@ -1,544 +0,0 @@
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.
Performance Considerations
--------------------------
**Minimize HTTP Calls**:
.. code-block:: python
# ❌ Bad - 3 HTTP calls
for i in range(3):
state = api_client.get_state()
print(state.tick)
# ✅ Good - 1 HTTP call (cached)
state = api_client.get_state()
for i in range(3):
print(state.tick)
**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
View File

@@ -1,59 +0,0 @@
# 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

592
docs/events.rst
View File

@@ -1,592 +0,0 @@
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
-------------------
Command vs. Execution
~~~~~~~~~~~~~~~~~~~~~
.. code-block:: python
# Tick 42: Controller sends command
elevator.go_to_floor(5, immediate=False)
# ← Command queued in elevator.next_target_floor
# Tick 43: Server processes
# ← _update_elevator_status() assigns target
# ← Elevator starts moving
# Tick 44-46: Elevator in motion
# ← Events: PASSING_FLOOR
# Tick 47: Elevator arrives
# ← Event: STOPPED_AT_FLOOR
There's a **one-tick delay** between command and execution (unless ``immediate=True``).
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)
Best Practices
--------------
1. **React to Events**: Don't poll state - implement event handlers
2. **Use Queued Commands**: Default ``immediate=False`` is safer
3. **Track Passengers**: Monitor ``on_passenger_call`` to know demand
4. **Optimize for Wait Time**: Reduce time between arrival and pickup
5. **Consider Load**: Check ``elevator.is_full`` before dispatching
6. **Handle Idle**: Always give idle elevators something to do (even if it's "go to floor 0")
Example: Efficient Dispatch
~~~~~~~~~~~~~~~~~~~~~~~~~~~~
.. code-block:: python
def on_passenger_call(self, passenger: ProxyPassenger, floor: ProxyFloor, direction: str) -> None:
"""Dispatch nearest suitable elevator"""
best_elevator = None
best_cost = float('inf')
for elevator in self.elevators:
# Skip if full
if elevator.is_full:
continue
# Calculate cost (distance + current load)
distance = abs(elevator.current_floor - floor.floor)
load_penalty = elevator.load_factor * 10
cost = distance + load_penalty
# Check if going in right direction
if elevator.target_floor_direction.value == direction:
cost *= 0.5 # Prefer elevators already going that way
if cost < best_cost:
best_cost = cost
best_elevator = elevator
if best_elevator:
best_elevator.go_to_floor(floor.floor)
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

122
docs/index.rst
View File

@@ -1,122 +0,0 @@
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
With Development Dependencies
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
.. code-block:: bash
pip install elevator-py[dev]
From Source
~~~~~~~~~~~
.. code-block:: bash
git clone https://github.com/ZGCA-Forge/Elevator.git
cd Elevator
pip install -e .[dev]
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
View File

@@ -1,347 +0,0 @@
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
View File

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

2
elevator_saga/__init__.py
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. realistic elevator dispatch algorithm development and testing.
""" """
__version__ = "0.0.3" __version__ = "1.0.0"
__author__ = "ZGCA Team" __author__ = "ZGCA Team"

16
elevator_saga/client/api_client.py
View File

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

40
elevator_saga/client/base_controller.py
View File

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

6
elevator_saga/client/proxy_models.py
View File

@@ -22,8 +22,6 @@ class ProxyFloor(FloorState):
"""获取 FloorState 实例""" """获取 FloorState 实例"""
state = self._api_client.get_state() state = self._api_client.get_state()
floor_data = next((f for f in state.floors if f.floor == self._floor_id), None) 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 return floor_data
def __getattribute__(self, name: str) -> Any: def __getattribute__(self, name: str) -> Any:
@@ -68,8 +66,6 @@ class ProxyElevator(ElevatorState):
# 获取当前状态 # 获取当前状态
state = self._api_client.get_state() state = self._api_client.get_state()
elevator_data = next((e for e in state.elevators if e.id == self._elevator_id), None) 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 return elevator_data
def __getattribute__(self, name: str) -> Any: def __getattribute__(self, name: str) -> Any:
@@ -117,8 +113,6 @@ class ProxyPassenger(PassengerInfo):
"""获取 PassengerInfo 实例""" """获取 PassengerInfo 实例"""
state = self._api_client.get_state() state = self._api_client.get_state()
passenger_data = state.passengers.get(self._passenger_id) 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 return passenger_data
def __getattribute__(self, name: str) -> Any: def __getattribute__(self, name: str) -> Any:

35
elevator_saga/core/models.py
View File

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

15
elevator_saga/scripts/client_examples/simple_example.py
View File

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

18
elevator_saga/server/simulator.py
View File

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

2
elevator_saga/traffic/down_peak.json
View File

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

2
elevator_saga/traffic/fire_evacuation.json
View File

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

29
elevator_saga/traffic/generators.py
View File

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

2
elevator_saga/traffic/high_density.json
View File

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

2
elevator_saga/traffic/inter_floor.json
View File

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

2
elevator_saga/traffic/lunch_rush.json
View File

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

2
elevator_saga/traffic/medical.json
View File

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

2
elevator_saga/traffic/meeting_event.json
View File

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

2
elevator_saga/traffic/mixed_scenario.json
View File

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

2
elevator_saga/traffic/progressive_test.json
View File

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

2
elevator_saga/traffic/random.json
View File

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

2
elevator_saga/traffic/up_peak.json
View File

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

143
pyproject.toml
View File

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

61
pyrightconfig.json
View File

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

157
run_all_tests.py Normal file
View File

@@ -0,0 +1,157 @@
#!/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())

59
setup.py Normal file
View File

@@ -0,0 +1,59 @@
#!/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,
)

15
elevator_saga/client_examples/simple_example.py → simple_example.py
View File

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

17
elevator_saga/client_examples/bus_example.py → test_example.py
View File

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

3
tests/__init__.py
View File

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

73
tests/test_imports.py
View File

@@ -1,73 +0,0 @@
"""
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"])