112 lines
3.9 KiB
Python
112 lines
3.9 KiB
Python
# -*- coding: utf-8 -*-
|
|
import logging
|
|
import sys
|
|
import os
|
|
from typing import Any, Dict
|
|
|
|
from pymsc.core.base_module import BaseModule
|
|
from pymsc.core.message_definitions import messages_1553
|
|
from pymsc.utils.profiler import monitor_execution
|
|
|
|
class Bus1553Module(BaseModule):
|
|
"""
|
|
Module responsible for 1553 communication.
|
|
Handles internal library aliasing to support moved packages.
|
|
"""
|
|
|
|
def __init__(self):
|
|
super().__init__(name="Bus1553")
|
|
self.logger = logging.getLogger('PyMsc')
|
|
self.core = None
|
|
|
|
def _setup_library_aliasing(self):
|
|
"""
|
|
Creates a symbolic link in sys.modules to allow the library
|
|
to find its own sub-packages using the original name.
|
|
"""
|
|
try:
|
|
# 1. Import the sub-package using the current real path
|
|
import pymsc.PyBusMonitor1553 as real_package
|
|
|
|
# 2. Alias it to the name expected by internal library imports
|
|
# This solves the "No module named 'pybusmonitor1553'" error
|
|
sys.modules['pybusmonitor1553'] = real_package
|
|
|
|
self.logger.debug("Successfully aliased PyBusMonitor1553 to pybusmonitor1553")
|
|
return True
|
|
except ImportError as e:
|
|
self.logger.error(f"Failed to setup library aliasing: {e}")
|
|
return False
|
|
|
|
def initialize(self, config: Dict[str, Any]) -> bool:
|
|
"""
|
|
Initializes the low-level core and binds message wrappers.
|
|
"""
|
|
# Set up the alias before any other library operation
|
|
if not self._setup_library_aliasing():
|
|
return False
|
|
|
|
try:
|
|
# Now we can import the core using the aliased name or the real one
|
|
from pybusmonitor1553.core.bus_monitor_core import BusMonitorCore
|
|
self.core = BusMonitorCore()
|
|
self.logger.info("BusMonitorCore instance created.")
|
|
except ImportError as e:
|
|
self.logger.error(f"Critical: Could not load BusMonitorCore. {e}")
|
|
return False
|
|
|
|
# Internal library initialization (where it loads messages)
|
|
# Now it will find 'pybusmonitor1553' in sys.modules
|
|
success = self.core.initialize(config)
|
|
|
|
if not success:
|
|
self.logger.error("BusMonitorCore internal initialization failed.")
|
|
return False
|
|
|
|
# Binding our local ctypes structures to the library's message wrappers
|
|
self._bind_messages()
|
|
|
|
return True
|
|
|
|
def _bind_messages(self):
|
|
"""
|
|
Connects local message instances to the hardware core wrappers.
|
|
"""
|
|
for msg_instance in messages_1553:
|
|
msg_id = msg_instance.message_id
|
|
# Retrieve the wrapper from the hardware core
|
|
wrapper = self.core.get_message(msg_id)
|
|
|
|
if wrapper:
|
|
msg_instance.bind_to_wrapper(wrapper)
|
|
self.logger.debug(f"Bound message {msg_id} to hardware wrapper.")
|
|
else:
|
|
self.logger.warning(f"Hardware wrapper for {msg_id} not available.")
|
|
|
|
@monitor_execution
|
|
def start_session(self) -> None:
|
|
if self.core:
|
|
self.core.start_session()
|
|
self.is_running = True
|
|
self.logger.info("1553 communication session started.")
|
|
|
|
@monitor_execution
|
|
def stop_session(self) -> None:
|
|
if self.core:
|
|
self.core.stop_session()
|
|
self.is_running = False
|
|
self.logger.info("1553 communication session stopped.")
|
|
|
|
def get_status(self) -> Dict[str, Any]:
|
|
if self.core:
|
|
return self.core.get_status()
|
|
return {"error": "Not initialized"}
|
|
|
|
@monitor_execution
|
|
def sync_all_messages(self) -> None:
|
|
"""Fetch latest data from the bus for all bound messages."""
|
|
if not self.is_running:
|
|
return
|
|
|
|
from pymsc.core.message_definitions import update_all_1553_messages
|
|
update_all_1553_messages() |