Skip to content

Group

SWAN group commands

BaseGroupComponent

Bases: BaseComponent

Base group component.

Base class for SWAN group components. This base class modifies the render() method to allow rendering of a list of components. It is not intended to be used directly but to be subclassed by other group components.

Source code in src/rompy_swan/components/group.py 
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
class BaseGroupComponent(BaseComponent):
    """Base group component.

    Base class for SWAN group components. This base class modifies the `render()`
    method to allow rendering of a list of components. It is not intended to be used
    directly but to be subclassed by other group components.

    """

    model_type: Literal["group", "GROUP"] = Field(
        default="group", description="Model type discriminator"
    )

    def render(self, *args, **kwargs) -> str:
        """Override base class to allow rendering list of components."""
        cmds = []
        for cmd in self.cmd():
            cmds.append(super().render(cmd))
        return "\n\n".join(cmds)

Attributes

model_type class-attribute instance-attribute 

model_type: Literal['group', 'GROUP'] = Field(default='group', description='Model type discriminator')

Functions

render 

render(*args, **kwargs) -> str

Override base class to allow rendering list of components.

Source code in src/rompy_swan/components/group.py 
94
95
96
97
98
99
def render(self, *args, **kwargs) -> str:
    """Override base class to allow rendering list of components."""
    cmds = []
    for cmd in self.cmd():
        cmds.append(super().render(cmd))
    return "\n\n".join(cmds)

STARTUP

Bases: BaseGroupComponent

Startup group component.

.. code-block:: text

PROJECT ...
SET ...
MODE ...
COORDINATES ...

This group component is used to group individual startup components. Only fields that are explicitly prescribed are rendered by this group component.

Examples

.. ipython:: python :okwarning:

from rompy_swan.components.startup import PROJECT, SET, MODE, COORDINATES
from rompy_swan.components.group import STARTUP
proj = PROJECT(nr="01")
set = SET(level=0.5, direction_convention="nautical")
mode = MODE(kind="nonstationary", dim="twodimensional")
coords = COORDINATES(kind=dict(model_type="spherical", projection="ccm"))
startup = STARTUP(
    project=proj,
    set=set,
    mode=mode,
    coordinates=coords,
)
print(startup.render())
Source code in src/rompy_swan/components/group.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
class STARTUP(BaseGroupComponent):
    """Startup group component.

    .. code-block:: text

        PROJECT ...
        SET ...
        MODE ...
        COORDINATES ...

    This group component is used to group individual startup components. Only fields
    that are explicitly prescribed are rendered by this group component.

    Examples
    --------

    .. ipython:: python
        :okwarning:

        from rompy_swan.components.startup import PROJECT, SET, MODE, COORDINATES
        from rompy_swan.components.group import STARTUP
        proj = PROJECT(nr="01")
        set = SET(level=0.5, direction_convention="nautical")
        mode = MODE(kind="nonstationary", dim="twodimensional")
        coords = COORDINATES(kind=dict(model_type="spherical", projection="ccm"))
        startup = STARTUP(
            project=proj,
            set=set,
            mode=mode,
            coordinates=coords,
        )
        print(startup.render())

    """

    model_type: Literal["startup", "STARTUP"] = Field(
        default="startup", description="Model type discriminator"
    )
    project: Optional[PROJECT_TYPE] = Field(default=None)
    set: Optional[SET_TYPE] = Field(default=None)
    mode: Optional[MODE_TYPE] = Field(default=None)
    coordinates: Optional[COORDINATES_TYPE] = Field(default=None)

    def cmd(self) -> str:
        """Command file string for this component."""
        repr = []
        if self.project is not None:
            repr += [f"{self.project.cmd()}"]
        if self.set is not None:
            repr += [f"{self.set.cmd()}"]
        if self.mode is not None:
            repr += [f"{self.mode.cmd()}"]
        if self.coordinates is not None:
            repr += [f"{self.coordinates.cmd()}"]
        return repr

Attributes

model_type class-attribute instance-attribute 

model_type: Literal['startup', 'STARTUP'] = Field(default='startup', description='Model type discriminator')

project class-attribute instance-attribute 

project: Optional[PROJECT_TYPE] = Field(default=None)

set class-attribute instance-attribute 

set: Optional[SET_TYPE] = Field(default=None)

mode class-attribute instance-attribute 

mode: Optional[MODE_TYPE] = Field(default=None)

coordinates class-attribute instance-attribute 

coordinates: Optional[COORDINATES_TYPE] = Field(default=None)

Functions

cmd 

cmd() -> str

Command file string for this component.

Source code in src/rompy_swan/components/group.py 
154
155
156
157
158
159
160
161
162
163
164
165
def cmd(self) -> str:
    """Command file string for this component."""
    repr = []
    if self.project is not None:
        repr += [f"{self.project.cmd()}"]
    if self.set is not None:
        repr += [f"{self.set.cmd()}"]
    if self.mode is not None:
        repr += [f"{self.mode.cmd()}"]
    if self.coordinates is not None:
        repr += [f"{self.coordinates.cmd()}"]
    return repr

INPGRIDS

Bases: BaseGroupComponent

SWAN input grids group component.

.. code-block:: text

INPGRID ...
READGRID ...

INPGRID ...
READGRID ...

...

This group component is a convenience to allow defining and rendering a list of input grid components.

