Skip to content

Commit

Permalink
FEAT: Add a Message Manager for App (#1068)
Browse files Browse the repository at this point in the history
Co-authored-by: pyansys-ci-bot <92810346+pyansys-ci-bot@users.noreply.github.com>
Co-authored-by: Dipin Nair <dkunhamb@chqmechlinux01.ansys.com>
  • Loading branch information
3 people authored Feb 10, 2025
1 parent 4c5d966 commit 6f71d2e
Show file tree
Hide file tree
Showing 4 changed files with 335 additions and 0 deletions.
1 change: 1 addition & 0 deletions doc/changelog.d/1068.added.md
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Add a Message Manager for App
10 changes: 10 additions & 0 deletions src/ansys/mechanical/core/embedding/app.py
Original file line number Diff line number Diff line change
Expand Up @@ -191,6 +191,7 @@ def __init__(self, db_file=None, private_appdata=False, **kwargs):
INSTANCES.append(self)
self._updated_scopes: typing.List[typing.Dict[str, typing.Any]] = []
self._subscribe()
self._messages = None

def __repr__(self):
"""Get the product info."""
Expand Down Expand Up @@ -437,6 +438,15 @@ def project_directory(self):
"""Returns the current project directory."""
return self.DataModel.Project.ProjectDirectory

@property
def messages(self):
"""Lazy-load the MessageManager."""
if self._messages is None:
from ansys.mechanical.core.embedding.messages import MessageManager

self._messages = MessageManager(self._app)
return self._messages

def _share(self, other) -> None:
"""Shares the state of self with other.
Expand Down
190 changes: 190 additions & 0 deletions src/ansys/mechanical/core/embedding/messages.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,190 @@
# Copyright (C) 2022 - 2025 ANSYS, Inc. and/or its affiliates.
# SPDX-License-Identifier: MIT
#
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.

"""Message Manager for App."""

# TODO: add functionality to filter only errors, warnings, info
# TODO: add max number of messages to display
# TODO: implement pep8 formatting

try: # noqa: F401
import pandas as pd

HAS_PANDAS = True
"""Whether or not pandas exists."""
except ImportError:
HAS_PANDAS = False


class MessageManager:
"""Message manager for adding, fetching, and printing messages."""

def __init__(self, app):
"""Initialize the message manager."""
self._app = app

# Import necessary classes
from Ansys.Mechanical.Application import Message
from Ansys.Mechanical.DataModel.Enums import MessageSeverityType

self._message_severity = MessageSeverityType
self._message = Message
self._messages = self._app.ExtAPI.Application.Messages

def _create_messages_data(self): # pragma: no cover
"""Update the local cache of messages."""
data = {
"Severity": [],
"TimeStamp": [],
"DisplayString": [],
"Source": [],
"StringID": [],
"Location": [],
"RelatedObjects": [],
}
for msg in self._app.ExtAPI.Application.Messages:
data["Severity"].append(str(msg.Severity).upper())
data["TimeStamp"].append(msg.TimeStamp)
data["DisplayString"].append(msg.DisplayString)
data["Source"].append(msg.Source)
data["StringID"].append(msg.StringID)
data["Location"].append(msg.Location)
data["RelatedObjects"].append(msg.RelatedObjects)

return data

def __repr__(self): # pragma: no cover
"""Provide a DataFrame representation of all messages."""
if not HAS_PANDAS:
return "Pandas is not available. Please pip install pandas to display messages."
data = self._create_messages_data()
return repr(pd.DataFrame(data))

def __str__(self):
"""Provide a custom string representation of the messages."""
if self._messages.Count == 0:
return "No messages to display."

formatted_messages = [f"[{msg.Severity}] : {msg.DisplayString}" for msg in self._messages]
return "\n".join(formatted_messages)

def __getitem__(self, index):
"""Allow indexed access to messages."""
if len(self._messages) == 0:
raise IndexError("No messages are available.")
if index >= len(self._messages) or index < 0:
raise IndexError("Message index out of range.")
return self._messages[index]

def __len__(self):
"""Return the number of messages."""
return self._messages.Count

def add(self, severity: str, text: str):
"""Add a message and update the cache.
Parameters
----------
severity : str
Severity of the message. Can be "info", "warning", or "error".
text : str
Message text.
Examples
--------
>>> app.messages.add("info", "User clicked the start button.")
"""
severity_map = {
"info": self._message_severity.Info,
"warning": self._message_severity.Warning,
"error": self._message_severity.Error,
}

if severity.lower() not in severity_map:
raise ValueError(f"Invalid severity: {severity}")

_msg = self._message(text, severity_map[severity.lower()])
self._messages.Add(_msg)

def remove(self, index: int):
"""Remove a message by index.
Parameters
----------
index : int
Index of the message to remove.
Examples
--------
>>> app.messages.remove(0)
"""
if index >= len(self._app.ExtAPI.Application.Messages) or index < 0:
raise IndexError("Message index out of range.")
_msg = self._messages[index]
self._messages.Remove(_msg)

def show(self, filter="Severity;DisplayString"):
"""Print all messages with full details.
Parameters
----------
filter : str, optional
Semicolon separated list of message attributes to display.
Default is "severity;message".
if filter is "*", all available attributes will be displayed.
Examples
--------
>>> app.messages.show()
... severity: info
... message: Sample message.
>>> app.messages.show(filter="time_stamp;severity;message")
... time_stamp: 1/30/2025 12:10:35 PM
... severity: info
... message: Sample message.
"""
if self._messages.Count == 0:
print("No messages to display.")
return

if filter == "*":
selected_columns = [
"TimeStamp",
"Severity",
"DisplayString",
"Source",
"StringID",
"Location",
"RelatedObjects",
]
else:
selected_columns = [col.strip() for col in filter.split(";")]

for msg in self._messages:
for key in selected_columns:
print(f"{key}: {getattr(msg, key, 'Specified attribute not found.')}")
print()

def clear(self):
"""Clear all messages."""
self._messages.Clear()
134 changes: 134 additions & 0 deletions tests/embedding/test_messages.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,134 @@
# Copyright (C) 2022 - 2025 ANSYS, Inc. and/or its affiliates.
# SPDX-License-Identifier: MIT
#
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.

"""Message manager test"""

import os
import re

import pytest


@pytest.mark.embedding
def test_message_manager(embedded_app, capsys):
"""Test message manager"""
# if license checkout takes time then there is a warning message
# get added to app. So, clear the messages before starting the test
embedded_app.messages.clear()
assert len(embedded_app.messages) == 0

print(embedded_app.messages)
captured = capsys.readouterr()
printed_output = captured.out.strip()
assert "No messages to display." in printed_output

embedded_app.messages.add("info", "Info message")

print(embedded_app.messages)
captured = capsys.readouterr()
printed_output = captured.out.strip()
assert "Info message" in printed_output


@pytest.mark.embedding
def test_message_add_and_clear(embedded_app):
"""Test adding and clearing messages"""
embedded_app.messages.clear()
assert len(embedded_app.messages) == 0

embedded_app.messages.add("info", "Info message")
assert len(embedded_app.messages) == 1
embedded_app.messages.add("warning", "Warning message")
assert len(embedded_app.messages) == 2
embedded_app.messages.add("error", "Error message")
assert len(embedded_app.messages) == 3

embedded_app.messages.remove(0)
assert len(embedded_app.messages) == 2

with pytest.raises(IndexError):
embedded_app.messages.remove(10)

with pytest.raises(ValueError):
embedded_app.messages.add("trace", "Trace message")


@pytest.mark.embedding
def test_message_show(embedded_app, capsys):
"""Test showing messages"""
embedded_app.messages.clear()
print(embedded_app.messages.show())
captured = capsys.readouterr()
printed_output = captured.out.strip()
assert "No messages to display." in printed_output

embedded_app.messages.add("info", "Info message")
embedded_app.messages.show()
captured = capsys.readouterr()
printed_output = captured.out.strip()
assert "Severity" in printed_output
assert "DisplayString" in printed_output
assert "Info message" in printed_output
embedded_app.messages.show(filter="TimeStamp")
captured = capsys.readouterr()
printed_output = captured.out.strip()
assert "TimeStamp" in printed_output

embedded_app.messages.show(filter="unknown")
captured = capsys.readouterr()
printed_output = captured.out.strip()
assert "Specified attribute not found" in printed_output


@pytest.mark.embedding
def test_message_get(embedded_app, assets, capsys):
"""Test getting a message"""
with pytest.raises(IndexError):
embedded_app.messages[10]

embedded_app.open(os.path.join(assets, "cube-hole.mechdb"))
_messages = embedded_app.messages
_msg1 = None
for _msg in _messages:
print(_msg.DisplayString)
if "Image file not found" in _msg.DisplayString:
_msg1 = _msg
break
assert _msg1 is not None, "Expected message not found in messages"

print(_msg1)
captured = capsys.readouterr()
printed_output = captured.out.strip()
assert "Ansys.Mechanical.Application.Message" in printed_output

assert str(_msg1.Severity) == "Warning"
assert "Image file not found" in _msg1.DisplayString
assert re.search(r"\d", str(_msg1.TimeStamp))
print(_msg1.StringID, _msg1.Source, _msg1.Location, _msg1.RelatedObjects)
captured = capsys.readouterr()
printed_output = captured.out.strip()
assert "Ansys.ACT.Automation.Mechanical.Image" in printed_output
assert "Ansys.Mechanical.DataModel.Interfaces.IDataModelObject" in printed_output
assert "Ansys.ACT.Core.Utilities.SelectionInf" in printed_output

with pytest.raises(IndexError):
embedded_app.messages[10]

0 comments on commit 6f71d2e

Please sign in to comment.