307 lines
9.7 KiB
Python
Executable File
307 lines
9.7 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Integration script for custom IR protocols
|
|
This script helps integrate your custom protocol decoder into the existing IR system
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import json
|
|
import shutil
|
|
from pathlib import Path
|
|
|
|
def backup_existing_files():
|
|
"""Backup existing IR system files"""
|
|
backup_dir = Path("backup_ir_system")
|
|
backup_dir.mkdir(exist_ok=True)
|
|
|
|
files_to_backup = [
|
|
"ir_remote.py",
|
|
"simple_ir_listener.py",
|
|
"simple_ir_listener_polling.py",
|
|
"ir_listener.py"
|
|
]
|
|
|
|
for file in files_to_backup:
|
|
if os.path.exists(file):
|
|
shutil.copy2(file, backup_dir / file)
|
|
print(f"Backed up {file} to {backup_dir}")
|
|
|
|
return backup_dir
|
|
|
|
def update_ir_remote_with_custom_protocol():
|
|
"""Update ir_remote.py to include custom protocol"""
|
|
print("Updating ir_remote.py to include custom protocol...")
|
|
|
|
# Read current ir_remote.py
|
|
with open("ir_remote.py", "r") as f:
|
|
content = f.read()
|
|
|
|
# Add import for custom protocol
|
|
import_line = "from custom_ir_protocol import CustomIRProtocol"
|
|
|
|
if import_line not in content:
|
|
# Find the import section and add our import
|
|
lines = content.split('\n')
|
|
import_section_end = 0
|
|
|
|
for i, line in enumerate(lines):
|
|
if line.startswith('import ') or line.startswith('from '):
|
|
import_section_end = i + 1
|
|
|
|
lines.insert(import_section_end, import_line)
|
|
content = '\n'.join(lines)
|
|
|
|
# Update the IRRemote class to include custom protocol
|
|
if "CustomIRProtocol()" not in content:
|
|
# Find the protocols initialization line
|
|
old_line = "self.protocols = protocols or [NECProtocol(), RC5Protocol()]"
|
|
new_line = "self.protocols = protocols or [NECProtocol(), RC5Protocol(), CustomIRProtocol()]"
|
|
|
|
content = content.replace(old_line, new_line)
|
|
|
|
# Write updated content
|
|
with open("ir_remote.py", "w") as f:
|
|
f.write(content)
|
|
|
|
print("Updated ir_remote.py successfully")
|
|
|
|
def update_simple_listeners_with_custom_protocol():
|
|
"""Update simple listeners to include custom protocol"""
|
|
listeners = [
|
|
"simple_ir_listener.py",
|
|
"simple_ir_listener_polling.py"
|
|
]
|
|
|
|
for listener_file in listeners:
|
|
if not os.path.exists(listener_file):
|
|
continue
|
|
|
|
print(f"Updating {listener_file}...")
|
|
|
|
with open(listener_file, "r") as f:
|
|
content = f.read()
|
|
|
|
# Add import
|
|
import_line = "from custom_ir_protocol import CustomIRProtocol"
|
|
|
|
if import_line not in content:
|
|
lines = content.split('\n')
|
|
import_section_end = 0
|
|
|
|
for i, line in enumerate(lines):
|
|
if line.startswith('import ') or line.startswith('from '):
|
|
import_section_end = i + 1
|
|
|
|
lines.insert(import_section_end, import_line)
|
|
content = '\n'.join(lines)
|
|
|
|
# Update protocol lists
|
|
if "CustomIRProtocol()" not in content:
|
|
# Find and update protocol initialization
|
|
old_patterns = [
|
|
"protocols or [NECProtocol(), RC5Protocol()]",
|
|
"[NECProtocol(), RC5Protocol()]"
|
|
]
|
|
|
|
for old_pattern in old_patterns:
|
|
if old_pattern in content:
|
|
new_pattern = old_pattern.replace("RC5Protocol()]", "RC5Protocol(), CustomIRProtocol()]")
|
|
content = content.replace(old_pattern, new_pattern)
|
|
break
|
|
|
|
with open(listener_file, "w") as f:
|
|
f.write(content)
|
|
|
|
print(f"Updated {listener_file} successfully")
|
|
|
|
def create_custom_protocol_mapping():
|
|
"""Create a mapping file for custom protocol commands"""
|
|
mapping_file = "custom_ir_mapping.json"
|
|
|
|
if os.path.exists(mapping_file):
|
|
print(f"{mapping_file} already exists, skipping creation")
|
|
return mapping_file
|
|
|
|
# Create example mapping for custom protocol
|
|
custom_mapping = {
|
|
"CUSTOM_0000_0001": {
|
|
"command": "power_toggle",
|
|
"description": "Power on/off",
|
|
"repeatable": True
|
|
},
|
|
"CUSTOM_0000_0002": {
|
|
"command": "channel_1",
|
|
"description": "Channel 1",
|
|
"repeatable": False
|
|
},
|
|
"CUSTOM_0000_0003": {
|
|
"command": "channel_2",
|
|
"description": "Channel 2",
|
|
"repeatable": False
|
|
},
|
|
"CUSTOM_0000_0004": {
|
|
"command": "volume_up",
|
|
"description": "Volume up",
|
|
"repeatable": True
|
|
},
|
|
"CUSTOM_0000_0005": {
|
|
"command": "volume_down",
|
|
"description": "Volume down",
|
|
"repeatable": True
|
|
},
|
|
"REPEAT": {
|
|
"command": "repeat_last",
|
|
"description": "Repeat last command",
|
|
"repeatable": False
|
|
}
|
|
}
|
|
|
|
with open(mapping_file, "w") as f:
|
|
json.dump(custom_mapping, f, indent=2)
|
|
|
|
print(f"Created {mapping_file} with example mappings")
|
|
return mapping_file
|
|
|
|
def update_main_ir_mapping():
|
|
"""Update the main IR mapping file to include custom protocol mappings"""
|
|
main_mapping_files = [
|
|
"/etc/video_player/ir_mapping.json",
|
|
"ir_mapping.json"
|
|
]
|
|
|
|
custom_mapping_file = "custom_ir_mapping.json"
|
|
|
|
if not os.path.exists(custom_mapping_file):
|
|
print("Custom mapping file not found, creating it first...")
|
|
create_custom_protocol_mapping()
|
|
|
|
# Load custom mappings
|
|
with open(custom_mapping_file, "r") as f:
|
|
custom_mappings = json.load(f)
|
|
|
|
# Update main mapping files
|
|
for mapping_file in main_mapping_files:
|
|
if os.path.exists(mapping_file):
|
|
print(f"Updating {mapping_file}...")
|
|
|
|
with open(mapping_file, "r") as f:
|
|
main_mappings = json.load(f)
|
|
|
|
# Add custom mappings
|
|
main_mappings.update(custom_mappings)
|
|
|
|
with open(mapping_file, "w") as f:
|
|
json.dump(main_mappings, f, indent=2)
|
|
|
|
print(f"Updated {mapping_file} successfully")
|
|
|
|
def create_test_script():
|
|
"""Create a test script for the custom protocol"""
|
|
test_script = """#!/usr/bin/env python3
|
|
'''
|
|
Test script for custom IR protocol
|
|
'''
|
|
|
|
import sys
|
|
import os
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
from custom_ir_protocol import CustomIRProtocol
|
|
import json
|
|
|
|
def test_custom_protocol():
|
|
'''Test the custom protocol with captured signals'''
|
|
|
|
protocol = CustomIRProtocol("TEST_CUSTOM")
|
|
|
|
# Test with example signals (replace with your actual captured data)
|
|
test_signals = [
|
|
# Example: 34 pulses for NEC-like protocol
|
|
[(True, 0.009), (False, 0.0045), (True, 0.00056), (False, 0.00169)] * 8 + [(True, 0.00056), (False, 0.00056)] * 8,
|
|
# Add more test signals here
|
|
]
|
|
|
|
print("Testing custom protocol decoder...")
|
|
|
|
for i, signal in enumerate(test_signals):
|
|
print(f"\\nTest signal {i+1}:")
|
|
command = protocol.decode(signal)
|
|
if command:
|
|
print(f" Decoded: {command}")
|
|
else:
|
|
print(f" Failed to decode")
|
|
|
|
# Analyze signal
|
|
pulse_times = [duration * 1000000 for _, duration in signal]
|
|
analysis = protocol.analyze_signal(pulse_times)
|
|
print(f" Analysis: {analysis['possible_structure']}")
|
|
|
|
if __name__ == "__main__":
|
|
test_custom_protocol()
|
|
"""
|
|
|
|
with open("test_custom_protocol.py", "w") as f:
|
|
f.write(test_script)
|
|
|
|
os.chmod("test_custom_protocol.py", 0o755)
|
|
print("Created test_custom_protocol.py")
|
|
|
|
def main():
|
|
"""Main integration function"""
|
|
print("=" * 60)
|
|
print("CUSTOM IR PROTOCOL INTEGRATION")
|
|
print("=" * 60)
|
|
|
|
# Check if custom protocol file exists
|
|
if not os.path.exists("custom_ir_protocol.py"):
|
|
print("Error: custom_ir_protocol.py not found!")
|
|
print("Please create and customize your protocol decoder first.")
|
|
return False
|
|
|
|
try:
|
|
# Backup existing files
|
|
print("\\n1. Backing up existing files...")
|
|
backup_dir = backup_existing_files()
|
|
|
|
# Update IR remote
|
|
print("\\n2. Updating IR remote system...")
|
|
update_ir_remote_with_custom_protocol()
|
|
|
|
# Update simple listeners
|
|
print("\\n3. Updating simple listeners...")
|
|
update_simple_listeners_with_custom_protocol()
|
|
|
|
# Create custom mapping
|
|
print("\\n4. Creating custom protocol mapping...")
|
|
create_custom_protocol_mapping()
|
|
|
|
# Update main mappings
|
|
print("\\n5. Updating main IR mappings...")
|
|
update_main_ir_mapping()
|
|
|
|
# Create test script
|
|
print("\\n6. Creating test script...")
|
|
create_test_script()
|
|
|
|
print("\\n" + "=" * 60)
|
|
print("INTEGRATION COMPLETE!")
|
|
print("=" * 60)
|
|
print("\\nNext steps:")
|
|
print("1. Customize the timing constants in custom_ir_protocol.py")
|
|
print("2. Update the command mappings in custom_ir_mapping.json")
|
|
print("3. Test with: python3 test_custom_protocol.py")
|
|
print("4. Test with real remote: python3 simple_ir_listener_polling.py")
|
|
print("\\nBackup files are in:", backup_dir)
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"\\nError during integration: {e}")
|
|
print("\\nYou can restore from backup if needed.")
|
|
return False
|
|
|
|
if __name__ == "__main__":
|
|
success = main()
|
|
sys.exit(0 if success else 1)
|