Examples

.. ipython:: python :okwarning:

from rompy_swan.components.inpgrid import REGULAR, ICE
from rompy_swan.components.group import INPGRIDS
inpgrid_bottom = REGULAR(
    grid_type="bottom",
    excval=-99.0,
    xpinp=172.0,
    ypinp=-41.0,
    alpinp=0.0,
    mxinp=99,
    myinp=99,
    dxinp=0.005,
    dyinp=0.005,
    readinp=dict(fname1="bottom.txt"),
)
inpgrid_wind = REGULAR(
    grid_type="wind",
    excval=-99.0,
    xpinp=172.0,
    ypinp=-41.0,
    alpinp=0.0,
    mxinp=99,
    myinp=99,
    dxinp=0.005,
    dyinp=0.005,
    readinp=dict(fname1="wind.txt"),
    nonstationary=dict(
        tbeg="2019-01-01T00:00:00",
        tend="2019-01-07 00:00:00",
        delt=3600,
        dfmt="hr",
    ),
)
inpgrid_ice_cte = ICE(aice=0.8, hice=2.0)
inpgrids = INPGRIDS(inpgrids=[inpgrid_bottom, inpgrid_wind, inpgrid_ice_cte])
print(inpgrids.render())
Source code in src/rompy_swan/components/group.py 
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
class INPGRIDS(BaseGroupComponent):
    """SWAN input grids group component.

    .. code-block:: text

        INPGRID ...
        READGRID ...

        INPGRID ...
        READGRID ...

        ...

    This group component is a convenience to allow defining and rendering
    a list of input grid components.

    Examples
    --------

    .. ipython:: python
        :okwarning:

        from rompy_swan.components.inpgrid import REGULAR, ICE
        from rompy_swan.components.group import INPGRIDS
        inpgrid_bottom = REGULAR(
            grid_type="bottom",
            excval=-99.0,
            xpinp=172.0,
            ypinp=-41.0,
            alpinp=0.0,
            mxinp=99,
            myinp=99,
            dxinp=0.005,
            dyinp=0.005,
            readinp=dict(fname1="bottom.txt"),
        )
        inpgrid_wind = REGULAR(
            grid_type="wind",
            excval=-99.0,
            xpinp=172.0,
            ypinp=-41.0,
            alpinp=0.0,
            mxinp=99,
            myinp=99,
            dxinp=0.005,
            dyinp=0.005,
            readinp=dict(fname1="wind.txt"),
            nonstationary=dict(
                tbeg="2019-01-01T00:00:00",
                tend="2019-01-07 00:00:00",
                delt=3600,
                dfmt="hr",
            ),
        )
        inpgrid_ice_cte = ICE(aice=0.8, hice=2.0)
        inpgrids = INPGRIDS(inpgrids=[inpgrid_bottom, inpgrid_wind, inpgrid_ice_cte])
        print(inpgrids.render())

    """

    model_type: Literal["inpgrids"] = Field(
        default="inpgrids", description="Model type discriminator"
    )
    inpgrids: list[INPGRID_TYPE] = Field(
        min_length=1,
        description="List of input grid components",
    )

    @field_validator("inpgrids")
    @classmethod
    def ensure_unique_grid_type(cls, inpgrids: INPGRID_TYPE) -> INPGRID_TYPE:
        """Ensure that each grid type is unique."""
        grid_types = [inp.grid_type for inp in inpgrids if hasattr(inp, "grid_type")]
        if len(grid_types) != len(set(grid_types)):
            raise ValueError("Each grid type must be unique")
        return inpgrids

    def cmd(self) -> str | list:
        repr = []
        for inpgrid in self.inpgrids:
            repr += [inpgrid.cmd()]
        return repr

Attributes

model_type class-attribute instance-attribute 

model_type: Literal['inpgrids'] = Field(default='inpgrids', description='Model type discriminator')

inpgrids class-attribute instance-attribute 

inpgrids: list[INPGRID_TYPE] = Field(min_length=1, description='List of input grid components')

Functions

ensure_unique_grid_type classmethod 

ensure_unique_grid_type(inpgrids: INPGRID_TYPE) -> INPGRID_TYPE

Ensure that each grid type is unique.

Source code in src/rompy_swan/components/group.py 
245
246
247
248
249
250
251
252
@field_validator("inpgrids")
@classmethod
def ensure_unique_grid_type(cls, inpgrids: INPGRID_TYPE) -> INPGRID_TYPE:
    """Ensure that each grid type is unique."""
    grid_types = [inp.grid_type for inp in inpgrids if hasattr(inp, "grid_type")]
    if len(grid_types) != len(set(grid_types)):
        raise ValueError("Each grid type must be unique")
    return inpgrids

cmd 

cmd() -> str | list
Source code in src/rompy_swan/components/group.py 
254
255
256
257
258
def cmd(self) -> str | list:
    repr = []
    for inpgrid in self.inpgrids:
        repr += [inpgrid.cmd()]
    return repr

PHYSICS

Bases: BaseGroupComponent

Physics group component.

The physics group component is a convenience to allow specifying several individual components in a single command and check for consistency between them.

Exemples

.. ipython:: python :okwarning:

