Source code for pyiecwind.generation

"""Wind condition generators and .wnd writing."""

from __future__ import annotations

import math
import os
import re
import tempfile
from dataclasses import dataclass
from pathlib import Path

import numpy as np

from .models import BETA, DT, EWM_ALPHA, PI, VCG, VERSION, IECParameters
from .parsing import parse_input_file

__all__ = [
    "GenerationError",
    "GenerationResult",
    "gen_ecd",
    "gen_edc",
    "gen_eog",
    "gen_ews",
    "gen_nwp",
    "gen_ewm",
    "generate_all",
    "generate_from_input_file",
]


[docs] @dataclass(frozen=True) class GenerationError: """A single condition that could not be generated.""" code: str message: str def __str__(self) -> str: return f"{self.code}: {self.message}"
[docs] @dataclass(frozen=True) class GenerationResult: """Outcome of a batch generation run. ``generated`` holds the paths written successfully and ``errors`` holds the conditions that were skipped (only populated when ``strict=False``). """ generated: tuple[Path, ...] = () errors: tuple[GenerationError, ...] = () @property def count(self) -> int: """Number of files written.""" return len(self.generated) @property def ok(self) -> bool: """True when every requested condition was generated.""" return not self.errors def __len__(self) -> int: return len(self.generated)
class WindFileWriter: """Build and save an InflowWind-compatible .wnd file.""" TAB = "\t" def __init__(self, filepath: Path, params: IECParameters) -> None: self.filepath = filepath self.params = params self._lines: list[str] = [] def _write(self, text: str = "") -> None: self._lines.append(text) def file_header(self) -> None: self._write(f"! This file was generated by pyIECWind v{VERSION}.") self._write("! Wind condition defined by IEC 61400-1.") def wind_header(self, condition_name: str) -> None: p = self.params lc = p.len_convert lu = p.len_unit self._write(f"! IEC Turbine Class {p.wtc}, Turbulence Category {p.catg.upper()}") self._write("! ----------------------------------------------------------") self._write(f"! Rotor diameter = {p.dia * lc:.1f} {lu}.") self._write(f"! {p.hh * lc:.1f} {lu} hub height yields {p.turb_scale:.1f} {lu} turbulence scale.") self._write("! ----------------------------------------------------------") self._write(f"! IEC Condition: {condition_name}.") self._write("! Stats for this wind file:") def time_header(self, duration: float) -> None: self._write(f"! The transient occurs over {duration:.1f} seconds.") def gust_header(self, v_gust_max: float, vhub: float, t_max: float) -> None: p = self.params lc = p.len_convert su = p.spd_unit self._write("!") self._write(f"! The maximum gust speed is {v_gust_max * lc:.3f} {su}.") self._write(f"! The maximum total wind speed is {(v_gust_max + vhub) * lc:.3f} {su}.") self._write(f"! This maximum occurs at {t_max:.3f} seconds.") def direction_header(self, max_dir_deg: float, t_max: float) -> None: self._write("!") self._write(f"! The extreme direction change is {max_dir_deg:.2f} degrees.") self._write(f"! This extreme occurs at {t_max:.2f} seconds.") def shear_header(self, dir_sign: float, is_horizontal: bool, shr_max: float, t_max: float) -> None: orient = "horizontal" if is_horizontal else "vertical" sign_word = "positive" if dir_sign > 0.0 else "negative" self._write("!") self._write(f"! The extreme linear {orient} wind shear is {dir_sign * shr_max:.3f}.") self._write(f"! This extreme occurs at {t_max:.2f} seconds.") self._write(f"! This sign of the shear is {sign_word}.") def slope_header(self) -> None: self._write("!") self._write( f"! The wind inflow inclination angle is {self.params.slope_deg:.1f} degrees to the horizontal." ) def nwp_header(self, vhub: float) -> None: p = self.params self._write(f"! The steady wind speed is {vhub * p.len_convert:.2f} {p.spd_unit}.") def ewm_header(self, vhub: float) -> None: p = self.params self._write(f"! The steady extreme wind speed is {vhub * p.len_convert:.2f} {p.spd_unit}.") def col_headers(self) -> None: t = self.TAB su = f"({self.params.spd_unit})" self._write("! ----------------------------------------------------------") self._write(f"! Time{t}Wind{t}Wind{t}Vertical{t}Horiz.{t}Pwr.Law{t}Lin.Vert.{t}Gust") self._write(f"!{t}Speed{t}Dir{t}Speed{t}Shear{t}Vert.Shr{t}Shear{t}Speed") self._write(f"!(sec){t}{su}{t}(deg){t}{su}{t}{t}{t}{t}{su}") def data_row( self, time: float, wind_spd: float, wind_dir: float, vert_spd: float, horiz_shear: float, alpha: float, lin_vert_shear: float, gust_spd: float, ) -> None: lc = self.params.len_convert values = [ time, wind_spd * lc, wind_dir, vert_spd * lc, horiz_shear, alpha, lin_vert_shear, gust_spd * lc, ] self._lines.append(self.TAB.join(f"{value:9.3f}" for value in values)) def save(self) -> Path: self.filepath.write_text("\n".join(self._lines) + "\n", encoding="utf-8") return self.filepath def _transient_times(t1: float, duration: float) -> np.ndarray: step_count = round(duration / DT) + 1 return t1 + np.arange(step_count) * DT def _hub_components(vhub: float, slope_rad: float) -> tuple[float, float]: return vhub * math.cos(slope_rad), vhub * math.sin(slope_rad) def _sigma1(turb_i: float, vhub: float) -> float: return turb_i * (0.75 * vhub + 5.6) def _resolve_output_path(code: str, output_dir: str | Path | None = None) -> Path: if output_dir is None: return Path(f"{code}.wnd") output_path = Path(output_dir) output_path.mkdir(parents=True, exist_ok=True) return output_path / f"{code}.wnd"
[docs] def gen_ecd(code: str, params: IECParameters, output_dir: str | Path | None = None) -> Path: """Generate an Extreme Coherent Gust with Direction Change (ECD) ``.wnd`` file. Parameters ---------- code : str Condition code of the form ``ECD[+/-]R[+/-modifier]`` (e.g. ``"ECD+R"``, ``"ECD-R+1.5"``). The sign sets the direction-change sense; the optional modifier (magnitude <= 2 m/s in user units) offsets the rated hub speed. params : IECParameters Turbine definition supplying rated speed, slope, and shear exponent. output_dir : str or pathlib.Path or None, optional Output directory; created if missing. ``None`` writes to the current working directory. Returns ------- pathlib.Path Path to the written ``.wnd`` file. Raises ------ ValueError If ``code`` is malformed, the modifier exceeds +/-2 m/s, or the resulting hub speed exceeds the reference wind speed. See Also -------- gen_edc : Extreme Direction Change. gen_eog : Extreme Operating Gust. Notes ----- Implements the IEC 61400-1 Extreme Coherent Gust with Direction Change: a coherent gust of 15 m/s with a direction change of 180 deg for Vhub <= 4 m/s (else 720/Vhub deg), each ramped over 10 s. See :doc:`theory` for the equations. """ match = re.match(r"^ECD([+-])R([+-]?\d*\.?\d*)$", code) if not match: raise ValueError(f"Cannot parse ECD condition '{code}'. Expected format: ECD[+/-]R[+/-speed_modifier]") dir_sign = 1.0 if match.group(1) == "+" else -1.0 speed_modifier = (float(match.group(2)) / params.len_convert) if match.group(2) else 0.0 if abs(speed_modifier) > 2.0: raise ValueError( f"ECD speed modifier must not exceed +/-2.0 m/s. " f"Got: {speed_modifier * params.len_convert:.2f} {params.spd_unit} [{code}]" ) vhub = params.vrated + speed_modifier if vhub > params.vref: raise ValueError(f"ECD: Vhub ({vhub:.2f} m/s) exceeds Vref ({params.vref:.2f} m/s). [{code}]") vhub_h, vhub_v0 = _hub_components(vhub, params.slope_rad) theta = (180.0 if vhub <= 4.0 else 720.0 / vhub) * dir_sign duration = 10.0 times = _transient_times(params.t1, duration) tau = times - params.t1 gust = 0.5 * VCG * (1.0 - np.cos(PI * tau / duration)) gust_h = gust * math.cos(params.slope_rad) vertical = (vhub + gust) * math.sin(params.slope_rad) theta_t = 0.5 * theta * (1.0 - np.cos(PI * tau / duration)) writer = WindFileWriter(_resolve_output_path(code, output_dir), params) writer.file_header() writer.wind_header("Extreme Coherent Gust with Direction Change") writer.time_header(duration) writer.gust_header(VCG, vhub, params.t1 + duration) writer.direction_header(theta, params.t1 + duration) writer.slope_header() writer.col_headers() writer.data_row(0.0, vhub_h, 0.0, vhub_v0, 0.0, params.alpha, 0.0, 0.0) for index, time in enumerate(times): writer.data_row( time, vhub_h, float(theta_t[index]), float(vertical[index]), 0.0, params.alpha, 0.0, float(gust_h[index]) ) return writer.save()
[docs] def gen_ews(code: str, params: IECParameters, output_dir: str | Path | None = None) -> Path: """Generate an Extreme Wind Shear (EWS) ``.wnd`` file. Parameters ---------- code : str Condition code ``EWS[V/H][+/-]<speed>`` (e.g. ``"EWSV+12.0"``). ``V``/``H`` selects vertical or horizontal shear, the sign selects the shear direction, and ``<speed>`` is the hub-height wind speed in user units. params : IECParameters Turbine definition supplying the operating range, turbulence, and geometry. output_dir : str or pathlib.Path or None, optional Output directory; created if missing. ``None`` writes to the current working directory. Returns ------- pathlib.Path Path to the written ``.wnd`` file. Raises ------ ValueError If ``code`` is malformed, or the wind speed is outside ``[vin, vout]``. Notes ----- Implements the IEC 61400-1 Extreme Wind Shear: a transient linear shear (vertical or horizontal) ramped over 12 s. See :doc:`theory` for the equations. """ match = re.match(r"^EWS([VH])([+-])(\d+\.?\d*)$", code) if not match: raise ValueError(f"Cannot parse EWS condition '{code}'. Expected format: EWS[V/H][+/-][wind_speed]") is_horizontal = match.group(1) == "H" dir_sign = 1.0 if match.group(2) == "+" else -1.0 vhub = float(match.group(3)) / params.len_convert if vhub < params.vin or vhub > params.vout: raise ValueError( f"EWS wind speed ({vhub * params.len_convert:.1f} {params.spd_unit}) must be between " f"Vin ({params.vin * params.len_convert:.1f}) and " f"Vout ({params.vout * params.len_convert:.1f}). [{code}]" ) vhub_h, vhub_v = _hub_components(vhub, params.slope_rad) sigma1 = _sigma1(params.turb_intensity, vhub) vg50 = BETA * sigma1 shr_max = 2.0 * (2.5 + 0.2 * vg50 * params.turb_rat**0.25) / vhub duration = 12.0 times = _transient_times(params.t1, duration) tau = times - params.t1 shear = dir_sign * 0.5 * shr_max * (1.0 - np.cos(2.0 * PI * tau / duration)) h_shr = shear if is_horizontal else np.zeros_like(shear) v_shr = shear if not is_horizontal else np.zeros_like(shear) writer = WindFileWriter(_resolve_output_path(code, output_dir), params) writer.file_header() writer.wind_header(f"Extreme {'Horizontal' if is_horizontal else 'Vertical'} Wind Shear") writer.time_header(duration) writer.shear_header(dir_sign, is_horizontal, shr_max, params.t1 + duration) writer.slope_header() writer.col_headers() writer.data_row(0.0, vhub_h, 0.0, vhub_v, 0.0, params.alpha, 0.0, 0.0) for index, time in enumerate(times): writer.data_row(time, vhub_h, 0.0, vhub_v, float(h_shr[index]), params.alpha, float(v_shr[index]), 0.0) return writer.save()
[docs] def gen_eog(code: str, params: IECParameters, output_dir: str | Path | None = None) -> Path: """Generate an Extreme Operating Gust (EOG) ``.wnd`` file. Parameters ---------- code : str Condition code ``EOGI``, ``EOGO``, ``EOGR``, or ``EOGR[+/-modifier]``. ``I``/``O``/``R`` select the cut-in, cut-out, or rated reference speed; a modifier (magnitude <= 2 m/s in user units) is valid only with ``R``. params : IECParameters Turbine definition supplying operating speeds, turbulence, and geometry. output_dir : str or pathlib.Path or None, optional Output directory; created if missing. ``None`` writes to the current working directory. Returns ------- pathlib.Path Path to the written ``.wnd`` file. Raises ------ ValueError If ``code`` is malformed (including a modifier on an ``I``/``O`` reference), or the modifier exceeds +/-2 m/s. Notes ----- Implements the IEC 61400-1 Extreme Operating Gust over a 10.5 s transient. See :doc:`theory` for the equations. """ # A speed modifier is only meaningful for the rated-speed (R) reference; the # grammar therefore rejects modifiers on the cut-in (I) and cut-out (O) cases # rather than silently ignoring them. match = re.match(r"^EOG(?:([IO])|R([+-]?\d*\.?\d*))$", code) if not match: raise ValueError( f"Cannot parse EOG condition '{code}'. Expected format: EOGI, EOGO, EOGR, or EOGR[+/-speed_modifier]" ) reference = match.group(1) if reference == "I": vhub = params.vin elif reference == "O": vhub = params.vout else: modifier_text = match.group(2) speed_modifier = (float(modifier_text) / params.len_convert) if modifier_text else 0.0 if abs(speed_modifier) > 2.0: raise ValueError( f"EOG speed modifier must not exceed +/-2.0 m/s. " f"Got: {speed_modifier * params.len_convert:.2f} {params.spd_unit} [{code}]" ) vhub = params.vrated + speed_modifier vhub_h, vhub_v0 = _hub_components(vhub, params.slope_rad) sigma1 = _sigma1(params.turb_intensity, vhub) gust_max = min(1.35 * (params.ve1 - vhub), 3.3 * sigma1 / (1.0 + 0.1 * params.turb_rat)) duration = 10.5 times = _transient_times(params.t1, duration) tau = times - params.t1 gust_factor = -0.37 * gust_max * np.sin(3.0 * PI * tau / duration) * (1.0 - np.cos(2.0 * PI * tau / duration)) gust_h = gust_factor * math.cos(params.slope_rad) vertical = (vhub + gust_factor) * math.sin(params.slope_rad) writer = WindFileWriter(_resolve_output_path(code, output_dir), params) writer.file_header() writer.wind_header("Extreme Operating Gust") writer.time_header(duration) writer.gust_header(2 * 0.37 * gust_max, vhub, params.t1 + 0.5 * duration) writer.slope_header() writer.col_headers() writer.data_row(0.0, vhub_h, 0.0, vhub_v0, 0.0, params.alpha, 0.0, 0.0) for index, time in enumerate(times): writer.data_row(time, vhub_h, 0.0, float(vertical[index]), 0.0, params.alpha, 0.0, float(gust_h[index])) return writer.save()
[docs] def gen_edc(code: str, params: IECParameters, output_dir: str | Path | None = None) -> Path: """Generate an Extreme Direction Change (EDC) ``.wnd`` file. Parameters ---------- code : str Condition code ``EDC[+/-]I``, ``EDC[+/-]O``, ``EDC[+/-]R``, or ``EDC[+/-]R[+/-modifier]``. The leading sign sets the yaw-excursion sense; ``I``/``O``/``R`` select the reference speed; a modifier is valid only with ``R``. params : IECParameters Turbine definition supplying operating speeds, turbulence, and geometry. output_dir : str or pathlib.Path or None, optional Output directory; created if missing. ``None`` writes to the current working directory. Returns ------- pathlib.Path Path to the written ``.wnd`` file. Raises ------ ValueError If ``code`` is malformed (including a modifier on an ``I``/``O`` reference), or the modifier exceeds +/-2 m/s. Notes ----- Implements the IEC 61400-1 Extreme Direction Change ramped over 6 s. See :doc:`theory` for the equations. """ # As with EOG, a speed modifier is only valid for the rated-speed (R) # reference; modifiers on I/O are rejected instead of being ignored. match = re.match(r"^EDC([+-])(?:([IO])|R([+-]?\d*\.?\d*))$", code) if not match: raise ValueError( f"Cannot parse EDC condition '{code}'. " f"Expected format: EDC[+/-]I, EDC[+/-]O, EDC[+/-]R, or EDC[+/-]R[+/-speed_modifier]" ) dir_sign = 1.0 if match.group(1) == "+" else -1.0 reference = match.group(2) if reference == "I": vhub = params.vin elif reference == "O": vhub = params.vout else: modifier_text = match.group(3) speed_modifier = (float(modifier_text) / params.len_convert) if modifier_text else 0.0 if abs(speed_modifier) > 2.0: raise ValueError( f"EDC speed modifier must not exceed +/-2.0 m/s. " f"Got: {speed_modifier * params.len_convert:.2f} {params.spd_unit} [{code}]" ) vhub = params.vrated + speed_modifier vhub_h, vhub_v = _hub_components(vhub, params.slope_rad) sigma1 = _sigma1(params.turb_intensity, vhub) theta = math.degrees(4.0 * math.atan(sigma1 / (vhub * (1.0 + 0.1 * params.turb_rat)))) * dir_sign duration = 6.0 times = _transient_times(params.t1, duration) tau = times - params.t1 theta_t = 0.5 * theta * (1.0 - np.cos(PI * tau / duration)) writer = WindFileWriter(_resolve_output_path(code, output_dir), params) writer.file_header() writer.wind_header("Extreme Direction Change") writer.time_header(duration) writer.direction_header(theta, params.t1 + duration) writer.slope_header() writer.col_headers() writer.data_row(0.0, vhub_h, 0.0, vhub_v, 0.0, params.alpha, 0.0, 0.0) for index, time in enumerate(times): writer.data_row(time, vhub_h, float(theta_t[index]), vhub_v, 0.0, params.alpha, 0.0, 0.0) return writer.save()
[docs] def gen_nwp(code: str, params: IECParameters, output_dir: str | Path | None = None) -> Path: """Generate a Normal Wind Profile (NWP) ``.wnd`` file. Parameters ---------- code : str Condition code ``NWP<speed>`` (e.g. ``"NWP10.0"``). The embedded speed is always interpreted in m/s, matching the historical IECWind convention, regardless of the ``si_unit`` setting. params : IECParameters Turbine definition supplying slope and the shear exponent. output_dir : str or pathlib.Path or None, optional Output directory; created if missing. ``None`` writes to the current working directory. Returns ------- pathlib.Path Path to the written ``.wnd`` file. Raises ------ ValueError If ``code`` is malformed. Notes ----- Implements the IEC 61400-1 Normal Wind Profile (power law). See :doc:`theory`. """ match = re.match(r"^NWP(\d+\.?\d*)$", code) if not match: raise ValueError(f"Cannot parse NWP condition '{code}'. Expected format: NWP[wind_speed_in_m/s]") vhub = float(match.group(1)) vhub_h, vhub_v = _hub_components(vhub, params.slope_rad) writer = WindFileWriter(_resolve_output_path(code, output_dir), params) writer.file_header() writer.wind_header("Normal Wind Profile") writer.nwp_header(vhub) writer.slope_header() writer.col_headers() writer.data_row(0.0, vhub_h, 0.0, vhub_v, 0.0, params.alpha, 0.0, 0.0) return writer.save()
[docs] def gen_ewm(code: str, params: IECParameters, output_dir: str | Path | None = None) -> Path: """Generate a steady Extreme Wind Model (EWM) ``.wnd`` file. Parameters ---------- code : str Condition code ``EWM50`` (50-year recurrence) or ``EWM01`` (1-year). params : IECParameters Turbine definition supplying the reference wind speed and slope. output_dir : str or pathlib.Path or None, optional Output directory; created if missing. ``None`` writes to the current working directory. Returns ------- pathlib.Path Path to the written ``.wnd`` file. Raises ------ ValueError If ``code`` is not ``EWM50`` or ``EWM01``. Notes ----- Implements the IEC 61400-1 steady Extreme Wind Model (Ve50 = 1.4*Vref, Ve1 = 0.8*Ve50). See :doc:`theory` for the equations. """ match = re.match(r"^EWM(50|01)$", code) if not match: raise ValueError(f"Cannot parse EWM condition '{code}'. Expected format: EWM50 or EWM01") if match.group(1) == "50": vhub = params.ve50 recurrence = "50-year" else: vhub = params.ve1 recurrence = "1-year" vhub_h, vhub_v = _hub_components(vhub, params.slope_rad) writer = WindFileWriter(_resolve_output_path(code, output_dir), params) writer.file_header() writer.wind_header(f"{recurrence} Extreme Wind Model") writer.ewm_header(vhub) writer.slope_header() writer.col_headers() writer.data_row(0.0, vhub_h, 0.0, vhub_v, 0.0, EWM_ALPHA, 0.0, 0.0) return writer.save()
_GENERATORS = { "ECD": gen_ecd, "EWS": gen_ews, "EOG": gen_eog, "EDC": gen_edc, "NWP": gen_nwp, "EWM": gen_ewm, } def _ordered_unique(codes: tuple[str, ...]) -> list[str]: """Return ``codes`` with duplicates removed, preserving first-seen order. Each condition code maps to a single ``<code>.wnd`` file, so a repeated code can only ever describe the same output. Generating it once keeps the result count honest and -- crucially for ``atomic=True`` -- avoids staging the same path twice and then failing to move it on the second commit pass. """ return list(dict.fromkeys(codes)) def _generate_one( code: str, params: IECParameters, output_dir: str | Path | None, *, strict: bool, generated: list[Path], errors: list[GenerationError], ) -> None: """Generate one condition, honouring the strict/lenient policy. Appends the written path to ``generated`` on success. On failure it re-raises when ``strict`` is ``True`` and otherwise records a :class:`GenerationError` in ``errors`` so the batch can continue. """ generator = _GENERATORS.get(code[:3]) if generator is None: message = f"Unknown condition type '{code[:3]}' in code '{code}'." if strict: raise ValueError(message) errors.append(GenerationError(code, message)) return try: generated.append(generator(code, params, output_dir=output_dir)) except ValueError as exc: if strict: raise errors.append(GenerationError(code, str(exc))) def _generate_all_atomic( params: IECParameters, output_dir: str | Path | None, *, strict: bool, ) -> GenerationResult: """All-or-nothing generation: stage every file, then commit by moving. Files are written first into a staging directory created *inside* the final output directory, so committing each file is a same-filesystem rename. The move phase only runs once every condition has been generated, so a failure partway through (under ``strict=True``) never leaves a partial batch behind. """ final_dir = Path.cwd() if output_dir is None else Path(output_dir) final_dir.mkdir(parents=True, exist_ok=True) staged: list[Path] = [] errors: list[GenerationError] = [] generated: list[Path] = [] with tempfile.TemporaryDirectory(prefix=".pyiecwind-staging-", dir=final_dir) as staging: staging_dir = Path(staging) for code in _ordered_unique(params.conditions): _generate_one(code, params, staging_dir, strict=strict, generated=staged, errors=errors) # Reaching here means every condition was generated (or, under # strict=False, recorded as an error). Now commit by moving into place. for staged_path in staged: destination = final_dir / staged_path.name os.replace(staged_path, destination) generated.append(destination) return GenerationResult(tuple(generated), tuple(errors))
[docs] def generate_all( params: IECParameters, output_dir: str | Path | None = None, *, strict: bool = True, atomic: bool = False, ) -> GenerationResult: """Generate every condition listed in ``params``. Fails closed by default: the first invalid condition raises, so a caller never silently receives partial output. Repeated condition codes describe the same ``<code>.wnd`` file and are generated once (first-seen order preserved). Parameters ---------- params : IECParameters Validated turbine definition and the list of condition codes to generate. output_dir : str or pathlib.Path or None, optional Directory to write the ``<code>.wnd`` files into; created if it does not exist. When ``None`` (the default) files are written to the current working directory. strict : bool, default True If ``True``, raise :class:`ValueError` on the first condition that cannot be generated. If ``False``, collect failures into the result and continue. atomic : bool, default False If ``True``, stage all files in a temporary directory and only move them into ``output_dir`` once the whole batch has been generated. Combined with ``strict=True`` this is all-or-nothing: an invalid condition raises and the output directory is left untouched, never partially populated. Returns ------- GenerationResult The paths written and any per-condition errors (errors are only populated when ``strict=False``). Raises ------ ValueError If ``strict`` is ``True`` and a condition code is unknown or invalid. Examples -------- >>> from pyiecwind import IECParameters, generate_all >>> params = IECParameters( ... si_unit=True, t1=40.0, wtc=2, catg="B", slope_deg=0.0, iec_edition=3, ... hh=80.0, dia=80.0, vin=4.0, vrated=10.0, vout=24.0, ... conditions=("EWM50", "NWP10.0"), ... ) >>> result = generate_all(params, output_dir="out", atomic=True) # doctest: +SKIP >>> result.count # doctest: +SKIP 2 """ if atomic: return _generate_all_atomic(params, output_dir, strict=strict) generated: list[Path] = [] errors: list[GenerationError] = [] for code in _ordered_unique(params.conditions): _generate_one(code, params, output_dir, strict=strict, generated=generated, errors=errors) return GenerationResult(tuple(generated), tuple(errors))
[docs] def generate_from_input_file( input_file: str | Path, *, output_dir: str | Path | None = None, strict: bool = True, atomic: bool = False, ) -> tuple[IECParameters, GenerationResult]: """Parse ``input_file`` and generate all of its conditions. A convenience wrapper combining :func:`parse_input_file` and :func:`generate_all`. Parameters ---------- input_file : str or pathlib.Path Path to an input file in any supported layout. output_dir : str or pathlib.Path or None, optional Directory for the generated ``.wnd`` files (see :func:`generate_all`). strict : bool, default True Passed through to :func:`generate_all`; fails closed by default. atomic : bool, default False Passed through to :func:`generate_all`; when combined with ``strict``, a single invalid condition leaves no files behind. Returns ------- tuple of (IECParameters, GenerationResult) The parsed parameters and the generation outcome. Raises ------ FileNotFoundError If ``input_file`` does not exist. ValueError For malformed input, or (when ``strict=True``) an invalid condition. """ params = parse_input_file(input_file) return params, generate_all(params, output_dir=output_dir, strict=strict, atomic=atomic)