diff --git a/pyproject.toml b/pyproject.toml index 25ea5a5..2c45e37 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -59,6 +59,7 @@ nbformat = "^5.10.4" ruff = "^0.11.12" mypy = "^1.16.0" pre-commit = "^4.2.0" +pytest = "^8.0" [tool.poetry.group.typing.dependencies] pandas-stubs = "2.2.2.240909" @@ -154,6 +155,14 @@ indent-style = "space" docstring-code-format = true +[tool.pytest.ini_options] +minversion = "8.0" +testpaths = ["tests"] +addopts = "-q" +filterwarnings = [ + "ignore::DeprecationWarning", +] + [tool.mypy] python_version = "3.13" warn_return_any = true diff --git a/src/eas_3d_pattern/parser.py b/src/eas_3d_pattern/parser.py index 7703180..37912d1 100644 --- a/src/eas_3d_pattern/parser.py +++ b/src/eas_3d_pattern/parser.py @@ -41,6 +41,18 @@ "Theta_Tilt": "Theta_Electrical_Tilt", } +# Coordinate transforms into the internal SPCS_Ericsson frame. +# Each entry maps (theta, phi) arrays -> (theta, phi) arrays. The phi operator +# (>= vs >) and the leading negation differ per system and are load-bearing: +# CW/Geo negate the wrapped value, so phi=180 maps consistently to -180 across +# all four systems. The dict keys also serve as the whitelist of source systems. +_TO_ERICSSON = { + "SPCS_Polar": lambda t, p: (t, np.where(p >= 180, p - 360, p)), + "SPCS_CW": lambda t, p: (t + 90, -np.where(p > 180, p - 360, p)), + "SPCS_CCW": lambda t, p: (t + 90, np.where(p >= 180, p - 360, p)), + "SPCS_Geo": lambda t, p: (np.flip(t), -np.where(p > 180, p - 360, p)), +} + class AntennaPattern: """Antenna pattern class to read, calculate and visualize JSON antenna pattern data. @@ -78,7 +90,21 @@ def __init__(self, data_filepath: str, validate: bool = False): self.raw_data: dict[str, Any] = self._normalize_json( self._load_data_from_file(data_filepath) ) - if validate and self._schema is not None: + if not self.raw_data.get("Data_Set"): + logger.error( + f"AntennaPattern: 'Data_Set' is empty or missing in {data_filepath}." + ) + raise ValueError( + f"AntennaPattern: 'Data_Set' is empty or missing in {data_filepath}." + ) + if validate: + if self._schema is None: + logger.error( + "AntennaPattern: Validation requested but no schema is available." + ) + raise ValueError( + "AntennaPattern: Validation requested but no schema is available." + ) self._validate_data_against_schema(self.raw_data, self._schema) # ---- Process the pattern data into one normalized format ---- @@ -142,12 +168,7 @@ def _validate_data_against_schema( error_path_str = ( " -> ".join(map(str, e.path)) if e.path else "document root" ) - full_error_message = ( - f"Antenna data validation FAILED for '{self.data_filepath}'.\n" - f"Schema source: '{NGMNSchema.source_message}'.\n" - f"Error at data path: '{error_path_str}'.\n" - f"Validation Message: {e.message} (Validator: '{e.validator}')" - ) + full_error_message = f"Antenna data validation FAILED for '{self.data_filepath}'.\nSchema source: '{NGMNSchema.source_message}'.\nError at data path: '{error_path_str}'.\nValidation Message: {e.message} (Validator: '{e.validator}')" logger.error(full_error_message) raise ValidationError(full_error_message) from e @@ -462,7 +483,7 @@ def _process_pattern_data(self) -> xr.Dataset: ) # coordinate system and grid - if self.coordinate_system not in DEFAULT_INTERNAL_COORD_SYSTEM: + if self.coordinate_system != DEFAULT_INTERNAL_COORD_SYSTEM: logger.warning( f"AntennaPattern: Coordinate system {self.coordinate_system} not used for calculations. Transforming 'Pattern_3D' attribute to {DEFAULT_INTERNAL_COORD_SYSTEM}." ) @@ -498,32 +519,38 @@ def _change_coordinate_system( raise NotImplementedError( f"Antenna Pattern: Change to coordinate system {to_system} not implemented yet. Use the default (SPCS_Ericsson) for now." ) + transformable_systems = tuple(_TO_ERICSSON) + if from_system not in _TO_ERICSSON: + logger.error( + f"AntennaPattern: Unsupported source coordinate system '{from_system}'. Expected one of {transformable_systems}." + ) + raise ValueError( + f"AntennaPattern: Unsupported source coordinate system '{from_system}'. Expected one of {transformable_systems}." + ) phi = Pattern_3D.coords["Phi"].values theta = Pattern_3D.coords["Theta"].values - if from_system == "SPCS_Polar": - if to_system == "SPCS_Ericsson": - Pattern_3D = Pattern_3D.assign_coords( - Theta=("Theta", theta), - Phi=("Phi", np.where(phi >= 180, phi - 360, phi)), - ) - if from_system == "SPCS_CW": - if to_system == "SPCS_Ericsson": - Pattern_3D = Pattern_3D.assign_coords( - Theta=("Theta", theta + 90), - Phi=("Phi", -np.where(phi > 180, phi - 360, phi)), - ) - if from_system == "SPCS_CCW": - if to_system == "SPCS_Ericsson": - Pattern_3D = Pattern_3D.assign_coords( - Theta=("Theta", theta + 90), - Phi=("Phi", np.where(phi >= 180, phi - 360, phi)), - ) - if from_system == "SPCS_Geo": - if to_system == "SPCS_Ericsson": - Pattern_3D = Pattern_3D.assign_coords( - Theta=("Theta", np.flip(theta)), - Phi=("Phi", -np.where(phi > 180, phi - 360, phi)), - ) + # to_system is guaranteed SPCS_Ericsson by the guard above. + new_theta, new_phi = _TO_ERICSSON[from_system](theta, phi) + Pattern_3D = Pattern_3D.assign_coords( + Theta=("Theta", new_theta), + Phi=("Phi", new_phi), + ) + new_theta = Pattern_3D.coords["Theta"].values + if new_theta.min() < 0 or new_theta.max() > 180: + logger.error( + f"AntennaPattern: Transformed theta out of range [0, 180] ([{new_theta.min()}, {new_theta.max()}]) converting from '{from_system}'. Input data is likely out of spec for that system." + ) + raise ValueError( + f"AntennaPattern: Transformed theta out of range [0, 180] ([{new_theta.min()}, {new_theta.max()}]) converting from '{from_system}'. Input data is likely out of spec for that system." + ) + new_phi = Pattern_3D.coords["Phi"].values + if new_phi.min() < -180 or new_phi.max() > 179: + logger.error( + f"AntennaPattern: Transformed phi out of range [-180, 179] ([{new_phi.min()}, {new_phi.max()}]) converting from '{from_system}'. Input data is likely out of spec for that system." + ) + raise ValueError( + f"AntennaPattern: Transformed phi out of range [-180, 179] ([{new_phi.min()}, {new_phi.max()}]) converting from '{from_system}'. Input data is likely out of spec for that system." + ) Pattern_3D = Pattern_3D.assign_attrs( coordinate_system=to_system, ) @@ -700,6 +727,13 @@ def calculate_beam_efficiency( weighted_field_values = self.Pattern_3D["dOmega"] * field_values Sp_overall = float(weighted_field_values.sum()) + if Sp_overall == 0: + logger.error( + "AntennaPattern: Overall power is zero; cannot compute beam efficiency." + ) + raise ValueError( + "AntennaPattern: Overall power is zero; cannot compute beam efficiency." + ) operators_dict = { "<": operator.lt, @@ -814,19 +848,37 @@ def calculate_top_3db_point(self, power: bool = False) -> float: vertical_cut_normed = vertical_cut["P_tp_dB"] else: vertical_cut_normed = vertical_cut["P_co_dB"] + # Fallback: if no point at/below -3 dB exists above the peak (e.g. a very + # narrow beam peaking at the top of the cut), the 3 dB border collapses to + # the peak theta itself instead of leaving ``top_border`` unbound. + top_border = float(theta_val_peak) + # Advance the border by the actual theta grid step rather than a hardcoded + # 1 deg, so the result is correct for any sampling resolution. + theta_axis = np.sort(vertical_cut_normed["Theta"].values) + if theta_axis.size > 1: + grid_step = float(np.median(np.diff(theta_axis))) + else: + grid_step = 1.0 for theta_val in np.flip( vertical_cut_normed.sel(Theta=slice(0, theta_val_peak))["Theta"] ): if vertical_cut_normed.sel(Theta=theta_val) <= -3: - top_border = float((theta_val + 1).values) + top_border = float(theta_val.values) + grid_step break + else: + logger.warning( + "AntennaPattern: No -3 dB crossing found above the peak; using peak theta as top 3 dB border." + ) # enrich with top_3db_point self.Pattern_3D.attrs["top_3db_point"] = top_border return top_border def plot( - self, component_name: str = "P_tp_dB", show_fig: bool = True, remove_layout_components: bool = False + self, + component_name: str = "P_tp_dB", + show_fig: bool = True, + remove_layout_components: bool = False, ) -> None | go.Figure: """Plots the radiation pattern as heatmap. @@ -860,13 +912,8 @@ def plot( zmin=-30, zmax=0, colorbar={"title": component_name, "thickness": 9}, - hovertemplate=( - "φ = %{x:.0f}°
" - "θ = %{y:.0f}°
" - "val = %{z:.2f}
" - "" - ), - showscale=not(remove_layout_components), + hovertemplate="φ = %{x:.0f}°
θ = %{y:.0f}°
val = %{z:.2f}
", + showscale=not (remove_layout_components), ) ) fig.update_yaxes(autorange="reversed") @@ -884,7 +931,7 @@ def plot( "zeroline": False, "showline": False, }, - plot_bgcolor="rgba(0,0,0,0)", # sin fondo gris en el área del heatmap + plot_bgcolor="rgba(0,0,0,0)", # sin fondo gris en el área del heatmap paper_bgcolor="rgba(0,0,0,0)", # sin fondo/gris alrededor margin={"t": 0, "l": 0, "r": 0, "b": 0}, # recorta al mínimo height=500, @@ -1077,11 +1124,11 @@ def __str__(self) -> str: "==== Parameters ====", f" Gain [dbi]: {self.gain_dbi if self.gain_dbi is not None else 'N/A'}", f" EIRP [dBm]: {self.eirp_dbm if self.eirp_dbm is not None else 'N/A'}", - f" Phi HPBW [deg]: {self.phi_hpbw if self.phi_hpbw else 'N/A'}", - f" Theta HPBW [deg]: {self.theta_hpbw if self.theta_hpbw else 'N/A'}", - f" Front to Back [db]: {self.front_to_back if self.front_to_back else 'N/A'}", + f" Phi HPBW [deg]: {self.phi_hpbw if self.phi_hpbw is not None else 'N/A'}", + f" Theta HPBW [deg]: {self.theta_hpbw if self.theta_hpbw is not None else 'N/A'}", + f" Front to Back [db]: {self.front_to_back if self.front_to_back is not None else 'N/A'}", "==== Frequency & Tilt ====", - f" Frequency [Hz]: {self.frequency_hz if self.frequency_hz else 'N/A'}", + f" Frequency [Hz]: {self.frequency_hz if self.frequency_hz is not None else 'N/A'}", f" Frequency Range [Hz]: {self.frequency_range if self.frequency_range is not None else 'N/A'}", f" Theta Electrical Tilt [deg]: {self.theta_eletrical_tilt if self.theta_eletrical_tilt is not None else 'N/A'}", f" Phi Electrical Pan [deg]: {self.phi_eletrical_pan if self.phi_eletrical_pan is not None else 'N/A'}", @@ -1093,7 +1140,4 @@ def __str__(self) -> str: return "\n".join(lines) def __repr__(self) -> str: - return ( - f"" - ) + return f"" diff --git a/src/eas_3d_pattern/sample_data/__init__.py b/src/eas_3d_pattern/sample_data/__init__.py index fc4fd58..ff454e2 100644 --- a/src/eas_3d_pattern/sample_data/__init__.py +++ b/src/eas_3d_pattern/sample_data/__init__.py @@ -7,32 +7,28 @@ SAMPLE_JSON: list[Path] = [] try: - resource_names = list(importlib.resources.contents(__package__)) - resource_names.sort() + anchor = importlib.resources.files(__package__) + resources = sorted(anchor.iterdir(), key=lambda res: res.name) - for item_name in resource_names: - if item_name.endswith(".json") and importlib.resources.is_resource( - __package__, item_name - ): + for resource in resources: + if resource.name.endswith(".json") and resource.is_file(): try: - with importlib.resources.path(__package__, item_name) as path_context: + with importlib.resources.as_file(resource) as path_context: resolved_path = Path(path_context) if resolved_path.is_file(): SAMPLE_JSON.append(resolved_path) else: logger.warning( - f"Path for sample '{item_name}' in '{__package__}' " - f" ('{resolved_path}') was not a file after context. Skipping." + f"Path for sample '{resource.name}' in '{__package__}' ('{resolved_path}') was not a file after context. Skipping." ) except FileNotFoundError: logger.warning( - f"Sample file '{item_name}' listed but not found by " - f"importlib.resources.path in '{__package__}'. Skipping." + f"Sample file '{resource.name}' listed but not found by importlib.resources in '{__package__}'. Skipping." ) except Exception as e_path: logger.error( - f"Error resolving path for sample '{item_name}' in '{__package__}': {e_path}" + f"Error resolving path for sample '{resource.name}' in '{__package__}': {e_path}" ) if SAMPLE_JSON: @@ -41,9 +37,7 @@ ) else: logger.debug( - f"No .json sample files found or resolved in '{__package__}'. " - "Ensure files exist, have .json extension, and 'include' " - "in pyproject.toml is correct for 'sample_data/*.json'." + f"No .json sample files found or resolved in '{__package__}'. Ensure files exist, have .json extension, and 'include' in pyproject.toml is correct for 'sample_data/*.json'." ) except ModuleNotFoundError: diff --git a/src/eas_3d_pattern/schema_manager.py b/src/eas_3d_pattern/schema_manager.py index de124bf..e0cfc95 100644 --- a/src/eas_3d_pattern/schema_manager.py +++ b/src/eas_3d_pattern/schema_manager.py @@ -160,9 +160,11 @@ def _load_bundled(self) -> dict[str, Any]: f"SchemaManager: Attempting to load bundled schema: {self.bundled_package_ref}/{self.bundled_filename}" ) try: - with importlib.resources.open_text( - self.bundled_package_ref, self.bundled_filename - ) as sf: + schema_resource = ( + importlib.resources.files(self.bundled_package_ref) + / self.bundled_filename + ) + with schema_resource.open(encoding="utf-8") as sf: content = json.load(sf) self.source_message = ( f"Bundled Schema ({self.bundled_package_ref}/{self.bundled_filename})" @@ -205,11 +207,7 @@ def _load_and_validate_schema(self): ) from e def __repr__(self) -> str: - return ( - f"<{self.__class__.__name__}(id=0x{id(self):x}, " - f"schema_loaded={'True' if self.schema_content else 'False'}, " - f"source={self.source_message})>" - ) + return f"<{self.__class__.__name__}(id=0x{id(self):x}, schema_loaded={'True' if self.schema_content else 'False'}, source={self.source_message})>" def __str__(self) -> str: return str(self.schema_content) diff --git a/src/eas_3d_pattern/sector_definitions.py b/src/eas_3d_pattern/sector_definitions.py index fea2acc..ceddf75 100644 --- a/src/eas_3d_pattern/sector_definitions.py +++ b/src/eas_3d_pattern/sector_definitions.py @@ -59,11 +59,7 @@ def __post_init__(self): ) def __str__(self): - return ( - f"'{self.name}': \t" - f"[{self.theta_min[0]:.1f}{self.theta_min[1]}Theta{self.theta_max[1]}{self.theta_max[0]:.1f}], " - f"[{self.phi_min[0]:.1f}{self.phi_min[1]}Phi{self.phi_max[1]}{self.phi_max[0]:.1f}]" - ) + return f"'{self.name}': \t[{self.theta_min[0]:.1f}{self.theta_min[1]}Theta{self.theta_max[1]}{self.theta_max[0]:.1f}], [{self.phi_min[0]:.1f}{self.phi_min[1]}Phi{self.phi_max[1]}{self.phi_max[0]:.1f}]" class SectorDefinition: diff --git a/src/eas_3d_pattern/util_func/report.py b/src/eas_3d_pattern/util_func/report.py index 86379f0..402d3d6 100644 --- a/src/eas_3d_pattern/util_func/report.py +++ b/src/eas_3d_pattern/util_func/report.py @@ -6,7 +6,10 @@ import pandas as pd from tqdm import tqdm -from eas_3d_pattern import AntennaPattern, SectorDefinition +from ..parser import AntennaPattern +from ..sector_definitions import SectorDefinition + +logger = logging.getLogger(__name__) SUBBANDS_DEFAULT = { "698-806": (698, 806), @@ -80,6 +83,7 @@ def generate_report_eas( output_directory.mkdir(parents=True, exist_ok=True) df_list: list[pd.DataFrame] = [] + skipped_files: list[str] = [] with temporarily_set_loglevel(logger_name="eas_3d_pattern", level=logging.ERROR): for file in tqdm(files): data = _process_a_file(file) @@ -87,8 +91,24 @@ def generate_report_eas( data_row, pattern, sectors = data df_list.append(data_row) if plot: - _save_figure(pattern, sectors, output_directory, remove_layout_components) - df_raw = pd.concat(df_list, ignore_index=True) + _save_figure( + pattern, sectors, output_directory, remove_layout_components + ) + else: + skipped_files.append(file.name) + + if skipped_files: + logger.warning( + f"Report: skipped {len(skipped_files)} of {len(files)} file(s) due to processing errors: {skipped_files}" + ) + if not df_list: + logger.error( + f"Report: all {len(files)} file(s) were skipped due to errors; no report generated." + ) + raise ValueError( + f"Report: all {len(files)} file(s) were skipped due to errors; no report generated." + ) + df_raw = pd.concat(df_list, ignore_index=True) report_name = output_directory / "BEreport.xlsx" _generate_excel_report(df_raw, report_name, subbands) @@ -118,7 +138,7 @@ def _process_a_file( top_border = pattern.calculate_top_3db_point(power=False) eas_sectors = SectorDefinition(load_default=True, top_border=top_border) if (data["Phi_HPBW"] <= 50) & (pattern.Pattern_3D.peak_coordinates[1] < -20): - logging.info( + logger.info( "Reporting: Identified dual beam antenna. Overwriting sectors to dual beam definition for reporting." ) eas_sectors.add_sector( @@ -136,7 +156,7 @@ def _process_a_file( phi_max=(180.0, "<="), ) if (data["Phi_HPBW"] <= 50) & (pattern.Pattern_3D.peak_coordinates[1] > 20): - logging.info( + logger.info( "Reporting: Identified dual beam antenna. Changing sectors to dual beam definition for reporting." ) eas_sectors.add_sector( @@ -169,7 +189,7 @@ def _process_a_file( data["filepath"] = str(pattern.data_filepath) return (pd.json_normalize(data, sep="_"), pattern, eas_sectors) except Exception as e: - logging.error( + logger.error( f"Reporting: Skipping corrupted or invalid file '{file_path.name}': {e}" ) return None @@ -192,7 +212,9 @@ def _save_figure( Returns: None """ - fig = pattern.plot(show_fig=False,remove_layout_components=remove_layout_components) + fig = pattern.plot( + show_fig=False, remove_layout_components=remove_layout_components + ) if fig is None: return for k in sector_definitions.sectors: @@ -262,7 +284,13 @@ def _generate_excel_report( ) pivot_df["Average"] = pivot_df.mean(axis=1) avg_df_list.append(pivot_df.reset_index()) - df_per_arrayandsubband_per_tilt = pd.concat(avg_df_list, ignore_index=True) + if avg_df_list: + df_per_arrayandsubband_per_tilt = pd.concat(avg_df_list, ignore_index=True) + else: + logger.warning( + "Report: No antenna frequency fell within any configured subband; the per-subband sheet will be empty." + ) + df_per_arrayandsubband_per_tilt = pd.DataFrame() with pd.ExcelWriter(report_name) as writer: df.to_excel(writer, index=False, sheet_name="Raw_Data") @@ -273,5 +301,5 @@ def _generate_excel_report( df_per_arrayandsubband_per_tilt.to_excel( writer, index=False, sheet_name="Mean_ArrayID_Subband_Tilt" ) - logging.info("Report: ✨Generated EAS BE report in %s✨", report_name) + logger.info("Report: ✨Generated EAS BE report in %s✨", report_name) return None diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..deaeffd --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,104 @@ +"""Shared pytest fixtures for eas-3d-pattern tests. + +Provides a synthetic NGMN-BASTA-like antenna pattern factory so tests can run +fast and offline without depending on the multi-megabyte bundled sample data. +The factory writes a minimal valid 3drp JSON file to a temp path and returns +that path, which is what ``AntennaPattern`` consumes. +""" + +from __future__ import annotations + +import json +from collections.abc import Callable +from pathlib import Path + +import numpy as np +import pytest + +# Default uniform grid: theta 0..180, phi -180..179 (SPCS_Ericsson native range). +DEFAULT_THETA_SAMPLING = [0.0, 5.0, 180.0] +DEFAULT_PHI_SAMPLING = [-180.0, 5.0, 175.0] + + +def _grid(sampling: list[float]) -> np.ndarray: + """Build a 1D grid from a [start, step, stop] NGMN sampling triple.""" + start, step, stop = sampling + return np.arange(start, stop + 1e-6, step) + + +def build_pattern_dict( + coordinate_system: str = "SPCS_Ericsson", + theta_sampling: list[float] | None = None, + phi_sampling: list[float] | None = None, + peak_theta: float = 90.0, + peak_phi: float = 0.0, + theta_rolloff: float = 0.05, + phi_rolloff: float = 0.01, + max_attenuation: float = 40.0, + data_set: list[list[float]] | None = None, + row_structure: list[str] | None = None, + extra: dict | None = None, +) -> dict: + """Build a minimal valid NGMN 3drp pattern dictionary. + + The synthetic main beam is a separable quadratic roll-off in theta and phi + centered on ``(peak_theta, peak_phi)``, clipped at ``max_attenuation`` dB. + """ + theta_sampling = theta_sampling or list(DEFAULT_THETA_SAMPLING) + phi_sampling = phi_sampling or list(DEFAULT_PHI_SAMPLING) + thetas = _grid(theta_sampling) + phis = _grid(phi_sampling) + + if data_set is None: + rows: list[list[float]] = [] + for th in thetas: + for ph in phis: + atten = ( + theta_rolloff * (th - peak_theta) ** 2 + + phi_rolloff * (ph - peak_phi) ** 2 + ) + atten = float(min(atten, max_attenuation)) + # MagAttenuationTP, MagAttenuationCo, MagAttenuationCr + rows.append([atten, atten, atten + 20.0]) + data_set = rows + row_structure = [ + "MagAttenuationTP", + "MagAttenuationCo", + "MagAttenuationCr", + ] + + pattern: dict = { + "Coordinate_System": coordinate_system, + "Gain": {"value": 15.0, "unit": "dBi"}, + "Phi_HPBW": 65.0, + "Theta_HPBW": 7.0, + "Front_to_Back": 30.0, + "Theta_Sampling": theta_sampling, + "Phi_Sampling": phi_sampling, + "Data_Set_Row_Structure": row_structure, + "Data_Set": data_set, + } + if extra: + pattern.update(extra) + return pattern + + +@pytest.fixture +def make_pattern_dict(): + """Return the synthetic pattern-dict builder for tests that write their own files.""" + return build_pattern_dict + + +@pytest.fixture +def pattern_path(tmp_path: Path) -> Callable[..., str]: + """Return a factory that writes a synthetic pattern JSON and returns its path.""" + counter = {"n": 0} + + def _make(**kwargs) -> str: + pattern = build_pattern_dict(**kwargs) + counter["n"] += 1 + file_path = tmp_path / f"pattern_{counter['n']}.json" + file_path.write_text(json.dumps(pattern), encoding="utf-8") + return str(file_path) + + return _make diff --git a/tests/test_coordinates.py b/tests/test_coordinates.py new file mode 100644 index 0000000..9fbaf31 --- /dev/null +++ b/tests/test_coordinates.py @@ -0,0 +1,108 @@ +"""Regression tests for coordinate-system transforms (Bugs 41-44). + +Empirical note: spec-valid SPCS_CW data uses theta in [-90, 90] and phi in +[0, 359] (verified against the bundled ANTMODEL2 sample). ``theta + 90`` then +correctly maps to [0, 180]. The real defect is the absence of a post-condition +check: out-of-spec input silently produces theta/phi outside the internal +SPCS_Ericsson ranges, yielding negative solid-angle weights and corrupted +calculations. +""" + +from __future__ import annotations + +import numpy as np +import pytest + +from eas_3d_pattern import AntennaPattern + + +def test_cw_out_of_spec_theta_is_rejected(pattern_path): + """Bug 41: CW theta outside spec must not silently yield negative dOmega. + + Feeding CW data with theta in [0, 180] (out of CW's [-90, 90] spec) makes + ``theta + 90`` land in [90, 270]; ``sin`` then goes negative, corrupting + directivity/beam-efficiency. The transform must reject this instead. + Phi here is kept in valid CW range [0, 355] so only theta is out of range. + """ + path = pattern_path( + coordinate_system="SPCS_CW", + theta_sampling=[0.0, 5.0, 180.0], + phi_sampling=[0.0, 5.0, 355.0], + peak_theta=90.0, + peak_phi=0.0, + ) + with pytest.raises(ValueError, match="(?i)theta"): + AntennaPattern(path, validate=False) + + +def test_valid_cw_pattern_constructs_with_internal_ranges(pattern_path): + """Spec-valid CW data (theta in [-90, 90]) must still convert successfully. + + Guards against the post-condition over-rejecting real CW data: after the + transform theta must be within [0, 180]. + """ + path = pattern_path( + coordinate_system="SPCS_CW", + theta_sampling=[-90.0, 5.0, 90.0], + phi_sampling=[0.0, 5.0, 355.0], + peak_theta=0.0, + peak_phi=0.0, + ) + pattern = AntennaPattern(path, validate=False) + theta = pattern.Pattern_3D.coords["Theta"].values + assert theta.min() >= 0.0 + assert theta.max() <= 180.0 + + +@pytest.mark.parametrize( + ("system", "theta_sampling"), + [ + ("SPCS_Polar", [0.0, 5.0, 180.0]), + ("SPCS_CW", [-90.0, 5.0, 90.0]), + ("SPCS_CCW", [-90.0, 5.0, 90.0]), + ("SPCS_Geo", [0.0, 5.0, 180.0]), + ], +) +def test_phi_boundary_consistent_across_transforms( + pattern_path, system, theta_sampling +): + """Bug 42: the phi=180 column maps consistently to -180 in every system. + + The ``phi >= 180`` (Polar/CCW) vs ``phi > 180`` (CW/Geo) difference is not a + bug: CW/Geo negate the wrapped value, so ``>`` is the correct compensation. + For spec-valid input all four systems map the phi=180 input column to -180 + and keep phi within the internal [-180, 179] range (no +180, no dropped + column). This regression test pins that correct behavior. + """ + path = pattern_path( + coordinate_system=system, + theta_sampling=theta_sampling, + phi_sampling=[0.0, 5.0, 355.0], + peak_theta=0.0 if system in ("SPCS_CW", "SPCS_CCW") else 90.0, + peak_phi=0.0, + ) + pattern = AntennaPattern(path, validate=False) + phi = pattern.Pattern_3D.coords["Phi"].values + assert phi.min() >= -180.0 + assert phi.max() <= 179.0 + assert not np.any(np.isclose(phi, 180.0)) + assert np.any(np.isclose(phi, -180.0)) + + +def test_transformed_phi_out_of_range_is_rejected(pattern_path): + """Bug 44: the post-condition must also reject phi outside [-180, 179]. + + A CW pattern with valid theta ([-90, 85] -> [0, 175]) but phi already in the + internal range [-180, 175] gets negated by the CW transform to [-175, 180], + pushing the phi=-180 column to +180 (out of the internal range). The + post-condition must catch this instead of silently keeping +180. + """ + path = pattern_path( + coordinate_system="SPCS_CW", + theta_sampling=[-90.0, 5.0, 85.0], + phi_sampling=[-180.0, 5.0, 175.0], + peak_theta=0.0, + peak_phi=0.0, + ) + with pytest.raises(ValueError, match="(?i)phi"): + AntennaPattern(path, validate=False) diff --git a/tests/test_normalize.py b/tests/test_normalize.py new file mode 100644 index 0000000..c394e79 --- /dev/null +++ b/tests/test_normalize.py @@ -0,0 +1,52 @@ +"""Tests for issue #5: vendor-specific JSON key normalization (_normalize_json). + +Some 3drp files use 'Theta_Tilt' (NGMN whitepaper naming) instead of the +schema-canonical 'Theta_Electrical_Tilt', which broke report generation. The +parser normalizes known variants on load via the module-level ALTERNATIVES map. +""" + +from __future__ import annotations + +from eas_3d_pattern import AntennaPattern + + +def _normalize(data: dict) -> dict: + """Call ``_normalize_json`` in isolation. + + The method does not use ``self``, so it can be invoked unbound for a fast, + construction-free unit test of the mapping logic. + """ + return AntennaPattern._normalize_json(None, data) # type: ignore[arg-type] + + +def test_canonical_key_passes_through_unchanged(): + """A file already using the canonical key is left untouched.""" + data = {"Theta_Electrical_Tilt": 6.0, "Gain": 1} + assert _normalize(dict(data)) == data + + +def test_variant_key_is_renamed_to_canonical(): + """'Theta_Tilt' is renamed to 'Theta_Electrical_Tilt' with its value kept.""" + out = _normalize({"Theta_Tilt": 6.0}) + assert out == {"Theta_Electrical_Tilt": 6.0} + assert "Theta_Tilt" not in out + + +def test_existing_canonical_is_not_overwritten_when_both_present(): + """When both keys coexist, the canonical value must not be overwritten.""" + out = _normalize({"Theta_Tilt": 1.0, "Theta_Electrical_Tilt": 2.0}) + assert out["Theta_Electrical_Tilt"] == 2.0 + + +def test_no_variant_leaves_data_untouched(): + """Data without any known variant is returned unchanged (no spurious keys).""" + data = {"Gain": 1, "Phi_HPBW": 65.0} + assert _normalize(dict(data)) == data + + +def test_construction_normalizes_theta_tilt(pattern_path): + """End-to-end: a file using 'Theta_Tilt' is normalized during init (issue #5).""" + path = pattern_path(extra={"Theta_Tilt": 6.0}) + pattern = AntennaPattern(path, validate=False) + assert pattern.raw_data["Theta_Electrical_Tilt"] == 6.0 + assert "Theta_Tilt" not in pattern.raw_data diff --git a/tests/test_parser.py b/tests/test_parser.py new file mode 100644 index 0000000..d9f8571 --- /dev/null +++ b/tests/test_parser.py @@ -0,0 +1,145 @@ +"""Regression tests for AntennaPattern parser bugs (v0.2.0 bug table).""" + +from __future__ import annotations + +import numpy as np +import pytest + +from eas_3d_pattern import AntennaPattern, SectorDefinition + + +def test_top_3db_point_no_crossing_does_not_raise(pattern_path): + """Bug 1: a beam peaking at the top of the cut must not raise UnboundLocalError. + + When the peak sits at theta=0, the upward vertical cut contains no point at + or below -3 dB, so the search loop never assigns ``top_border``. The method + must return a finite float fallback instead of crashing. + """ + path = pattern_path(peak_theta=0.0, peak_phi=0.0) + pattern = AntennaPattern(path, validate=False) + top = pattern.calculate_top_3db_point(power=False) + assert isinstance(top, float) + assert np.isfinite(top) + + +def test_unknown_coordinate_system_raises(pattern_path): + """Bug 2: an unrecognized coordinate system must not silently no-op. + + Previously ``_change_coordinate_system`` matched no branch for an unknown + system yet still stamped the attrs as converted. It must raise instead. + """ + path = pattern_path(coordinate_system="SPCS_Unknown") + with pytest.raises(ValueError, match="coordinate system"): + AntennaPattern(path, validate=False) + + +def test_substring_coordinate_system_not_treated_as_internal(pattern_path): + """Bug 3: substring check must be equality. + + ``"SPCS_Eri"`` is a substring of ``"SPCS_Ericsson"``. The old + ``not in DEFAULT_INTERNAL_COORD_SYSTEM`` check treated it as already-internal + and skipped conversion. With ``!=`` it is correctly recognized as a distinct + (here unsupported) system and rejected. + """ + path = pattern_path(coordinate_system="SPCS_Eri") + with pytest.raises(ValueError, match="coordinate system"): + AntennaPattern(path, validate=False) + + +def test_beam_efficiency_zero_overall_power_raises(pattern_path): + """Bug 5: zero overall power must not silently produce inf/nan efficiencies. + + A pattern whose linear power underflows to exactly 0 everywhere makes + ``Sp_overall == 0``. The division ``Sp_region / Sp_overall`` then yields + nan/inf silently. The method must instead raise a clear error. + """ + thetas = np.arange(0.0, 180.0 + 1e-6, 5.0) + phis = np.arange(-180.0, 175.0 + 1e-6, 5.0) + n = len(thetas) * len(phis) + # 4000 dB attenuation -> 10**(-400) underflows to 0.0 for all components. + data_set = [[4000.0, 4000.0, 4000.0] for _ in range(n)] + path = pattern_path( + data_set=data_set, + row_structure=["MagAttenuationTP", "MagAttenuationCo", "MagAttenuationCr"], + ) + pattern = AntennaPattern(path, validate=False) + sectors = SectorDefinition(load_default=False) + sectors.add_sector( + name="all", + theta_min=(0.0, "<="), + theta_max=(180.0, "<="), + phi_min=(-180.0, "<="), + phi_max=(180.0, "<="), + ) + with pytest.raises(ValueError, match="(?i)overall power"): + pattern.calculate_beam_efficiency(sector_definitions=sectors) + + +def test_empty_data_set_raises_clear_error(pattern_path): + """Bug 6: an empty Data_Set must raise a clear error early in __init__. + + Previously an empty Data_Set surfaced only as a confusing downstream error + (sampling-count mismatch) or silently produced an empty dataset. + """ + path = pattern_path( + data_set=[], + row_structure=["MagAttenuationTP", "MagAttenuationCo", "MagAttenuationCr"], + ) + with pytest.raises(ValueError, match="(?i)data_set"): + AntennaPattern(path, validate=False) + + +def test_top_3db_point_uses_grid_step_not_hardcoded_one(pattern_path): + """Bug 43: the 3 dB border must advance by the grid step, not a fixed +1. + + On a 5 deg grid with the beam peaking at theta=90, the -3 dB crossing falls + on theta=80 (atten 5 dB), so the top border should be 80 + 5 = 85, not the + hardcoded 80 + 1 = 81. + """ + path = pattern_path(peak_theta=90.0, peak_phi=0.0) # default 5 deg grid + pattern = AntennaPattern(path, validate=False) + top = pattern.calculate_top_3db_point(power=False) + assert top == 85.0 + + +def test_validate_true_without_schema_raises(pattern_path, monkeypatch): + """Bug 46: requesting validation with no schema must not silently skip it. + + If the schema failed to load, ``validate=True`` previously did nothing and + gave no signal. It must raise so the caller knows validation was not done. + """ + from eas_3d_pattern.schema_manager import NGMNSchema + + monkeypatch.setattr(NGMNSchema, "schema_content", None) + path = pattern_path() + with pytest.raises(ValueError, match="(?i)schema"): + AntennaPattern(path, validate=True) + + +def test_str_shows_zero_values_not_na(pattern_path): + """Bug 49: a legitimate 0.0 value must render as 0.0, not 'N/A'. + + ``__str__`` used truthiness checks (``x if x else 'N/A'``), so a real 0.0 + HPBW or front-to-back ratio was displayed as 'N/A'. Must use ``is not None``. + """ + path = pattern_path( + extra={ + "Supplier": "S", + "Antenna_Model": "M", + "Antenna_Type": "T", + "Revision_Version": "R", + "Released_Date": "2020", + "Pattern_Type": "P", + "Nominal_Polarization": "NP", + "Optional_Comments": "none", + "Phi_HPBW": 0.0, + "Theta_HPBW": 0.0, + "Front_to_Back": 0.0, + "Frequency": {"value": 2100.0, "unit": "MHz"}, + } + ) + pattern = AntennaPattern(path, validate=False) + text = str(pattern) + assert "Phi HPBW [deg]: 0.0" in text + assert "Theta HPBW [deg]: 0.0" in text + assert "Front to Back [db]: 0.0" in text diff --git a/tests/test_report.py b/tests/test_report.py new file mode 100644 index 0000000..4de8b18 --- /dev/null +++ b/tests/test_report.py @@ -0,0 +1,125 @@ +"""Regression tests for util_func.report bugs (v0.2.0 bug table).""" + +from __future__ import annotations + +import ast +from pathlib import Path + +import eas_3d_pattern.util_func.report as report_mod + + +def test_report_does_not_import_from_top_level_package(): + """Bug 4: report.py must not import from the top-level package. + + ``from eas_3d_pattern import AntennaPattern, SectorDefinition`` creates a + circular import that only works because of import ordering in ``__init__``. + The module must use relative imports of the concrete submodules instead. + Inspect real import statements via AST so docstring examples are ignored. + """ + source = Path(report_mod.__file__).read_text(encoding="utf-8") + tree = ast.parse(source) + offending = [ + node + for node in ast.walk(tree) + if isinstance(node, ast.ImportFrom) + and node.level == 0 + and node.module == "eas_3d_pattern" + ] + assert not offending, "report.py must not absolute-import the top-level package" + assert report_mod.AntennaPattern is not None + assert report_mod.SectorDefinition is not None + + +def test_report_uses_module_logger(): + """Bug 8: report.py must log via a module logger, not the root logger. + + Direct ``logging.info``/``logging.error`` calls bypass the user's logging + configuration. The module must define ``logger = logging.getLogger(__name__)`` + and route all log calls through it. + """ + assert hasattr(report_mod, "logger") + assert report_mod.logger.name == "eas_3d_pattern.util_func.report" + + source = Path(report_mod.__file__).read_text(encoding="utf-8") + tree = ast.parse(source) + log_methods = { + "debug", + "info", + "warning", + "error", + "critical", + "exception", + "log", + } + root_logger_calls = [ + node.func.attr + for node in ast.walk(tree) + if isinstance(node, ast.Call) + and isinstance(node.func, ast.Attribute) + and node.func.attr in log_methods + and isinstance(node.func.value, ast.Name) + and node.func.value.id == "logging" + ] + assert root_logger_calls == [], root_logger_calls + + +def test_generate_excel_report_handles_no_matching_subband(tmp_path): + """Bug 47: empty avg_df_list must not crash pd.concat. + + When no antenna frequency falls in any configured subband, ``avg_df_list`` + is empty and ``pd.concat([])`` raised ``ValueError: No objects to + concatenate``. The report must still be generated. + """ + import pandas as pd + + df = pd.DataFrame( + { + "Supplier": ["S"], + "Antenna_Model": ["M"], + "Revision_Version": ["R"], + "Array_ID": ["A"], + "Cell": [50.0], + "Theta_Electrical_Tilt": [0.0], + "Frequency_value": [100.0], # below every SUBBANDS_DEFAULT range + } + ) + report_name = tmp_path / "BEreport.xlsx" + report_mod._generate_excel_report(df, report_name, report_mod.SUBBANDS_DEFAULT) + assert report_name.is_file() + + +def test_generate_report_surfaces_skipped_files(tmp_path, make_pattern_dict, caplog): + """Bug 48: files that fail processing must be surfaced, not silently dropped. + + One valid pattern and one invalid file are placed in the input directory. + The report must be generated from the valid file while logging a warning + that reports the number of skipped files. + """ + import json + import logging + + input_dir = tmp_path / "in" + input_dir.mkdir() + output_dir = tmp_path / "out" + + good = make_pattern_dict( + coordinate_system="SPCS_Ericsson", + peak_theta=90.0, + peak_phi=0.0, + extra={ + "Supplier": "S", + "Antenna_Model": "M", + "Revision_Version": "R", + "Array_ID": "A", + "Theta_Electrical_Tilt": 0.0, + "Frequency": {"value": 2100.0, "unit": "MHz"}, + }, + ) + (input_dir / "good.json").write_text(json.dumps(good), encoding="utf-8") + (input_dir / "bad.json").write_text("{}", encoding="utf-8") + + with caplog.at_level(logging.WARNING, logger="eas_3d_pattern.util_func.report"): + df = report_mod.generate_report_eas(input_dir, output_dir) + + assert len(df) == 1 + assert "skipped" in caplog.text.lower() diff --git a/tests/test_resources.py b/tests/test_resources.py new file mode 100644 index 0000000..c926b94 --- /dev/null +++ b/tests/test_resources.py @@ -0,0 +1,51 @@ +"""Regression tests for deprecated importlib.resources usage (Bug 7).""" + +from __future__ import annotations + +import ast +import importlib +import warnings +from pathlib import Path + +import eas_3d_pattern.sample_data as sample_data +from eas_3d_pattern import schema_manager + +_DEPRECATED = { + "importlib.resources.contents", + "importlib.resources.is_resource", + "importlib.resources.path", + "importlib.resources.open_text", +} + + +def _deprecated_resource_calls(source: str) -> list[str]: + tree = ast.parse(source) + found: list[str] = [] + for node in ast.walk(tree): + if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute): + dotted = ast.unparse(node.func) + if dotted in _DEPRECATED: + found.append(dotted) + return found + + +def test_no_deprecated_importlib_resources_apis(): + """Bug 7: neither sample_data nor schema_manager may call deprecated APIs. + + ``contents``/``is_resource``/``path``/``open_text`` are deprecated and + removed in Python 3.14. They must be replaced by the ``files()`` API. + """ + for module in (sample_data, schema_manager): + source = Path(module.__file__).read_text(encoding="utf-8") + assert _deprecated_resource_calls(source) == [], module.__name__ + + +def test_sample_data_no_deprecated_warning_and_finds_samples(): + """Bug 7: reloading sample_data emits no DeprecationWarning and finds samples.""" + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + importlib.reload(sample_data) + + deprecations = [w for w in caught if issubclass(w.category, DeprecationWarning)] + assert not deprecations, [str(w.message) for w in deprecations] + assert len(sample_data.SAMPLE_JSON) >= 1 diff --git a/tests/test_smoke.py b/tests/test_smoke.py new file mode 100644 index 0000000..703a9a5 --- /dev/null +++ b/tests/test_smoke.py @@ -0,0 +1,13 @@ +"""Smoke tests verifying the test harness and synthetic pattern factory work.""" + +from __future__ import annotations + +from eas_3d_pattern import AntennaPattern + + +def test_import_and_construct(pattern_path): + """A synthetic pattern can be loaded into AntennaPattern offline.""" + path = pattern_path() + pattern = AntennaPattern(path, validate=False) + assert pattern.Pattern_3D is not None + assert "P_tp_lin" in pattern.Pattern_3D.data_vars