from rompy_swan.components.group import PHYSICS
gen = {"model_type": "gen3", "source_terms": {"model_type": "komen"}}
phys = PHYSICS(gen=gen)
print(phys.render())
phys = PHYSICS(
    gen=dict(model_type="gen3", source_terms={"model_type": "st6c1"}),
    negatinp={"model_type": "negatinp", "rdcoef": 0.04},
    sswell={"model_type": "zieger"},
    breaking={"model_type": "constant", "alpha": 1.0, "gamma": 0.73},
    friction={"model_type": "jonswap", "cfjon": 0.038},
)
print(phys.render())
Source code in src/rompy_swan/components/group.py 
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
class PHYSICS(BaseGroupComponent):
    """Physics group component.

    The physics group component is a convenience to allow specifying several individual
    components in a single command and check for consistency between them.

    Exemples
    --------

    .. ipython:: python
        :okwarning:

        from rompy_swan.components.group import PHYSICS
        gen = {"model_type": "gen3", "source_terms": {"model_type": "komen"}}
        phys = PHYSICS(gen=gen)
        print(phys.render())
        phys = PHYSICS(
            gen=dict(model_type="gen3", source_terms={"model_type": "st6c1"}),
            negatinp={"model_type": "negatinp", "rdcoef": 0.04},
            sswell={"model_type": "zieger"},
            breaking={"model_type": "constant", "alpha": 1.0, "gamma": 0.73},
            friction={"model_type": "jonswap", "cfjon": 0.038},
        )
        print(phys.render())

    """

    model_type: Literal["physics", "PHYSICS"] = Field(
        default="physics", description="Model type discriminator"
    )
    gen: Optional[GEN_TYPE] = Field(default=None)
    sswell: Optional[SSWELL_TYPE] = Field(default=None)
    negatinp: Optional[NEGATINP_TYPE] = Field(default=None)
    wcapping: Optional[WCAPPING_TYPE] = Field(default=None)
    quadrupl: Optional[QUADRUPL_TYPE] = Field(default=None)
    breaking: Optional[BREAKING_TYPE] = Field(default=None)
    friction: Optional[FRICTION_TYPE] = Field(default=None)
    triad: Optional[TRIAD_TYPE] = Field(default=None)
    vegetation: Optional[VEGETATION_TYPE] = Field(default=None)
    mud: Optional[MUD_TYPE] = Field(default=None)
    sice: Optional[SICE_TYPE] = Field(default=None)
    turbulence: Optional[TURBULENCE_TYPE] = Field(default=None)
    bragg: Optional[BRAGG_TYPE] = Field(default=None)
    limiter: Optional[LIMITER_TYPE] = Field(default=None)
    obstacle: Optional[OBSTACLE_TYPE] = Field(default=None)
    setup: Optional[SETUP_TYPE] = Field(default=None)
    diffraction: Optional[DIFFRACTION_TYPE] = Field(default=None)
    surfbeat: Optional[SURFBEAT_TYPE] = Field(default=None)
    scat: Optional[SCAT_TYPE] = Field(default=None)
    deactivate: Optional[OFF_TYPE] = Field(default=None)

    @field_validator("deactivate")
    @classmethod
    def deactivate_physics(cls, off: OFF_TYPE) -> OFF_TYPE:
        """Convert OFF to OFFS so list is rendered."""
        for phys in PhysicsOff:
            print(phys.value)
        return off

    @model_validator(mode="after")
    def negatinp_only_with_zieger(self) -> "PHYSICS":
        """Log a warning if NEGATINP is used with a non-ZIEGER SSWELL."""
        if self.negatinp is None:
            return self
        elif self.sswell is None:
            logger.warning(
                "The negative wind input NEGATINP is only intended to use with the "
                "swell dissipation SSWELL ZIEGER but no SSWELL has been specified."
            )
        elif self.sswell.model_type != "zieger":
            logger.warning(
                "The negative wind input NEGATINP is only intended to use with the "
                "swell dissipation SSWELL ZIEGER but the SSWELL "
                f"{self.sswell.model_type.upper()} has been specified."
            )
        return self

    def cmd(self):
        repr = []
        if self.gen is not None:
            repr += [self.gen.cmd()]
        if self.sswell is not None:
            repr += [f"{self.sswell.cmd()}"]
        if self.negatinp is not None:
            repr += [self.negatinp.cmd()]
        if self.wcapping is not None:
            repr += [self.wcapping.cmd()]
        if self.quadrupl is not None:
            repr += [self.quadrupl.cmd()]
        if self.breaking is not None:
            repr += [self.breaking.cmd()]
        if self.friction is not None:
            repr += [self.friction.cmd()]
        if self.triad is not None:
            repr += [self.triad.cmd()]
        if self.vegetation is not None:
            repr += [self.vegetation.cmd()]
        if self.mud is not None:
            repr += [self.mud.cmd()]
        if self.sice is not None:
            repr += [self.sice.cmd()]
        if self.turbulence is not None:
            repr += [self.turbulence.cmd()]
        if self.bragg is not None:
            repr += [self.bragg.cmd()]
        if self.limiter is not None:
            repr += [self.limiter.cmd()]
        if self.obstacle is not None:
            repr += self.obstacle.cmd()  # Object returns a list of components
        if self.setup is not None:
            repr += [self.setup.cmd()]
        if self.diffraction is not None:
            repr += [self.diffraction.cmd()]
        if self.surfbeat is not None:
            repr += [self.surfbeat.cmd()]
        if self.scat is not None:
            repr += [self.scat.cmd()]
        if self.deactivate is not None:
            repr += self.deactivate.cmd()  # Object returns a list of components
        return repr

