"""
TrGEN Configuration Manager
This module provides utilities for exporting and importing TrGEN configurations
to/from JSON files with .trgen extension. It manages the complete lifecycle of
configuration files including creation, loading, saving, and validation.
The configuration files use JSON format with custom .trgen extension for
easy identification and maximum compatibility.
"""
import json
import os
from datetime import datetime
from typing import List, Optional
from .configuration import TrgenConfiguration, TriggerPortConfig, PortProgrammingState
from .trgen_pin import TrgenPin
[docs]
class TrgenConfigurationManager:
"""
Manages the export and import of TrGEN configurations in JSON format.
Supports files with custom .trgen extension for easy identification.
Uses native JSON for maximum compatibility without external dependencies.
"""
[docs]
@staticmethod
def export_configuration(client, file_path: str, project_name: str = "",
description: str = "", author: str = "") -> str:
"""
Export the current TrGEN client configuration to a .trgen file (JSON format).
Args:
client: TrgenClient instance to export configuration from
file_path: File path (without extension or with .trgen)
project_name: Name of the project/experiment
description: Description of the configuration
author: Author of the configuration
Returns:
str: Complete path of the saved file
Raises:
ValueError: If client is None
IOError: If file cannot be written
"""
if client is None:
raise ValueError("Client cannot be None")
# Ensure the file has .trgen extension
final_path = TrgenConfigurationManager._ensure_trgen_extension(file_path)
config = TrgenConfigurationManager._create_configuration_from_client(
client, project_name, description, author
)
try:
# Create directory if it doesn't exist
directory = os.path.dirname(final_path)
if directory and not os.path.exists(directory):
os.makedirs(directory)
# Serialize to JSON (pretty printed for readability)
json_content = json.dumps(config.to_dict(), indent=2)
# Add custom header
header = TrgenConfigurationManager._generate_file_header(config)
final_content = header + json_content
# Save the file
with open(final_path, 'w', encoding='utf-8') as f:
f.write(final_content)
print(f"[TRGEN] Configuration exported successfully: {final_path}")
return final_path
except Exception as e:
print(f"[TRGEN] Error during export: {e}")
raise
[docs]
@staticmethod
def import_configuration(client, file_path: str, apply_network_settings: bool = False) -> TrgenConfiguration:
"""
Import a configuration from a .trgen file and apply it to the client.
Args:
client: TrgenClient instance to apply configuration to
file_path: Path to the .trgen file to import
apply_network_settings: Whether to apply network settings
Returns:
TrgenConfiguration: The imported configuration
Raises:
ValueError: If client is None
FileNotFoundError: If configuration file doesn't exist
ValueError: If file doesn't have .trgen extension
ValueError: If configuration cannot be deserialized
"""
if client is None:
raise ValueError("Client cannot be None")
if not os.path.exists(file_path):
raise FileNotFoundError(f"Configuration file not found: {file_path}")
if not file_path.lower().endswith(".trgen"):
raise ValueError("File must have .trgen extension")
try:
# Read file content
with open(file_path, 'r', encoding='utf-8') as f:
file_content = f.read()
# Remove header if present
json_content = TrgenConfigurationManager._remove_file_header(file_content)
# Deserialize from JSON
config_dict = json.loads(json_content)
config = TrgenConfiguration.from_dict(config_dict)
if config is None:
raise ValueError("Unable to deserialize configuration from file")
# Apply configuration to client
TrgenConfigurationManager._apply_configuration_to_client(
client, config, apply_network_settings
)
print(f"[TRGEN] Configuration imported successfully from: {file_path}")
return config
except Exception as e:
print(f"[TRGEN] Error during import: {e}")
raise
[docs]
@staticmethod
def load_configuration(file_path: str) -> TrgenConfiguration:
"""
Load a configuration from file without applying it to any client.
Args:
file_path: Path to the .trgen file
Returns:
TrgenConfiguration: The loaded configuration
Raises:
FileNotFoundError: If configuration file doesn't exist
ValueError: If file doesn't have .trgen extension
ValueError: If configuration cannot be deserialized
"""
if not os.path.exists(file_path):
raise FileNotFoundError(f"Configuration file not found: {file_path}")
if not file_path.lower().endswith(".trgen"):
raise ValueError("File must have .trgen extension")
try:
with open(file_path, 'r', encoding='utf-8') as f:
file_content = f.read()
json_content = TrgenConfigurationManager._remove_file_header(file_content)
config_dict = json.loads(json_content)
config = TrgenConfiguration.from_dict(config_dict)
if config is None:
raise ValueError("Unable to deserialize configuration from file")
return config
except Exception as e:
print(f"[TRGEN] Error during loading: {e}")
raise
[docs]
@staticmethod
def save_configuration(config: TrgenConfiguration, file_path: str) -> str:
"""
Save a configuration to a .trgen file.
Args:
config: Configuration to save
file_path: Path to save the file to
Returns:
str: Complete path of the saved file
Raises:
IOError: If file cannot be written
"""
final_path = TrgenConfigurationManager._ensure_trgen_extension(file_path)
try:
directory = os.path.dirname(final_path)
if directory and not os.path.exists(directory):
os.makedirs(directory)
# Update modification timestamp
config.metadata.modified_at = datetime.now()
json_content = json.dumps(config.to_dict(), indent=2)
header = TrgenConfigurationManager._generate_file_header(config)
final_content = header + json_content
with open(final_path, 'w', encoding='utf-8') as f:
f.write(final_content)
print(f"[TRGEN] Configuration saved successfully: {final_path}")
return final_path
except Exception as e:
print(f"[TRGEN] Error during saving: {e}")
raise
[docs]
@staticmethod
def list_configuration_files(directory: str, recursive: bool = False) -> List[str]:
"""
List all .trgen files in a directory.
Args:
directory: Directory to search in
recursive: Whether to search recursively in subdirectories
Returns:
List[str]: List of .trgen file paths found
"""
if not os.path.exists(directory):
return []
trgen_files = []
if recursive:
for root, dirs, files in os.walk(directory):
for file in files:
if file.lower().endswith('.trgen'):
trgen_files.append(os.path.join(root, file))
else:
for file in os.listdir(directory):
if file.lower().endswith('.trgen') and os.path.isfile(os.path.join(directory, file)):
trgen_files.append(os.path.join(directory, file))
return trgen_files
[docs]
@staticmethod
def apply_port_configurations_selective(client, config: TrgenConfiguration, port_filters: List[str] = None) -> int:
"""
Apply configuration only to selected ports and program them to hardware.
Args:
client: TrgenClient instance to apply configuration to
config: Configuration containing port settings
port_filters: List of port keys to apply (e.g., ['NS0', 'SA1']). If None, applies all enabled ports.
Returns:
int: Number of ports successfully programmed
Raises:
ValueError: If client is None
"""
if client is None:
raise ValueError("Client cannot be None")
ports_to_apply = config.trigger_ports
if port_filters:
ports_to_apply = {k: v for k, v in config.trigger_ports.items() if k in port_filters}
print(f"[TRGEN] Applying selective configuration to ports: {', '.join(port_filters)}")
ports_programmed = 0
for port_key, port_config in ports_to_apply.items():
try:
if not port_config.enabled:
print(f"[TRGEN] Skipping disabled port {port_key}")
continue
# Create TrgenPort and apply configuration
trgen_port = client.createTrgenPort(port_config.id)
port_config.apply_to_trgen_port(trgen_port)
# Program the port to hardware
client.writeTrgenMemory(trgen_port)
ports_programmed += 1
if port_config.has_programmed_instructions():
print(f"[TRGEN] Programmed {port_key} with custom instructions: {port_config.get_instructions_string()}")
else:
print(f"[TRGEN] Programmed {port_key} with default configuration")
except Exception as e:
print(f"[TRGEN] Error programming port {port_key}: {e}")
print(f"[TRGEN] Selective configuration complete: {ports_programmed} ports programmed to hardware")
return ports_programmed
# Private Methods
@staticmethod
def _ensure_trgen_extension(file_path: str) -> str:
"""Ensure the file path has .trgen extension."""
if not file_path:
raise ValueError("File path cannot be empty")
return file_path if file_path.lower().endswith(".trgen") else file_path + ".trgen"
@staticmethod
def _create_configuration_from_client(client, project_name: str, description: str, author: str) -> TrgenConfiguration:
"""Create a configuration object from a TrgenClient."""
config = TrgenConfiguration()
# Set metadata
config.metadata.project_name = project_name
config.metadata.description = description
config.metadata.author = author
config.metadata.created_at = datetime.now()
config.metadata.modified_at = datetime.now()
# Set default settings - use the client's default timing if available
if hasattr(client, 'default_trigger_duration_us'):
config.defaults.default_trigger_duration_us = client.default_trigger_duration_us
else:
config.defaults.default_trigger_duration_us = 20 # Default value as per request
config.defaults.default_log_level = "warn"
# Create port configurations
TrgenConfigurationManager._create_port_configurations(config)
# Set network settings from client
config.network.ip_address = client.ip
config.network.port = client.port
config.network.timeout_ms = int(client.timeout * 1000)
return config
@staticmethod
def _create_port_configurations(config: TrgenConfiguration):
"""Create default port configurations for all available ports."""
# NeuroScan Ports (NS0-NS7)
for i in range(8):
port_config = TriggerPortConfig(32) # 32 is standard memory length
port_config.id = i
port_config.name = f"NeuroScan {i}"
port_config.type = "NeuroScan"
port_config.enabled = True
config.trigger_ports[f"NS{i}"] = port_config
# Synamps Ports (SA0-SA7)
for i in range(8):
port_config = TriggerPortConfig(32)
port_config.id = 8 + i
port_config.name = f"Synamps {i}"
port_config.type = "Synamps"
port_config.enabled = True
config.trigger_ports[f"SA{i}"] = port_config
# TMS Ports
bnco_config = TriggerPortConfig(32)
bnco_config.id = 16
bnco_config.name = "BNC Output"
bnco_config.type = "BNCO"
bnco_config.enabled = True
config.trigger_ports["BNCO"] = bnco_config
bnci_config = TriggerPortConfig(32)
bnci_config.id = 17
bnci_config.name = "BNC Input"
bnci_config.type = "BNCI"
bnci_config.enabled = True
config.trigger_ports["BNCI"] = bnci_config
# GPIO Ports (GPIO0-GPIO7)
for i in range(8):
port_config = TriggerPortConfig(32)
port_config.id = 18 + i
port_config.name = f"GPIO {i}"
port_config.type = "GPIO"
port_config.enabled = True
config.trigger_ports[f"GPIO{i}"] = port_config
@staticmethod
def _apply_configuration_to_client(client, config: TrgenConfiguration, apply_network_settings: bool):
"""Apply configuration to a TrgenClient."""
print(f"[TRGEN] Applying configuration: {config.metadata.project_name} (v{config.metadata.version})")
# Apply default settings
if hasattr(client, 'setDefaultDuration'):
client.setDefaultDuration(config.defaults.default_trigger_duration_us)
print(f"[TRGEN] Set default trigger duration to {config.defaults.default_trigger_duration_us} µs")
else:
# Set as attribute if method doesn't exist
client.default_trigger_duration_us = config.defaults.default_trigger_duration_us
print(f"[TRGEN] Set default trigger duration attribute to {config.defaults.default_trigger_duration_us} µs")
# Apply port configurations with memory instructions and program them to hardware
TrgenConfigurationManager._apply_port_configurations_to_client(client, config)
if apply_network_settings:
print("[TRGEN] Warning: Network settings require creating a new client to be applied.")
print(f"[TRGEN] Current network config - IP: {config.network.ip_address}, Port: {config.network.port}, Timeout: {config.network.timeout_ms}ms")
print(f"[TRGEN] Configuration '{config.metadata.project_name}' applied successfully")
@staticmethod
def _apply_port_configurations_to_client(client, config: TrgenConfiguration):
"""Apply port configurations including memory instructions to client."""
ports_applied = 0
ports_with_instructions = 0
ports_programmed = 0
for port_key, port_config in config.trigger_ports.items():
try:
# Create TrgenPort and apply configuration
trgen_port = client.createTrgenPort(port_config.id)
port_config.apply_to_trgen_port(trgen_port)
# Always program the port to the hardware device if it's enabled
if port_config.enabled:
client.writeTrgenMemory(trgen_port)
ports_programmed += 1
# Log additional info if port has specific instructions
if port_config.has_programmed_instructions():
ports_with_instructions += 1
print(f"[TRGEN] Programmed port {port_key} with custom instructions: {port_config.get_instructions_string()}")
else:
print(f"[TRGEN] Programmed port {port_key} with default configuration")
else:
print(f"[TRGEN] Skipped disabled port {port_key}")
ports_applied += 1
except Exception as e:
print(f"[TRGEN] Error applying configuration for {port_key}: {e}")
print(f"[TRGEN] Configuration summary: {ports_applied} ports configured, {ports_programmed} ports programmed to hardware, {ports_with_instructions} with custom instructions")
@staticmethod
def _generate_file_header(config: TrgenConfiguration) -> str:
"""Generate a custom header for the configuration file."""
return f"""# TrGEN Configuration File (JSON Format)
# Generated by trgenpy library
# Project: {config.metadata.project_name}
# Author: {config.metadata.author}
# Created: {config.metadata.created_at.strftime('%Y-%m-%d %H:%M:%S')}
# Description: {config.metadata.description}
#
# This file contains trigger configuration data for TrGEN hardware
# File format version: {config.metadata.version}
# Format: JSON (Python native compatibility)
#
"""
@staticmethod
def _remove_file_header(content: str) -> str:
"""Remove the custom header from file content."""
lines = content.split('\n')
start_index = 0
# Find the first line that doesn't start with # or is empty
for i, line in enumerate(lines):
line = line.strip()
if not line.startswith('#') and line:
start_index = i
break
return '\n'.join(lines[start_index:])