#!/usr/bin/env python3 """ LEGO Dimensions + Moonlight Drive-In Integration Integrates the LEGO Dimensions portal reader with the Moonlight Drive-In video player system using NFC card IDs (10-digit decimal format). When a LEGO Dimensions disc is placed on the portal, it sends the card ID to the player's API to trigger video playback. v1.3.0 Changes: - Uses correct API endpoint: POST /api/video/play - Sends JSON body: {"nfc_id": "card_id"} - Compatible with Moonlight web_interface.py """ import sys import time import json import requests from typing import Dict, Optional from lego_dimensions_reader import ( LegoDimensionsReader, TagInfo, Pad, COLORS ) __version__ = "1.3.0" class MoonlightDimensionsClient: """ Integration client connecting LEGO Dimensions portal to Moonlight Drive-In video player. Sends NFC card ID to /api/video/play endpoint to trigger video playback based on the player's mapping configuration. """ def __init__(self, api_url: str, mapping_file: Optional[str] = None): """ Initialize the integration client. Args: api_url: Base URL for Moonlight Drive-In API (e.g., "http://100.94.163.117:8547") mapping_file: Optional JSON file for local display of mappings """ self.api_url = api_url.rstrip('/') self.reader = LegoDimensionsReader() # Optional local mapping for display purposes only # The actual video mappings are configured on the player side self.video_mapping: Dict[str, str] = {} if mapping_file: self._load_mapping(mapping_file) # Set up callbacks self.reader.on_tag_insert = self._on_tag_insert self.reader.on_tag_remove = self._on_tag_remove self.reader.on_connect = self._on_connect self.reader.on_disconnect = self._on_disconnect self.reader.on_error = self._on_error # Track currently playing self._current_card_id: Optional[str] = None def _load_mapping(self, filepath: str): """Load card ID to video mapping from JSON file (optional, for display).""" try: with open(filepath, 'r') as f: data = json.load(f) raw_mappings = data.get('mappings', data) for key, value in raw_mappings.items(): self.video_mapping[str(key)] = value print(f"Loaded {len(self.video_mapping)} local video mappings") except FileNotFoundError: pass # Mapping file is optional except json.JSONDecodeError as e: print(f"Warning: Invalid JSON in mapping file: {e}") def _on_connect(self): """Handle portal connection.""" print("\n✓ LEGO Dimensions Portal Connected") print(" Ready to detect discs...") # Flash all pads blue to indicate ready for pad in [Pad.CENTER, Pad.LEFT, Pad.RIGHT]: self.reader.flash_pad(pad, COLORS['BLUE'], count=2) def _on_disconnect(self): """Handle portal disconnection.""" print("\n✗ Portal Disconnected") def _on_error(self, error: Exception): """Handle errors.""" print(f"\n⚠ Error: {error}") def _on_tag_insert(self, tag: TagInfo): """Handle tag placement - send card ID to player.""" # Get the card ID (use the decimal format the player expects) card_id = str(tag.nfc_card_id) # e.g., "983187584" card_id_padded = tag.nfc_card_id_str # e.g., "0983187584" (10 digits) print(f"\n{'='*50}") print(f"✓ TAG DETECTED on {tag.pad.name} pad") print(f" UID: {tag.uid_hex}") print(f" Card ID: {card_id}") print(f" Card ID (padded): {card_id_padded}") # Send card ID to player API success = self._play_video(card_id) if success: # Set pad to green to indicate success self.reader.set_pad_color(tag.pad, COLORS['GREEN']) else: # Flash red to indicate error self.reader.flash_pad(tag.pad, COLORS['RED'], count=3) # Show local mapping info if available video_path = self.video_mapping.get(card_id) or \ self.video_mapping.get(card_id_padded) if video_path: print(f" Local mapping: {video_path}") print(f"{'='*50}") def _on_tag_remove(self, tag: TagInfo): """Handle tag removal.""" card_id = str(tag.nfc_card_id) print(f"\n✗ TAG REMOVED from {tag.pad.name} pad") print(f" Card ID: {card_id}") # Turn off pad LED self.reader.set_pad_color(tag.pad, COLORS['OFF']) self._current_card_id = None def _play_video(self, card_id: str) -> bool: """ Send play command to Moonlight API. Uses POST /api/video/play with JSON body {"nfc_id": "card_id"} Args: card_id: The NFC card ID (decimal string) Returns: True if successful, False otherwise """ try: response = requests.post( f"{self.api_url}/api/video/play", json={"nfc_id": card_id}, headers={'Content-Type': 'application/json'}, timeout=5 ) if response.ok: result = response.json() if result.get('success'): self._current_card_id = card_id print(f" ▶ Playing video for card {card_id}") return True else: message = result.get('message', 'Unknown error') print(f" ⚠ Player error: {message}") return False else: print(f" ⚠ API Error: {response.status_code}") return False except requests.exceptions.ConnectionError: print(f" ⚠ Cannot connect to player at {self.api_url}") return False except requests.exceptions.Timeout: print(f" ⚠ Request timed out") return False except Exception as e: print(f" ⚠ Error: {e}") return False def _skip_video(self) -> bool: """Send skip command to player.""" try: response = requests.post( f"{self.api_url}/api/video/skip", timeout=5 ) if response.ok: print(f" ⏭ Video skipped") return True except requests.exceptions.RequestException: pass return False def start(self): """Start the integration client.""" print("\n" + "="*50) print(" LEGO Dimensions + Moonlight Drive-In") print(f" Integration v{__version__}") print("="*50) print(f"\nPlayer URL: {self.api_url}") print(f"API Endpoint: POST /api/video/play") print(f"Request Format: {{\"nfc_id\": \"card_id\"}}") print("\nStarting portal connection...") self.reader.start() def stop(self): """Stop the integration client.""" self.reader.disconnect() print("\nClient stopped.") def main(): """Main entry point.""" # Configuration - UPDATE THIS TO YOUR PLAYER IP MOONLIGHT_API = "http://100.94.163.117:8547" MAPPING_FILE = "video_mappings.json" # Optional local reference # Allow command line override of API URL if len(sys.argv) > 1: MOONLIGHT_API = sys.argv[1] print(f"Using API URL from command line: {MOONLIGHT_API}") client = MoonlightDimensionsClient( api_url=MOONLIGHT_API, mapping_file=MAPPING_FILE ) try: client.start() print("\nPlace LEGO Dimensions discs on the portal...") print("Press Ctrl+C to exit\n") while True: time.sleep(1) except KeyboardInterrupt: print("\n\nShutting down...") except ConnectionError as e: print(f"\nFailed to connect: {e}") return 1 finally: client.stop() return 0 if __name__ == "__main__": sys.exit(main())