Attributes

model_type class-attribute instance-attribute 

model_type: Literal['physics', 'PHYSICS'] = Field(default='physics', description='Model type discriminator')

gen class-attribute instance-attribute 

gen: Optional[GEN_TYPE] = Field(default=None)

sswell class-attribute instance-attribute 

sswell: Optional[SSWELL_TYPE] = Field(default=None)

negatinp class-attribute instance-attribute 

negatinp: Optional[NEGATINP_TYPE] = Field(default=None)

wcapping class-attribute instance-attribute 

wcapping: Optional[WCAPPING_TYPE] = Field(default=None)

quadrupl class-attribute instance-attribute 

quadrupl: Optional[QUADRUPL_TYPE] = Field(default=None)

breaking class-attribute instance-attribute 

breaking: Optional[BREAKING_TYPE] = Field(default=None)

friction class-attribute instance-attribute 

friction: Optional[FRICTION_TYPE] = Field(default=None)

triad class-attribute instance-attribute 

triad: Optional[TRIAD_TYPE] = Field(default=None)

vegetation class-attribute instance-attribute 

vegetation: Optional[VEGETATION_TYPE] = Field(default=None)

mud class-attribute instance-attribute 

mud: Optional[MUD_TYPE] = Field(default=None)

sice class-attribute instance-attribute 

sice: Optional[SICE_TYPE] = Field(default=None)

turbulence class-attribute instance-attribute 

turbulence: Optional[TURBULENCE_TYPE] = Field(default=None)

bragg class-attribute instance-attribute 

bragg: Optional[BRAGG_TYPE] = Field(default=None)

limiter class-attribute instance-attribute 

limiter: Optional[LIMITER_TYPE] = Field(default=None)

obstacle class-attribute instance-attribute 

obstacle: Optional[OBSTACLE_TYPE] = Field(default=None)

setup class-attribute instance-attribute 

setup: Optional[SETUP_TYPE] = Field(default=None)

diffraction class-attribute instance-attribute 

diffraction: Optional[DIFFRACTION_TYPE] = Field(default=None)

surfbeat class-attribute instance-attribute 

surfbeat: Optional[SURFBEAT_TYPE] = Field(default=None)

scat class-attribute instance-attribute 

scat: Optional[SCAT_TYPE] = Field(default=None)

deactivate class-attribute instance-attribute 

deactivate: Optional[OFF_TYPE] = Field(default=None)

Functions

deactivate_physics classmethod 

deactivate_physics(off: OFF_TYPE) -> OFF_TYPE

Convert OFF to OFFS so list is rendered.

Source code in src/rompy_swan/components/group.py 
361
362
363
364
365
366
367
@field_validator("deactivate")
@classmethod
def deactivate_physics(cls, off: OFF_TYPE) -> OFF_TYPE:
    """Convert OFF to OFFS so list is rendered."""
    for phys in PhysicsOff:
        print(phys.value)
    return off

negatinp_only_with_zieger 

negatinp_only_with_zieger() -> PHYSICS

Log a warning if NEGATINP is used with a non-ZIEGER SSWELL.

Source code in src/rompy_swan/components/group.py 
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
@model_validator(mode="after")
def negatinp_only_with_zieger(self) -> "PHYSICS":
    """Log a warning if NEGATINP is used with a non-ZIEGER SSWELL."""
    if self.negatinp is None:
        return self
    elif self.sswell is None:
        logger.warning(
            "The negative wind input NEGATINP is only intended to use with the "
            "swell dissipation SSWELL ZIEGER but no SSWELL has been specified."
        )
    elif self.sswell.model_type != "zieger":
        logger.warning(
            "The negative wind input NEGATINP is only intended to use with the "
            "swell dissipation SSWELL ZIEGER but the SSWELL "
            f"{self.sswell.model_type.upper()} has been specified."
        )
    return self

cmd 

cmd()
Source code in src/rompy_swan/components/group.py 
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
def cmd(self):
    repr = []
    if self.gen is not None:
        repr += [self.gen.cmd()]
    if self.sswell is not None:
        repr += [f"{self.sswell.cmd()}"]
    if self.negatinp is not None:
        repr += [self.negatinp.cmd()]
    if self.wcapping is not None:
        repr += [self.wcapping.cmd()]
    if self.quadrupl is not None:
        repr += [self.quadrupl.cmd()]
    if self.breaking is not None:
        repr += [self.breaking.cmd()]
    if self.friction is not None:
        repr += [self.friction.cmd()]
    if self.triad is not None:
        repr += [self.triad.cmd()]
    if self.vegetation is not None:
        repr += [self.vegetation.cmd()]
    if self.mud is not None:
        repr += [self.mud.cmd()]
    if self.sice is not None:
        repr += [self.sice.cmd()]
    if self.turbulence is not None:
        repr += [self.turbulence.cmd()]
    if self.bragg is not None:
        repr += [self.bragg.cmd()]
    if self.limiter is not None:
        repr += [self.limiter.cmd()]
    if self.obstacle is not None:
        repr += self.obstacle.cmd()  # Object returns a list of components
    if self.setup is not None:
        repr += [self.setup.cmd()]
    if self.diffraction is not None:
        repr += [self.diffraction.cmd()]
    if self.surfbeat is not None:
        repr += [self.surfbeat.cmd()]
    if self.scat is not None:
        repr += [self.scat.cmd()]
    if self.deactivate is not None:
        repr += self.deactivate.cmd()  # Object returns a list of components
    return repr

