# Copyright (C) 2023 - 2026 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.
"""Module for writing fields to buffers."""
import copy
import dataclasses
import io
import logging
import typing
import hollerith as holler
import pandas as pd
import pandas._libs.missing as libmissing
from ansys.dyna.core.lib.field import Field, to_long
from ansys.dyna.core.lib.format_type import format_type
[docs]
logger = logging.getLogger(__name__)
def _write_string_right(value, width):
# TODO - put a left/right flag in hollerith
return f"{{0:>{width}}}".format(value)
def _write_null(width):
buffer = io.StringIO()
holler.write_spaces(buffer, width)
return buffer.getvalue()
def _field_iterator(fields: typing.List[Field], long_format: bool) -> typing.Iterable[Field]:
if len(fields) == 0:
return []
if fields[0].offset > 0:
# insert a blank field in the beginning up to the offset
blank_field = Field(None, None, 0, fields[0].offset)
fields = [blank_field] + fields
offset = 0
for field in fields:
if long_format:
field = to_long(field, offset)
pos, width = (field.offset, field.width)
# check pos, add a null if its not correct
if pos < offset:
raise ValueError(f"Field {field.name} has offset {pos} less than previous field offset {offset}.")
if pos != offset:
empty_width = pos - offset
yield Field(None, None, offset, empty_width)
offset += empty_width
if pos != offset:
raise ValueError(f"Field {field.name} has offset {pos} not equal to previous field offset {offset}.")
if dataclasses.is_dataclass(field.type):
field_offset = field.offset
total_width = 0
for member in dataclasses.fields(field.type):
if field.value != None:
value = getattr(field.value, member.name)
else:
value = None
yield Field(member.name, member.type, field_offset, field.width, value=value)
total_width += field.width
field_offset += field.width
offset += total_width
else:
yield field
offset += width
[docs]
def check_field_type(field_type: type):
"""Check that the field type is valid."""
if field_type != str and field_type != int and field_type != float:
raise TypeError(f"Unexpected field type: {field_type}. Expected str, int, or float.")
[docs]
def write_field_c(buf: typing.IO[typing.AnyStr], field_type: type, value: typing.Any, width: int) -> None:
"""Write a single field to the buffer."""
if libmissing.checknull(value):
holler.write_spaces(buf, width)
elif field_type == str:
holler.write_string(buf, value, width)
elif field_type == int:
holler.write_int(buf, value, width)
elif field_type == float:
holler.write_float(buf, value, width)
[docs]
def write_field(buf: typing.IO[typing.AnyStr], field_type: type, value: typing.Any, width: int) -> None:
"""Write a single field to the buffer."""
check_field_type(field_type)
write_field_c(buf, field_type, value, width)
[docs]
def write_c_dataframe(
buf: typing.IO[typing.AnyStr], fields: typing.List[Field], table: pd.DataFrame, format: format_type
) -> None:
"""Method to write a DataFrame to a buffer using hollerith format."""
def convert_field(field: Field) -> holler.Field:
"""Method to convert a Field to a holler.Field."""
return holler.Field(type=field.type, width=field.width)
long_format = format == format_type.long
converted_fields = list(_field_iterator(fields, long_format))
spec = [convert_field(field) for field in converted_fields]
num_defined_rows = len(table)
index = 0
any_none = False
for field in converted_fields:
if field.type == None:
any_none = True
if any_none:
full_table = pd.DataFrame()
for field in converted_fields:
if field.type == None:
values = [None] * num_defined_rows
index += 1
field.name = f"unused {index}"
else:
values = table[field.name]
full_table[field.name] = values
table = full_table
holler.write_table(buf, table, num_defined_rows, spec)
[docs]
def write_fields(
buf: typing.IO[typing.AnyStr],
fields: typing.List[Field],
values: typing.Optional[typing.List[typing.Any]] = None,
format: typing.Optional[format_type] = format_type.default,
) -> None:
"""Write `fields` representing a line of a keyword to `buf`.
Use the fixed column offsets and width
Parameters
----------
buf: IO
buffer to write to
fields: List
fields to write
values: List
optional - list of values for the field. If not set, use the value property of each field.
used by TableCard
format: format_type
optional - format to write
>>> s=io.String()
>>> fields = [
... Field("a", int, 0, 10, 1),
... Field("b", str, 10, 10, "hello")
... ]
>>> write_fields(s, fields)
>>> s.getvalue()
' 1 hello'
"""
if values != None:
fields = copy.deepcopy(fields)
for field, value in zip(fields, values):
field.value = value
for field in _field_iterator(fields, format == format_type.long):
if field.type is None:
buf.write(_write_null(field.width))
else:
field_value, field_type = field.io_info()
write_field(buf, field_type, field_value, field.width)
def _format_csv_value(value: typing.Any, field_type: type) -> str:
"""Format a single value for CSV output.
Parameters
----------
value : Any
The value to format.
field_type : type
The type of the field (int, float, str).
Returns
-------
str
The formatted string representation.
"""
if libmissing.checknull(value):
return ""
if field_type == int:
return str(int(value))
if field_type == float:
# Use repr for floats to preserve precision, but strip unnecessary trailing zeros
formatted = repr(float(value))
return formatted
if field_type == str:
return str(value)
return str(value)
[docs]
def write_fields_csv(
buf: typing.IO[typing.AnyStr],
fields: typing.List[Field],
values: typing.Optional[typing.List[typing.Any]] = None,
) -> None:
"""Write `fields` representing a line of a keyword to `buf` in comma-delimited format.
This produces LS-DYNA compatible comma-delimited (free format) output.
Parameters
----------
buf : IO
Buffer to write to.
fields : List[Field]
Fields to write.
values : List[Any], optional
List of values for the fields. If not set, use the value property of each field.
Examples
--------
>>> s = io.StringIO()
>>> fields = [
... Field("a", int, 0, 10, value=1),
... Field("b", str, 10, 10, value="hello")
... ]
>>> write_fields_csv(s, fields)
>>> s.getvalue()
'1,hello'
"""
logger.debug("Writing %d fields in CSV format", len(fields))
if values is not None:
fields = copy.deepcopy(fields)
for field, value in zip(fields, values):
field.value = value
csv_values = []
for field in fields:
if field.type is None:
csv_values.append("")
else:
field_value, field_type = field.io_info()
csv_values.append(_format_csv_value(field_value, field_type))
buf.write(",".join(csv_values))
[docs]
def write_c_dataframe_csv(
buf: typing.IO[typing.AnyStr],
fields: typing.List[Field],
table: pd.DataFrame,
) -> None:
"""Write a DataFrame to `buf` in comma-delimited format.
Parameters
----------
buf : IO
Buffer to write to.
fields : List[Field]
Field definitions for the columns.
table : pd.DataFrame
The data to write.
"""
logger.debug("Writing DataFrame with %d rows in CSV format", len(table))
for row_idx in range(len(table)):
csv_values = []
for field in fields:
if field.name in table.columns:
value = table[field.name].iloc[row_idx]
csv_values.append(_format_csv_value(value, field.type))
else:
csv_values.append("")
buf.write(",".join(csv_values))
buf.write("\n")
