Source code for Karana.KUtils._PythonReentryMonitor

# Copyright (c) 2024-2026 Karana Dynamics Pty Ltd. All rights reserved.
#
# NOTICE TO USER:
#
# This source code and/or documentation (the "Licensed Materials") is
# the confidential and proprietary information of Karana Dynamics Inc.
# Use of these Licensed Materials is governed by the terms and conditions
# of a separate software license agreement between Karana Dynamics and the
# Licensee ("License Agreement"). Unless expressly permitted under that
# agreement, any reproduction, modification, distribution, or disclosure
# of the Licensed Materials, in whole or in part, to any third party
# without the prior written consent of Karana Dynamics is strictly prohibited.
#
# THE LICENSED MATERIALS ARE PROVIDED "AS IS" WITHOUT WARRANTY OF ANY KIND.
# KARANA DYNAMICS DISCLAIMS ALL WARRANTIES, EXPRESS OR IMPLIED, INCLUDING
# BUT NOT LIMITED TO WARRANTIES OF MERCHANTABILITY, NON-INFRINGEMENT, AND
# FITNESS FOR A PARTICULAR PURPOSE.
#
# IN NO EVENT SHALL KARANA DYNAMICS BE LIABLE FOR ANY DAMAGES WHATSOEVER,
# INCLUDING BUT NOT LIMITED TO LOSS OF PROFITS, DATA, OR USE, EVEN IF
# ADVISED OF THE POSSIBILITY OF SUCH DAMAGES, WHETHER IN CONTRACT, TORT,
# OR OTHERWISE ARISING OUT OF OR IN CONNECTION WITH THE LICENSED MATERIALS.
#
# U.S. Government End Users: The Licensed Materials are a "commercial item"
# as defined at 48 C.F.R. 2.101, and are provided to the U.S. Government
# only as a commercial end item under the terms of this license.
#
# Any use of the Licensed Materials in individual or commercial software must
# include, in the user documentation and internal source code comments,
# this Notice, Disclaimer, and U.S. Government Use Provision.

import traceback
from sys import setprofile, getprofile




class PythonReentryMonitor:
    """Monitor when you re-enter Python from C++.

    This class is designed to help you identify places in your simulation where you are
    re-entering the Python interpreter from C++. This can help identify places in your
    simulation where you can improve performance by migrating to C++.
    """

    def __init__(self):
        self._depth = 0
        self.traces = []



    def trace(self, frame, event, _):
        """Trace the current line of Python code.

        This will track any call event. This does not count recursive events, so it will
        only track call events made at the top-level.

        Parameters
        ----------
        frame :
            The Python frame.
        event :
            The event being traced.
        _ :
            arg, unused.

        Returns
        -------
            The trace method:
        """
        if event == "call":
            if self._depth == 0:
                # Capture full stack trace at entry
                stack = traceback.format_stack(frame)
                self.traces.append(stack)

            self._depth += 1

        elif event == "return":
            self._depth -= 1

        return self.trace




    def __enter__(self):
        """Use this as a context.

        This will start monitoring all calls to Python in the context.

        Returns
        -------
        Self
            This object.
        """
        # Set the dept to 1, because the return from this __enter__ method will decrease the depth by 1
        self._depth = 1
        self._old_profile = getprofile()
        setprofile(self.trace)
        return self




    def __exit__(self, *_):
        """
        Exit the context.

        See __enter__ for details.

        Parameters
        ----------
        *_
            Exit context parameters. Unused.

        Returns
        -------
        bool
            True if this exited successfully.
        """
        setprofile(self._old_profile)

        # Pop the last trace, because __exit__ will be the most recent trace
        self.traces.pop(-1)
        return True




    def dump(self):
        """Dump the traces captured by this monitor."""
        print(f"There were {len(self.traces)} calls back into Python. Stack traces follow.")
        for k, t in enumerate(self.traces):
            print(f"Trace {k}:")
            print("".join(t))




    def clear(self):
        """Clear out all the traces this monitor has captured."""
        self.traces.clear()