OUTPUT

Bases: BaseGroupComponent

Output group component.

.. code-block:: text

FRAME 'sname' ...
GROUP 'sname' ...
CURVE 'sname' ...
RAY 'rname' ...
ISOLINE 'sname' 'rname' ...
POINTS 'sname ...
NGRID 'sname' ...
QUANTITY ...
OUTPUT OPTIONS ...
BLOCK 'sname' ...
TABLE 'sname' ...
SPECOUT 'sname' ...
NESTOUT 'sname ...

This group component is used to define multiple types of output locations and write components in a single model. Only fields that are explicitly prescribed are rendered by this group component.

Note

The components prescribed are validated according to some constraints as defined in the SWAN manual:

  • The name 'sname' of each Locations component must be unique.
  • The Locations 'sname' assigned to each write component must be defined.
  • The BLOCK component must be associated with either a FRAME or GROUP.
  • The ISOLINE write component must be associated with a RAY component.
  • The NGRID and NESTOUT components must be defined together.

Examples

.. ipython:: python :okwarning:

from rompy_swan.components.output import POINTS, BLOCK, QUANTITIES, TABLE
from rompy_swan.components.group import OUTPUT
points = POINTS(sname="outpts", xp=[172.3, 172.4], yp=[-39, -39])
quantity = QUANTITIES(
    quantities=[
        dict(output=["depth", "hsign", "tps", "dir", "tm01"], excv=-9),
    ]
)
times = dict(tbeg="2012-01-01T00:00:00", delt="PT30M", tfmt=1, dfmt="min")
block = BLOCK(
    model_type="block",
    sname="COMPGRID",
    fname="./swangrid.nc",
    output=["depth", "hsign", "tps", "dir"],
    times=times,
)
table = TABLE(
    sname="outpts",
    format="noheader",
    fname="./swantable.nc",
    output=["hsign", "hswell", "dir", "tps", "tm01", "watlev", "qp"],
    times=times,
)
out = OUTPUT(
    points=points,
    quantity=quantity,
    block=block,
    table=table,
)
print(out.render())
Source code in src/rompy_swan/components/group.py 
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
class OUTPUT(BaseGroupComponent):
    """Output group component.

    .. code-block:: text

        FRAME 'sname' ...
        GROUP 'sname' ...
        CURVE 'sname' ...
        RAY 'rname' ...
        ISOLINE 'sname' 'rname' ...
        POINTS 'sname ...
        NGRID 'sname' ...
        QUANTITY ...
        OUTPUT OPTIONS ...
        BLOCK 'sname' ...
        TABLE 'sname' ...
        SPECOUT 'sname' ...
        NESTOUT 'sname ...

    This group component is used to define multiple types of output locations and
    write components in a single model. Only fields that are explicitly prescribed are
    rendered by this group component.

    Note
    ----
    The components prescribed are validated according to some constraints as defined
    in the SWAN manual:

    - The name `'sname'` of each Locations component must be unique.
    - The Locations `'sname'` assigned to each write component must be defined.
    - The BLOCK component must be associated with either a `FRAME` or `GROUP`.
    - The ISOLINE write component must be associated with a `RAY` component.
    - The NGRID and NESTOUT components must be defined together.

    Examples
    --------

    .. ipython:: python
        :okwarning:

        from rompy_swan.components.output import POINTS, BLOCK, QUANTITIES, TABLE
        from rompy_swan.components.group import OUTPUT
        points = POINTS(sname="outpts", xp=[172.3, 172.4], yp=[-39, -39])
        quantity = QUANTITIES(
            quantities=[
                dict(output=["depth", "hsign", "tps", "dir", "tm01"], excv=-9),
            ]
        )
        times = dict(tbeg="2012-01-01T00:00:00", delt="PT30M", tfmt=1, dfmt="min")
        block = BLOCK(
            model_type="block",
            sname="COMPGRID",
            fname="./swangrid.nc",
            output=["depth", "hsign", "tps", "dir"],
            times=times,
        )
        table = TABLE(
            sname="outpts",
            format="noheader",
            fname="./swantable.nc",
            output=["hsign", "hswell", "dir", "tps", "tm01", "watlev", "qp"],
            times=times,
        )
        out = OUTPUT(
            points=points,
            quantity=quantity,
            block=block,
            table=table,
        )
        print(out.render())

    """

    model_type: Literal["output", "OUTPUT"] = Field(
        default="output", description="Model type discriminator"
    )
    frame: Optional[FRAME_TYPE] = Field(default=None)
    group: Optional[GROUP_TYPE] = Field(default=None)
    curve: Optional[CURVE_TYPE] = Field(default=None)
    ray: Optional[RAY_TYPE] = Field(default=None)
    isoline: Optional[ISOLINE_TYPE] = Field(default=None)
    points: Optional[POINTS_TYPE] = Field(default=None)
    ngrid: Optional[NGRID_TYPE] = Field(default=None)
    quantity: Optional[QUANTITY_TYPE] = Field(default=None)
    output_options: Optional[OUTOPT_TYPE] = Field(default=None)
    block: Optional[BLOCK_TYPE] = Field(default=None)
    table: Optional[TABLE_TYPE] = Field(default=None)
    specout: Optional[SPECOUT_TYPE] = Field(default=None)
    nestout: Optional[NESTOUT_TYPE] = Field(default=None)
    test: Optional[TEST_TYPE] = Field(default=None)
    _location_fields: list = ["frame", "group", "curve", "isoline", "points", "ngrid"]
    _write_fields: list = ["block", "table", "specout", "nestout"]

    @model_validator(mode="after")
    def write_locations_exists(self) -> "OUTPUT":
        """Ensure the location component requested by a write component exists."""
        for write in self.write_set:
            obj = getattr(self, write)
            if obj is None:
                continue
            snames = obj.sname if isinstance(obj.sname, list) else [obj.sname]
            for sname in snames:
                if sname in SPECIAL_NAMES:
                    return self
                try:
                    self._filter_location(sname)
                except ValueError as err:
                    raise ValueError(
                        f"Write component '{write}' specified with sname='{sname}' but "
                        f"no location component with sname='{sname}' has been defined"
                    ) from err
        return self

    @model_validator(mode="after")
    def locations_sname_unique(self) -> "OUTPUT":
        """Ensure same `sname` isn't used in more than one set of output locations."""
        duplicates = {x for x in self.snames if self.snames.count(x) > 1}
        if duplicates:
            raise ValueError(
                "The following snames are used to define more than one set of output "
                f"components: {duplicates}, please ensure each location component has "
                "a unique `sname`"
            )
        return self

    @model_validator(mode="after")
    def block_with_frame_or_group(self) -> "OUTPUT":
        """Ensure Block is only defined for FRAME or GROUP locations."""
        if self.block is not None:
            snames = self.block.sname
            if isinstance(snames, str):
                snames = [self.block.sname]
            for sname in snames:
                if sname not in ["BOTTGRID", "COMPGRID"]:
                    location = self._filter_location(sname)
                    component = location.model_type.upper().split("_")[0]
                    if component not in ["FRAME", "GROUP"]:
                        raise ValueError(
                            f"Block sname='{sname}' specified with {component} "
                            "location component but only only FRAME or GROUP "
                            "components are supported"
                        )
        return self

    @model_validator(mode="after")
    def isoline_ray_defined(self) -> "OUTPUT":
        """Ensure the isoline ray has been defined."""
        if self.isoline is not None:
            if self.ray is None:
                raise ValueError(
                    f"Isoline {self.isoline} requires RAY rname='{self.isoline.rname}'"
                    " but no RAY component has been defined"
                )
            elif self.ray.rname != self.isoline.rname:
                raise ValueError(
                    f"Isoline rname='{self.isoline.rname}' does not match "
                    f"the ray rname='{self.ray.rname}'"
                )
        return self

    @model_validator(mode="after")
    def ngrid_and_nestout(self) -> "OUTPUT":
        """Ensure NGRID and NESTOUT are specified together."""
        if self.ngrid is not None and self.nestout is None:
            raise ValueError(
                "NGRID component specified but no NESTOUT component has been defined"
            )
        elif self.ngrid is None and self.nestout is not None:
            raise ValueError(
                "NESTOUT component specified but no NGRID component has been defined"
            )
        elif self.ngrid is not None and self.nestout is not None:
            if self.ngrid.sname != self.nestout.sname:
                raise ValueError(
                    f"NGRID sname='{self.ngrid.sname}' does not match "
                    f"the NESTOUT sname='{self.nestout.sname}'"
                )
        return self

    @property
    def locations_set(self):
        """List of specified location fields."""
        return [fld for fld in self.model_fields_set if fld in self._location_fields]

    @property
    def write_set(self):
        """List of specified write fields."""
        return [fld for fld in self.model_fields_set if fld in self._write_fields]

    @property
    def snames(self):
        """List of snames from specified location components."""
        snames = []
        for field in self.locations_set:
            obj = getattr(self, field)
            if obj is None:
                continue
            sname = obj.sname
            if isinstance(sname, str):
                sname = [sname]
            snames.extend(sname)
        return snames

    def _filter_location(self, sname):
        """Filter the location component defined with the specified sname."""
        for field in self.locations_set:
            obj = getattr(self, field)
            if obj is None:
                continue
            obj_snames = obj.sname if isinstance(obj.sname, list) else [obj.sname]
            for obj_sname in obj_snames:
                if obj_sname == sname:
                    return obj
        raise ValueError(f"Location component with sname='{sname}' not found")

    def cmd(self) -> list:
        """Command file string for this component."""
        repr = []
        if self.frame is not None:
            repr += [f"{self.frame.cmd()}"]
        if self.group is not None:
            repr += [f"{self.group.cmd()}"]
        if self.curve is not None:
            # Component renders a list
            repr += self.curve.cmd()
        if self.ray is not None:
            repr += [f"{self.ray.cmd()}"]
        if self.isoline is not None:
            repr += [f"{self.isoline.cmd()}"]
        if self.points is not None:
            repr += [f"{self.points.cmd()}"]
        if self.ngrid is not None:
            repr += [f"{self.ngrid.cmd()}"]
        if self.quantity is not None:
            # Component renders a list
            repr += self.quantity.cmd()
        if self.output_options is not None:
            repr += [f"{self.output_options.cmd()}"]
        if self.block is not None:
            # Component may or may not render a list, handles both
            cmds = self.block.cmd()
            if not isinstance(cmds, list):
                cmds = [cmds]
            for cmd in cmds:
                repr += [f"{cmd}"]
        if self.table is not None:
            repr += [f"{self.table.cmd()}"]
        if self.specout is not None:
            repr += [f"{self.specout.cmd()}"]
        if self.nestout is not None:
            repr += [f"{self.nestout.cmd()}"]
        if self.test is not None:
            repr += [f"{self.test.cmd()}"]
        return repr

