Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion .github/workflows/python-app.yml
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
# This workflow will install Python dependencies, run tests and lint with a single version of Python
# For more information see: https://docs.github.com/en/actions/automating-builds-and-tests/building-and-testing-python

name: Application Build
name: Python application

on:
pull_request:
Expand Down
4 changes: 2 additions & 2 deletions .github/workflows/release.yml
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,7 @@ jobs:
if: steps.version.outputs.has_changes != '0'
run: |
git add VERSION CHANGELOG.md
git commit -m "chore: v${{ steps.version.outputs.version }} [skip ci]"
git commit -m "chore: release v${{ steps.version.outputs.version }} [skip ci]"
git push origin main

- name: Create Git Tag
Expand All @@ -65,7 +65,7 @@ jobs:
uses: softprops/action-gh-release@v2
with:
tag_name: "v${{ steps.version.outputs.version }}"
name: "v${{ steps.version.outputs.version }}"
name: "release v${{ steps.version.outputs.version }}"
body_path: RELEASE_NOTES.md
draft: false
prerelease: false
Expand Down
21 changes: 0 additions & 21 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -14,27 +14,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0






## [1.1.2] - 2025-10-27

### Changed
- Fix/brainwave reading (#11) (055584f)

## [1.1.1] - 2025-10-25

### Fixed
- Format indentation fix(indentation) (685700a)

### Security
- Update high vulnerability (#10)(tensorflow) (ca02989)

## [1.1.0] - 2025-10-24

### Added
- Automate release (#9)(semver) (1ebea7c)

## [1.0.7] - 2025-10-24

### Changed
Expand Down
245 changes: 27 additions & 218 deletions GUI5_BrainwaveReading/GUI5_BrainwaveReading.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,236 +6,59 @@
from PySide6.QtCore import QObject, Signal, Slot
from PySide6.QtGui import QGuiApplication
from PySide6.QtQml import QQmlApplicationEngine
from djitellopy import TelloException

# Add parent directory to path to import BrainwavesBackend from GUI5.py
parent_dir = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(parent_dir))

from GUI5 import BrainwavesBackend
class BrainwavesBackend(QObject):
# Define signals to update QML components
flightLogUpdated = Signal(list)
predictionsTableUpdated = Signal(list)


def normalize_bci_label(label):
"""
Normalize BCI prediction labels to drone action commands.
Converts various label formats to standardized lowercase action strings.

Args:
label (str): The raw BCI prediction label

Returns:
str: Normalized drone action command in lowercase
"""
if not label:
return ""

# Convert to lowercase for standardization
label_lower = label.lower().strip()

# Map label variations to drone actions
label_mapping = {
"move forward": "forward",
"move backward": "backward",
"move left": "left",
"move right": "right",
"move up": "up",
"move down": "down",
"take off": "takeoff",
"landing": "land",
# Direct mappings (already correct)
"forward": "forward",
"backward": "backward",
"left": "left",
"right": "right",
"up": "up",
"down": "down",
"takeoff": "takeoff",
"land": "land",
"turn_left": "turn_left",
"turn_right": "turn_right",
}

normalized = label_mapping.get(label_lower, "")
if not normalized:
print(f"Warning: Unknown BCI label '{label}' - ignoring")
return normalized



class BrainwaveReadingBackend(BrainwavesBackend):
"""
Extended BrainwavesBackend specifically for the BrainwaveReading module.
Overrides key methods to integrate BCI label normalization and Tello execution.
"""

def __init__(self, mock_mode=True, bci_connection=None):
"""
Initialize the BrainwaveReading backend.

Args:
mock_mode (bool): If True, uses mock predictions for testing.
If False, requires actual BCI connection.
bci_connection: BCI connection object for real brainwave reading.
Required when mock_mode=False.
"""
def __init__(self):
super().__init__()
self.mock_mode = mock_mode
self.bci_connection = bci_connection
self.mock_predictions = ["forward", "backward", "left", "right", "takeoff", "land"]
self.mock_index = 0

if not mock_mode and bci_connection is None:
print("Warning: mock_mode=False but no BCI connection provided. Falling back to mock mode.")
self.mock_mode = True

self.flight_log = [] # List to store flight log entries
self.predictions_log = [] # List to store prediction records
self.current_prediction_label = ""

@Slot()
def readMyMind(self):
"""
Read brainwave data and generate prediction.
Uses mock data if mock_mode=True, otherwise calls actual BCI system.
"""
if self.mock_mode:
# Cycle through mock predictions for testing
self.current_prediction_label = self.mock_predictions[self.mock_index]
self.mock_index = (self.mock_index + 1) % len(self.mock_predictions)
server_name = "Mock Server (Testing)"
else:
# TODO: Implement actual BCI prediction
# Example integration:
# try:
# prediction_response = self.bci_connection.use_brainflow()
# self.current_prediction_label = prediction_response["prediction_label"]
# server_name = "BCI Server"
# except Exception as e:
# self.logMessage.emit(f"BCI prediction failed: {e}")
# return
raise NotImplementedError(
"Real BCI prediction not yet implemented. "
"Set mock_mode=True or provide BCI connection implementation."
)

# Normalize the label for drone commands
normalized_label = normalize_bci_label(self.current_prediction_label)

# Mock function to simulate brainwave reading
self.current_prediction_label = "Move Forward"
# Update the predictions log
self.predictions_log.append(
{
"count": str(len(self.predictions_log) + 1),
"server": server_name,
"count": "1",
"server": "Prediction Server",
"label": self.current_prediction_label,
}
)
self.predictionsTableUpdated.emit(self.predictions_log)

# Log to flight log
mode_indicator = "[MOCK] " if self.mock_mode else ""
self.flight_log.insert(0, f"{mode_indicator}BCI Prediction: {self.current_prediction_label} → {normalized_label}")
self.flightLogUpdated.emit(self.flight_log)

@Slot(str)
def notWhatIWasThinking(self, manual_action):
"""
Handle manual action input when BCI prediction is incorrect.
Normalizes the manual input and executes it on the drone.
"""
if not manual_action or manual_action.strip() == "":
self.logMessage.emit("No manual action provided")
return

# Normalize the manual action
normalized_action = normalize_bci_label(manual_action)

if not normalized_action:
self.logMessage.emit(f"Unknown action '{manual_action}' - no drone command executed")
return

# Add to predictions log
# Handle manual action input
self.predictions_log.append(
{
"count": "manual",
"server": "manual",
"label": manual_action
}
{"count": "manual", "server": "manual", "label": manual_action}
)
self.predictionsTableUpdated.emit(self.predictions_log)

# Execute the manual action on drone
self.getDroneAction(normalized_action)

# Log the manual override
self.flight_log.insert(0, f"Manual override: {manual_action} → {normalized_action}")
self.flightLogUpdated.emit(self.flight_log)

@Slot()
def executeAction(self):
"""
Execute the current BCI prediction on the Tello drone.
Normalizes the label and sends it to getDroneAction().
"""
print(f"DEBUG: executeAction() called, current_prediction_label='{self.current_prediction_label}'")

if not self.current_prediction_label:
msg = "No prediction to execute - click 'Read my mind...' first"
print(f"DEBUG: {msg}")
self.logMessage.emit(msg)
self.flight_log.insert(0, msg)
self.flightLogUpdated.emit(self.flight_log)
return

# Normalize the BCI label to drone action format
normalized_label = normalize_bci_label(self.current_prediction_label)
print(f"DEBUG: Normalized '{self.current_prediction_label}' → '{normalized_label}'")

if not normalized_label:
msg = f"Cannot execute - unknown action '{self.current_prediction_label}'"
print(f"DEBUG: {msg}")
self.logMessage.emit(msg)
self.flight_log.insert(0, msg)
# Execute the current prediction
if self.current_prediction_label:
self.flight_log.insert(0, f"Executed: {self.current_prediction_label}")
self.flightLogUpdated.emit(self.flight_log)
return

print(f"DEBUG: Calling getDroneAction('{normalized_label}')")
# Execute on the drone
self.getDroneAction(normalized_label)

# Update flight log
log_msg = f"Executed: {self.current_prediction_label} → {normalized_label}"
print(f"DEBUG: {log_msg}")
self.flight_log.insert(0, log_msg)
self.flightLogUpdated.emit(self.flight_log)
self.logMessage.emit(f"Executed action: {normalized_label}")

@Slot()
def connectDrone(self):
"""
Connect to the Tello drone using the actual getDroneAction method.
"""
self.getDroneAction('connect')
self.flight_log.insert(0, "Connecting to drone...")
# Mock function to simulate drone connection
self.flight_log.insert(0, "Drone connected.")
self.flightLogUpdated.emit(self.flight_log)

@Slot()
def keepDroneAlive(self):
"""
Send keep-alive signal to maintain Tello connection.
Queries battery status to keep the connection active.
"""
if not self.is_connected:
self.logMessage.emit("Drone not connected. Cannot send keep-alive.")
self.flight_log.insert(0, "Keep-alive failed: Not connected")
self.flightLogUpdated.emit(self.flight_log)
return

try:
# Query battery to keep connection alive
battery = self.tello.query_battery()
self.logMessage.emit(f"Keep-alive sent. Battery: {battery}%")
self.flight_log.insert(0, f"Keep-alive: Battery {battery}%")
self.flightLogUpdated.emit(self.flight_log)
except (AttributeError, ConnectionError, TimeoutError, TelloException) as e:
self.logMessage.emit(f"Keep-alive error: {e}")
self.flight_log.insert(0, f"Keep-alive error: {e}")
self.flightLogUpdated.emit(self.flight_log)
# Mock function to simulate sending keep-alive signal
self.flight_log.insert(0, "Keep alive signal sent.")
self.flightLogUpdated.emit(self.flight_log)


if __name__ == "__main__":
Expand All @@ -245,36 +68,22 @@ def keepDroneAlive(self):
app = QGuiApplication(sys.argv)
engine = QQmlApplicationEngine()

# Determine if we should use mock mode or real BCI
# Check for --mock or --real command line argument
mock_mode = True # Default to mock mode for safety
if "--real" in sys.argv:
mock_mode = False
print("Starting in REAL BCI mode")
# TODO: Initialize actual BCI connection here
# bci_conn = bciConnection(...)
# backend = BrainwaveReadingBackend(mock_mode=False, bci_connection=bci_conn)
elif "--mock" in sys.argv or len(sys.argv) == 1:
print("Starting in MOCK mode (use --real for actual BCI)")

# Create the backend with full Tello integration
backend = BrainwaveReadingBackend(mock_mode=mock_mode)
engine.rootContext().setContextProperty("backend", backend)

# Load the QML file
qml_file = Path(__file__).resolve().parent / "GUI5_BrainwaveReading.qml"

# Check if the QML file exists
if not qml_file.exists():
print(f"Error: QML file not found at {qml_file}")
sys.exit(-1)

# Load the QML file
engine.load(str(qml_file))

# Check if the QML engine loaded successfully
if not engine.rootObjects():
print("Error: Failed to load QML file")
sys.exit(-1)

# Create and set the backend context
backend = BrainwavesBackend()
engine.rootContext().setContextProperty("backend", backend)

sys.exit(app.exec())
Loading