Attributes

model_type class-attribute instance-attribute 

model_type: Literal['output', 'OUTPUT'] = Field(default='output', description='Model type discriminator')

frame class-attribute instance-attribute 

frame: Optional[FRAME_TYPE] = Field(default=None)

group class-attribute instance-attribute 

group: Optional[GROUP_TYPE] = Field(default=None)

curve class-attribute instance-attribute 

curve: Optional[CURVE_TYPE] = Field(default=None)

ray class-attribute instance-attribute 

ray: Optional[RAY_TYPE] = Field(default=None)

isoline class-attribute instance-attribute 

isoline: Optional[ISOLINE_TYPE] = Field(default=None)

points class-attribute instance-attribute 

points: Optional[POINTS_TYPE] = Field(default=None)

ngrid class-attribute instance-attribute 

ngrid: Optional[NGRID_TYPE] = Field(default=None)

quantity class-attribute instance-attribute 

quantity: Optional[QUANTITY_TYPE] = Field(default=None)

output_options class-attribute instance-attribute 

output_options: Optional[OUTOPT_TYPE] = Field(default=None)

block class-attribute instance-attribute 

block: Optional[BLOCK_TYPE] = Field(default=None)

table class-attribute instance-attribute 

table: Optional[TABLE_TYPE] = Field(default=None)

specout class-attribute instance-attribute 

specout: Optional[SPECOUT_TYPE] = Field(default=None)

nestout class-attribute instance-attribute 

nestout: Optional[NESTOUT_TYPE] = Field(default=None)

test class-attribute instance-attribute 

test: Optional[TEST_TYPE] = Field(default=None)

locations_set property 

locations_set

List of specified location fields.

write_set property 

write_set

List of specified write fields.

snames property 

snames

List of snames from specified location components.

Functions

write_locations_exists 

write_locations_exists() -> OUTPUT

Ensure the location component requested by a write component exists.

Source code in src/rompy_swan/components/group.py 
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
@model_validator(mode="after")
def write_locations_exists(self) -> "OUTPUT":
    """Ensure the location component requested by a write component exists."""
    for write in self.write_set:
        obj = getattr(self, write)
        if obj is None:
            continue
        snames = obj.sname if isinstance(obj.sname, list) else [obj.sname]
        for sname in snames:
            if sname in SPECIAL_NAMES:
                return self
            try:
                self._filter_location(sname)
            except ValueError as err:
                raise ValueError(
                    f"Write component '{write}' specified with sname='{sname}' but "
                    f"no location component with sname='{sname}' has been defined"
                ) from err
    return self

locations_sname_unique 

locations_sname_unique() -> OUTPUT

Ensure same sname isn't used in more than one set of output locations.

Source code in src/rompy_swan/components/group.py 
573
574
575
576
577
578
579
580
581
582
583
@model_validator(mode="after")
def locations_sname_unique(self) -> "OUTPUT":
    """Ensure same `sname` isn't used in more than one set of output locations."""
    duplicates = {x for x in self.snames if self.snames.count(x) > 1}
    if duplicates:
        raise ValueError(
            "The following snames are used to define more than one set of output "
            f"components: {duplicates}, please ensure each location component has "
            "a unique `sname`"
        )
    return self

block_with_frame_or_group 

block_with_frame_or_group() -> OUTPUT

Ensure Block is only defined for FRAME or GROUP locations.

Source code in src/rompy_swan/components/group.py 
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
@model_validator(mode="after")
def block_with_frame_or_group(self) -> "OUTPUT":
    """Ensure Block is only defined for FRAME or GROUP locations."""
    if self.block is not None:
        snames = self.block.sname
        if isinstance(snames, str):
            snames = [self.block.sname]
        for sname in snames:
            if sname not in ["BOTTGRID", "COMPGRID"]:
                location = self._filter_location(sname)
                component = location.model_type.upper().split("_")[0]
                if component not in ["FRAME", "GROUP"]:
                    raise ValueError(
                        f"Block sname='{sname}' specified with {component} "
                        "location component but only only FRAME or GROUP "
                        "components are supported"
                    )
    return self

isoline_ray_defined 

isoline_ray_defined() -> OUTPUT

Ensure the isoline ray has been defined.

Source code in src/rompy_swan/components/group.py 
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
@model_validator(mode="after")
def isoline_ray_defined(self) -> "OUTPUT":
    """Ensure the isoline ray has been defined."""
    if self.isoline is not None:
        if self.ray is None:
            raise ValueError(
                f"Isoline {self.isoline} requires RAY rname='{self.isoline.rname}'"
                " but no RAY component has been defined"
            )
        elif self.ray.rname != self.isoline.rname:
            raise ValueError(
                f"Isoline rname='{self.isoline.rname}' does not match "
                f"the ray rname='{self.ray.rname}'"
            )
    return self

ngrid_and_nestout 

ngrid_and_nestout() -> OUTPUT

Ensure NGRID and NESTOUT are specified together.

Source code in src/rompy_swan/components/group.py 
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
@model_validator(mode="after")
def ngrid_and_nestout(self) -> "OUTPUT":
    """Ensure NGRID and NESTOUT are specified together."""
    if self.ngrid is not None and self.nestout is None:
        raise ValueError(
            "NGRID component specified but no NESTOUT component has been defined"
        )
    elif self.ngrid is None and self.nestout is not None:
        raise ValueError(
            "NESTOUT component specified but no NGRID component has been defined"
        )
    elif self.ngrid is not None and self.nestout is not None:
        if self.ngrid.sname != self.nestout.sname:
            raise ValueError(
                f"NGRID sname='{self.ngrid.sname}' does not match "
                f"the NESTOUT sname='{self.nestout.sname}'"
            )
    return self

cmd 

cmd() -> list

Command file string for this component.

Source code in src/rompy_swan/components/group.py 
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
def cmd(self) -> list:
    """Command file string for this component."""
    repr = []
    if self.frame is not None:
        repr += [f"{self.frame.cmd()}"]
    if self.group is not None:
        repr += [f"{self.group.cmd()}"]
    if self.curve is not None:
        # Component renders a list
        repr += self.curve.cmd()
    if self.ray is not None:
        repr += [f"{self.ray.cmd()}"]
    if self.isoline is not None:
        repr += [f"{self.isoline.cmd()}"]
    if self.points is not None:
        repr += [f"{self.points.cmd()}"]
    if self.ngrid is not None:
        repr += [f"{self.ngrid.cmd()}"]
    if self.quantity is not None:
        # Component renders a list
        repr += self.quantity.cmd()
    if self.output_options is not None:
        repr += [f"{self.output_options.cmd()}"]
    if self.block is not None:
        # Component may or may not render a list, handles both
        cmds = self.block.cmd()
        if not isinstance(cmds, list):
            cmds = [cmds]
        for cmd in cmds:
            repr += [f"{cmd}"]
    if self.table is not None:
        repr += [f"{self.table.cmd()}"]
    if self.specout is not None:
        repr += [f"{self.specout.cmd()}"]
    if self.nestout is not None:
        repr += [f"{self.nestout.cmd()}"]
    if self.test is not None:
        repr += [f"{self.test.cmd()}"]
    return repr

LOCKUP

Bases: BaseComponent

Lockup group component.

.. code-block:: text

COMPUTE ...
HOTFILE ...
COMPUTE ...
HOTFILE ...
...
STOP

This is a group component to specify SWAN "Lockup" commands including multiple COMPUTE commands that may or may not be interleaved with HOTFILE commands, and a final STOP command.

Examples

.. ipython:: python :okwarning:

from rompy_swan.components.group import LOCKUP
lockup = LOCKUP(
    compute=dict(
        model_type="stat",
        times=dict(
            model_type="nonstationary",
            tbeg="1990-01-01T00:00:00",
            tend="1990-01-01T03:00:00",
            delt="PT1H",
            dfmt="hr",
        ),
        hotfile=dict(fname="hotfile"),
        hottimes=[-1],
    ),
)
print(lockup.render())
Source code in src/rompy_swan/components/group.py 
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
class LOCKUP(BaseComponent):
    """Lockup group component.

    .. code-block:: text

        COMPUTE ...
        HOTFILE ...
        COMPUTE ...
        HOTFILE ...
        ...
        STOP

    This is a group component to specify SWAN "Lockup" commands including multiple
    `COMPUTE` commands that may or may not be interleaved with `HOTFILE` commands,
    and a final `STOP` command.

    Examples
    --------

    .. ipython:: python
        :okwarning:

        from rompy_swan.components.group import LOCKUP
        lockup = LOCKUP(
            compute=dict(
                model_type="stat",
                times=dict(
                    model_type="nonstationary",
                    tbeg="1990-01-01T00:00:00",
                    tend="1990-01-01T03:00:00",
                    delt="PT1H",
                    dfmt="hr",
                ),
                hotfile=dict(fname="hotfile"),
                hottimes=[-1],
            ),
        )
        print(lockup.render())

    """

    model_type: Literal["lockup", "LOCKUP"] = Field(
        default="lockup", description="Model type discriminator"
    )
    compute: COMPUTE_TYPE = Field(description="Compute components")

    def cmd(self) -> list:
        """Command file strings for this component."""
        return self.compute.cmd() + [STOP().render()]

Attributes

model_type class-attribute instance-attribute 

model_type: Literal['lockup', 'LOCKUP'] = Field(default='lockup', description='Model type discriminator')

compute class-attribute instance-attribute 

compute: COMPUTE_TYPE = Field(description='Compute components')

Functions

cmd 

cmd() -> list

Command file strings for this component.

Source code in src/rompy_swan/components/group.py 
771
772
773
def cmd(self) -> list:
    """Command file strings for this component."""
    return self.compute.cmd() + [STOP().render()]