diff --git a/lib/iris/experimental/ugrid.py b/lib/iris/experimental/ugrid.py index 30498aa819..86c80bb2c1 100644 --- a/lib/iris/experimental/ugrid.py +++ b/lib/iris/experimental/ugrid.py @@ -903,6 +903,18 @@ def __init__( edge_dimension=None, face_dimension=None, ): + """ + .. note:: + + :attr:`node_dimension`, :attr:`edge_dimension` and + :attr:`face_dimension` are stored to help round-tripping of UGRID + files. As such their presence in :class:`Mesh` is not a direct + mirror of that written in the UGRID specification, where + :attr:`node_dimension` is not mentioned, while + :attr:`edge_dimension` is only present for + :attr:`topology_dimension` ``>=2``. + + """ # TODO: support volumes. # TODO: support (coord, "z") @@ -1130,7 +1142,16 @@ def face_dimension(self): @face_dimension.setter def face_dimension(self, name): - if not name or not isinstance(name, str): + if self.topology_dimension < 2: + face_dimension = None + if name: + # Tell the user it is not being set if they expected otherwise. + message = ( + "Not setting face_dimension (inappropriate for " + f"topology_dimension={self.topology_dimension} ." + ) + logger.debug(message) + elif not name or not isinstance(name, str): face_dimension = f"Mesh{self.topology_dimension}d_face" else: face_dimension = name @@ -1199,14 +1220,19 @@ def add_coords( face_x=None, face_y=None, ): - self._coord_manager.add( - node_x=node_x, - node_y=node_y, - edge_x=edge_x, - edge_y=edge_y, - face_x=face_x, - face_y=face_y, - ) + # Filter out absent arguments - only expecting face coords sometimes, + # same will be true of volumes in future. + kwargs = { + "node_x": node_x, + "node_y": node_y, + "edge_x": edge_x, + "edge_y": edge_y, + "face_x": face_x, + "face_y": face_y, + } + kwargs = {k: v for k, v in kwargs.items() if v} + + self._coord_manager.add(**kwargs) def add_connectivities(self, *connectivities): self._connectivity_manager.add(*connectivities) @@ -1219,9 +1245,9 @@ def connectivities( var_name=None, attributes=None, cf_role=None, - node=None, - edge=None, - face=None, + contains_node=None, + contains_edge=None, + contains_face=None, ): return self._connectivity_manager.filters( item=item, @@ -1230,9 +1256,9 @@ def connectivities( var_name=var_name, attributes=attributes, cf_role=cf_role, - node=node, - edge=edge, - face=face, + contains_node=contains_node, + contains_edge=contains_edge, + contains_face=contains_face, ) def connectivity( @@ -1243,9 +1269,9 @@ def connectivity( var_name=None, attributes=None, cf_role=None, - node=None, - edge=None, - face=None, + contains_node=None, + contains_edge=None, + contains_face=None, ): return self._connectivity_manager.filter( item=item, @@ -1254,9 +1280,9 @@ def connectivity( var_name=var_name, attributes=attributes, cf_role=cf_role, - node=node, - edge=edge, - face=face, + contains_node=contains_node, + contains_edge=contains_edge, + contains_face=contains_face, ) def coord( @@ -1267,9 +1293,9 @@ def coord( var_name=None, attributes=None, axis=None, - node=None, - edge=None, - face=None, + include_nodes=None, + include_edges=None, + include_faces=None, ): return self._coord_manager.filter( item=item, @@ -1278,9 +1304,9 @@ def coord( var_name=var_name, attributes=attributes, axis=axis, - node=node, - edge=edge, - face=face, + include_nodes=include_nodes, + include_edges=include_edges, + include_faces=include_faces, ) def coords( @@ -1291,9 +1317,9 @@ def coords( var_name=None, attributes=None, axis=None, - node=False, - edge=False, - face=False, + include_nodes=None, + include_edges=None, + include_faces=None, ): return self._coord_manager.filters( item=item, @@ -1302,9 +1328,9 @@ def coords( var_name=var_name, attributes=attributes, axis=axis, - node=node, - edge=edge, - face=face, + include_nodes=include_nodes, + include_edges=include_edges, + include_faces=include_faces, ) def remove_connectivities( @@ -1315,9 +1341,9 @@ def remove_connectivities( var_name=None, attributes=None, cf_role=None, - node=None, - edge=None, - face=None, + contains_node=None, + contains_edge=None, + contains_face=None, ): return self._connectivity_manager.remove( item=item, @@ -1326,9 +1352,9 @@ def remove_connectivities( var_name=var_name, attributes=attributes, cf_role=cf_role, - node=node, - edge=edge, - face=face, + contains_node=contains_node, + contains_edge=contains_edge, + contains_face=contains_face, ) def remove_coords( @@ -1339,21 +1365,26 @@ def remove_coords( var_name=None, attributes=None, axis=None, - node=None, - edge=None, - face=None, + include_nodes=None, + include_edges=None, + include_faces=None, ): - return self._coord_manager.remove( - item=item, - standard_name=standard_name, - long_name=long_name, - var_name=var_name, - attributes=attributes, - axis=axis, - node=node, - edge=edge, - face=face, - ) + # Filter out absent arguments - only expecting face coords sometimes, + # same will be true of volumes in future. + kwargs = { + "item": item, + "standard_name": standard_name, + "long_name": long_name, + "var_name": var_name, + "attributes": attributes, + "axis": axis, + "include_nodes": include_nodes, + "include_edges": include_edges, + "include_faces": include_faces, + } + kwargs = {k: v for k, v in kwargs.items() if v} + + return self._coord_manager.remove(**kwargs) def xml_element(self): # TBD @@ -1651,16 +1682,19 @@ def filters( var_name=None, attributes=None, axis=None, - node=None, - edge=None, - face=None, + include_nodes=None, + include_edges=None, + include_faces=None, ): # TBD: support coord_systems? - # rationalise the tri-state behaviour - args = [node, edge, face] + # Preserve original argument before modifying. + face_requested = include_faces + + # Rationalise the tri-state behaviour. + args = [include_nodes, include_edges, include_faces] state = not any(set(filter(lambda arg: arg is not None, args))) - node, edge, face = map( + include_nodes, include_edges, include_faces = map( lambda arg: arg if arg is not None else state, args ) @@ -1668,14 +1702,14 @@ def populated_coords(coords_tuple): return list(filter(None, list(coords_tuple))) members = [] - if node: + if include_nodes: members += populated_coords(self.node_coords) - if edge: + if include_edges: members += populated_coords(self.edge_coords) if hasattr(self, "face_coords"): - if face: + if include_faces: members += populated_coords(self.face_coords) - else: + elif face_requested: dmsg = "Ignoring request to filter non-existent 'face_coords'" logger.debug(dmsg, extra=dict(cls=self.__class__.__name__)) @@ -1704,8 +1738,8 @@ def remove( var_name=None, attributes=None, axis=None, - node=None, - edge=None, + include_nodes=None, + include_edges=None, ): return self._remove( item=item, @@ -1714,8 +1748,8 @@ def remove( var_name=var_name, attributes=attributes, axis=axis, - node=node, - edge=edge, + include_nodes=include_nodes, + include_edges=include_edges, ) @@ -1794,9 +1828,9 @@ def remove( var_name=None, attributes=None, axis=None, - node=None, - edge=None, - face=None, + include_nodes=None, + include_edges=None, + include_faces=None, ): return self._remove( item=item, @@ -1805,9 +1839,9 @@ def remove( var_name=var_name, attributes=attributes, axis=axis, - node=node, - edge=edge, - face=face, + include_nodes=include_nodes, + include_edges=include_edges, + include_faces=include_faces, ) @@ -1820,9 +1854,7 @@ def __init__(self, *connectivities): cf_roles = [c.cf_role for c in connectivities] for requisite in self.REQUIRED: if requisite not in cf_roles: - message = ( - f"{self.__name__} requires a {requisite} Connectivity." - ) + message = f"{type(self).__name__} requires a {requisite} Connectivity." raise ValueError(message) self.ALL = self.REQUIRED + self.OPTIONAL @@ -1880,7 +1912,7 @@ def add(self, *connectivities): for connectivity in connectivities: if not isinstance(connectivity, Connectivity): message = f"Expected Connectivity, got: {type(connectivity)} ." - raise ValueError(message) + raise TypeError(message) cf_role = connectivity.cf_role if cf_role not in self.ALL: message = ( @@ -1954,9 +1986,9 @@ def filters( var_name=None, attributes=None, cf_role=None, - node=None, - edge=None, - face=None, + contains_node=None, + contains_edge=None, + contains_face=None, ): members = [c for c in self._members.values() if c is not None] @@ -1987,16 +2019,16 @@ def location_filter(instances, loc_arg, loc_name): return filtered for arg, loc in ( - (node, "node"), - (edge, "edge"), - (face, "face"), + (contains_node, "node"), + (contains_edge, "edge"), + (contains_face, "face"), ): members = location_filter(members, arg, loc) # No need to actually modify filtering behaviour - already won't return # any face cf-roles if none are present. supports_faces = any(["face" in role for role in self.ALL]) - if face and not supports_faces: + if contains_face and not supports_faces: message = ( "Ignoring request to filter for non-existent 'face' cf-roles." ) @@ -2026,9 +2058,9 @@ def remove( var_name=None, attributes=None, cf_role=None, - node=None, - edge=None, - face=None, + contains_node=None, + contains_edge=None, + contains_face=None, ): removal_dict = self.filters( item=item, @@ -2037,9 +2069,9 @@ def remove( var_name=var_name, attributes=attributes, cf_role=cf_role, - node=node, - edge=edge, - face=face, + contains_node=contains_node, + contains_edge=contains_edge, + contains_face=contains_face, ) for cf_role in self.REQUIRED: excluded = removal_dict.pop(cf_role, None) diff --git a/lib/iris/tests/unit/experimental/ugrid/test_Mesh.py b/lib/iris/tests/unit/experimental/ugrid/test_Mesh.py new file mode 100644 index 0000000000..2678a24e6e --- /dev/null +++ b/lib/iris/tests/unit/experimental/ugrid/test_Mesh.py @@ -0,0 +1,1089 @@ +# Copyright Iris contributors +# +# This file is part of Iris and is released under the LGPL license. +# See COPYING and COPYING.LESSER in the root of the repository for full +# licensing details. +"""Unit tests for the :class:`iris.experimental.ugrid.Mesh` class.""" + +# Import iris.tests first so that some things can be initialised before +# importing anything else. +import iris.tests as tests + +import numpy as np + +from iris.coords import AuxCoord +from iris.exceptions import ConnectivityNotFoundError, CoordinateNotFoundError +from iris.experimental import ugrid + + +class TestMeshCommon(tests.IrisTest): + @classmethod + def setUpClass(cls): + # A collection of minimal coords and connectivities describing an + # equilateral triangle. + cls.NODE_LON = AuxCoord( + [0, 2, 1], + standard_name="longitude", + long_name="long_name", + var_name="node_lon", + attributes={"test": 1}, + ) + cls.NODE_LAT = AuxCoord( + [0, 0, 1], standard_name="latitude", var_name="node_lat" + ) + cls.EDGE_LON = AuxCoord( + [1, 1.5, 0.5], standard_name="longitude", var_name="edge_lon" + ) + cls.EDGE_LAT = AuxCoord( + [0, 0.5, 0.5], standard_name="latitude", var_name="edge_lat" + ) + cls.FACE_LON = AuxCoord( + [0.5], standard_name="longitude", var_name="face_lon" + ) + cls.FACE_LAT = AuxCoord( + [0.5], standard_name="latitude", var_name="face_lat" + ) + + cls.EDGE_NODE = ugrid.Connectivity( + [[0, 1], [1, 2], [2, 0]], + cf_role="edge_node_connectivity", + long_name="long_name", + var_name="var_name", + attributes={"test": 1}, + ) + cls.FACE_NODE = ugrid.Connectivity( + [[0, 1, 2]], cf_role="face_node_connectivity" + ) + cls.FACE_EDGE = ugrid.Connectivity( + [[0, 1, 2]], cf_role="face_edge_connectivity" + ) + # (Actually meaningless:) + cls.FACE_FACE = ugrid.Connectivity( + [[0, 0, 0]], cf_role="face_face_connectivity" + ) + # (Actually meaningless:) + cls.EDGE_FACE = ugrid.Connectivity( + [[0, 0], [0, 0], [0, 0]], cf_role="edge_face_connectivity" + ) + cls.BOUNDARY_NODE = ugrid.Connectivity( + [[0, 1], [1, 2], [2, 0]], cf_role="boundary_node_connectivity" + ) + + +class TestProperties1D(TestMeshCommon): + # Tests that can re-use a single instance for greater efficiency. + @classmethod + def setUpClass(cls): + super().setUpClass() + # Mesh kwargs with topology_dimension=1 and all applicable + # arguments populated - this tests correct property setting. + cls.kwargs = { + "topology_dimension": 1, + "node_coords_and_axes": ((cls.NODE_LON, "x"), (cls.NODE_LAT, "y")), + "connectivities": cls.EDGE_NODE, + "long_name": "my_topology_mesh", + "var_name": "mesh", + "attributes": {"notes": "this is a test"}, + "node_dimension": "NodeDim", + "edge_dimension": "EdgeDim", + "edge_coords_and_axes": ((cls.EDGE_LON, "x"), (cls.EDGE_LAT, "y")), + } + cls.mesh = ugrid.Mesh(**cls.kwargs) + + def test__metadata_manager(self): + self.assertEqual( + self.mesh._metadata_manager.cls.__name__, + ugrid.MeshMetadata.__name__, + ) + + def test___getstate__(self): + expected = ( + self.mesh._metadata_manager, + self.mesh._coord_manager, + self.mesh._connectivity_manager, + ) + self.assertEqual(expected, self.mesh.__getstate__()) + + def test___repr__(self): + expected = ( + "Mesh(topology_dimension=1, node_coords_and_axes=[(AuxCoord(" + "array([0, 2, 1]), standard_name='longitude', units=Unit(" + "'unknown'), long_name='long_name', var_name='node_lon', " + "attributes={'test': 1}), 'x'), (AuxCoord(array([0, 0, 1]), " + "standard_name='latitude', units=Unit('unknown'), " + "var_name='node_lat'), 'y')], connectivities=Connectivity(" + "cf_role='edge_node_connectivity', start_index=0), " + "edge_coords_and_axes=[(AuxCoord(array([1. , 1.5, 0.5]), " + "standard_name='longitude', units=Unit('unknown'), " + "var_name='edge_lon'), 'x'), (AuxCoord(array([0. , 0.5, 0.5]), " + "standard_name='latitude', units=Unit('unknown'), " + "var_name='edge_lat'), 'y')], long_name='my_topology_mesh', " + "var_name='mesh', attributes={'notes': 'this is a test'}, " + "node_dimension='NodeDim', edge_dimension='EdgeDim')" + ) + self.assertEqual(expected, self.mesh.__repr__()) + + def test_all_connectivities(self): + expected = ugrid.Mesh1DConnectivities(self.EDGE_NODE) + self.assertEqual(expected, self.mesh.all_connectivities) + + def test_all_coords(self): + expected = ugrid.Mesh1DCoords( + self.NODE_LON, self.NODE_LAT, self.EDGE_LON, self.EDGE_LAT + ) + self.assertEqual(expected, self.mesh.all_coords) + + def test_boundary_node(self): + with self.assertRaises(AttributeError): + _ = self.mesh.boundary_node_connectivity + + def test_cf_role(self): + self.assertEqual("mesh_topology", self.mesh.cf_role) + # Read only. + self.assertRaises(AttributeError, setattr, self.mesh.cf_role, "foo", 1) + + def test_connectivities(self): + # General results. Method intended for inheritance. + positive_kwargs = ( + {"item": self.EDGE_NODE}, + {"item": "long_name"}, + {"long_name": "long_name"}, + {"var_name": "var_name"}, + {"attributes": {"test": 1}}, + {"cf_role": "edge_node_connectivity"}, + ) + + fake_connectivity = tests.mock.Mock( + __class__=ugrid.Connectivity, cf_role="fake" + ) + negative_kwargs = ( + {"item": fake_connectivity}, + {"item": "foo"}, + {"standard_name": "air_temperature"}, + {"long_name": "foo"}, + {"var_name": "foo"}, + {"attributes": {"test": 2}}, + {"cf_role": "foo"}, + ) + + func = self.mesh.connectivities + for kwargs in positive_kwargs: + self.assertEqual( + self.EDGE_NODE, func(**kwargs)["edge_node_connectivity"] + ) + for kwargs in negative_kwargs: + self.assertNotIn("edge_node_connectivity", func(**kwargs)) + + def test_connectivities_locations(self): + # topology_dimension-specific results. Method intended to be overridden. + positive_kwargs = ( + {"contains_node": True}, + {"contains_edge": True}, + {"contains_node": True, "contains_edge": True}, + ) + negative_kwargs = ( + {"contains_node": False}, + {"contains_edge": False}, + {"contains_edge": True, "contains_node": False}, + {"contains_edge": False, "contains_node": False}, + ) + + expected = {self.EDGE_NODE.cf_role: self.EDGE_NODE} + func = self.mesh.connectivities + for kwargs in positive_kwargs: + self.assertEqual(expected, func(**kwargs)) + for kwargs in negative_kwargs: + self.assertEqual({}, func(**kwargs)) + + with self.assertLogs(ugrid.logger, level="DEBUG") as log: + self.assertEqual({}, func(contains_face=True)) + self.assertIn("filter for non-existent", log.output[0]) + + def test_coord(self): + # See Mesh.coords tests for thorough coverage of cases. + func = self.mesh.coord + exception = CoordinateNotFoundError + self.assertRaisesRegex( + exception, ".*but found 2", func, include_nodes=True + ) + self.assertRaisesRegex(exception, ".*but found none", func, axis="t") + + def test_coords(self): + # General results. Method intended for inheritance. + positive_kwargs = ( + {"item": self.NODE_LON}, + {"item": "longitude"}, + {"standard_name": "longitude"}, + {"long_name": "long_name"}, + {"var_name": "node_lon"}, + {"attributes": {"test": 1}}, + ) + + fake_coord = AuxCoord([0]) + negative_kwargs = ( + {"item": fake_coord}, + {"item": "foo"}, + {"standard_name": "air_temperature"}, + {"long_name": "foo"}, + {"var_name": "foo"}, + {"attributes": {"test": 2}}, + ) + + func = self.mesh.coords + for kwargs in positive_kwargs: + self.assertEqual(self.NODE_LON, func(**kwargs)["node_x"]) + for kwargs in negative_kwargs: + self.assertNotIn("node_x", func(**kwargs)) + + def test_coords_locations(self): + # topology_dimension-specific results. Method intended to be overridden. + all_expected = { + "node_x": self.NODE_LON, + "node_y": self.NODE_LAT, + "edge_x": self.EDGE_LON, + "edge_y": self.EDGE_LAT, + } + + kwargs_expected = ( + ({"axis": "x"}, ["node_x", "edge_x"]), + ({"axis": "y"}, ["node_y", "edge_y"]), + ({"include_nodes": True}, ["node_x", "node_y"]), + ({"include_edges": True}, ["edge_x", "edge_y"]), + ({"include_nodes": False}, ["edge_x", "edge_y"]), + ({"include_edges": False}, ["node_x", "node_y"]), + ( + {"include_nodes": True, "include_edges": True}, + ["node_x", "node_y", "edge_x", "edge_y"], + ), + ({"include_nodes": False, "include_edges": False}, []), + ( + {"include_nodes": False, "include_edges": True}, + ["edge_x", "edge_y"], + ), + ) + + func = self.mesh.coords + for kwargs, expected in kwargs_expected: + expected = { + k: all_expected[k] for k in expected if k in all_expected + } + self.assertEqual(expected, func(**kwargs)) + + with self.assertLogs(ugrid.logger, level="DEBUG") as log: + self.assertEqual({}, func(include_faces=True)) + self.assertIn("filter non-existent", log.output[0]) + + def test_edge_dimension(self): + self.assertEqual( + self.kwargs["edge_dimension"], self.mesh.edge_dimension + ) + + def test_edge_coords(self): + expected = ugrid.MeshEdgeCoords(self.EDGE_LON, self.EDGE_LAT) + self.assertEqual(expected, self.mesh.edge_coords) + + def test_edge_face(self): + with self.assertRaises(AttributeError): + _ = self.mesh.edge_face_connectivity + + def test_edge_node(self): + self.assertEqual(self.EDGE_NODE, self.mesh.edge_node_connectivity) + + def test_face_coords(self): + with self.assertRaises(AttributeError): + _ = self.mesh.face_coords + + def test_face_dimension(self): + self.assertIsNone(self.mesh.face_dimension) + + def test_face_edge(self): + with self.assertRaises(AttributeError): + _ = self.mesh.face_edge_connectivity + + def test_face_face(self): + with self.assertRaises(AttributeError): + _ = self.mesh.face_face_connectivity + + def test_face_node(self): + with self.assertRaises(AttributeError): + _ = self.mesh.face_node_connectivity + + def test_node_coords(self): + expected = ugrid.MeshNodeCoords(self.NODE_LON, self.NODE_LAT) + self.assertEqual(expected, self.mesh.node_coords) + + def test_node_dimension(self): + self.assertEqual( + self.kwargs["node_dimension"], self.mesh.node_dimension + ) + + def test_topology_dimension(self): + self.assertEqual( + self.kwargs["topology_dimension"], self.mesh.topology_dimension + ) + # Read only. + self.assertRaises( + AttributeError, setattr, self.mesh.topology_dimension, "foo", 1 + ) + + +class TestProperties2D(TestProperties1D): + # Additional/specialised tests for topology_dimension=2. + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.kwargs["topology_dimension"] = 2 + cls.kwargs["connectivities"] = ( + cls.FACE_NODE, + cls.EDGE_NODE, + cls.FACE_EDGE, + cls.FACE_FACE, + cls.EDGE_FACE, + cls.BOUNDARY_NODE, + ) + cls.kwargs["face_dimension"] = "FaceDim" + cls.kwargs["face_coords_and_axes"] = ( + (cls.FACE_LON, "x"), + (cls.FACE_LAT, "y"), + ) + cls.mesh = ugrid.Mesh(**cls.kwargs) + + def test___repr__(self): + expected = ( + "Mesh(topology_dimension=2, node_coords_and_axes=[(AuxCoord(" + "array([0, 2, 1]), standard_name='longitude', units=Unit(" + "'unknown'), long_name='long_name', var_name='node_lon', " + "attributes={'test': 1}), 'x'), (AuxCoord(array([0, 0, 1]), " + "standard_name='latitude', units=Unit('unknown'), " + "var_name='node_lat'), 'y')], connectivities=[Connectivity(" + "cf_role='face_node_connectivity', start_index=0), Connectivity(" + "cf_role='edge_node_connectivity', start_index=0), Connectivity(" + "cf_role='face_edge_connectivity', start_index=0), Connectivity(" + "cf_role='face_face_connectivity', start_index=0), Connectivity(" + "cf_role='edge_face_connectivity', start_index=0), Connectivity(" + "cf_role='boundary_node_connectivity', start_index=0)], " + "edge_coords_and_axes=[(AuxCoord(array([1. , 1.5, 0.5]), " + "standard_name='longitude', units=Unit('unknown'), " + "var_name='edge_lon'), 'x'), (AuxCoord(array([0. , 0.5, 0.5]), " + "standard_name='latitude', units=Unit('unknown'), " + "var_name='edge_lat'), 'y')], face_coords_and_axes=[(AuxCoord(" + "array([0.5]), standard_name='longitude', units=Unit('unknown'), " + "var_name='face_lon'), 'x'), (AuxCoord(array([0.5]), " + "standard_name='latitude', units=Unit('unknown'), " + "var_name='face_lat'), 'y')], long_name='my_topology_mesh', " + "var_name='mesh', attributes={'notes': 'this is a test'}, " + "node_dimension='NodeDim', edge_dimension='EdgeDim', " + "face_dimension='FaceDim')" + ) + self.assertEqual(expected, self.mesh.__repr__()) + + def test_all_connectivities(self): + expected = ugrid.Mesh2DConnectivities( + self.FACE_NODE, + self.EDGE_NODE, + self.FACE_EDGE, + self.FACE_FACE, + self.EDGE_FACE, + self.BOUNDARY_NODE, + ) + self.assertEqual(expected, self.mesh.all_connectivities) + + def test_all_coords(self): + expected = ugrid.Mesh2DCoords( + self.NODE_LON, + self.NODE_LAT, + self.EDGE_LON, + self.EDGE_LAT, + self.FACE_LON, + self.FACE_LAT, + ) + self.assertEqual(expected, self.mesh.all_coords) + + def test_boundary_node(self): + self.assertEqual( + self.BOUNDARY_NODE, self.mesh.boundary_node_connectivity + ) + + def test_connectivity(self): + # See Mesh.connectivities tests for thorough coverage of cases. + # Can only test Mesh.connectivity for 2D since we need >1 connectivity. + func = self.mesh.connectivity + exception = ConnectivityNotFoundError + self.assertRaisesRegex( + exception, ".*but found 3", func, contains_node=True + ) + self.assertRaisesRegex( + exception, + ".*but found none", + func, + contains_node=False, + contains_edge=False, + contains_face=False, + ) + + def test_connectivities_locations(self): + kwargs_expected = ( + ( + {"contains_node": True}, + [self.EDGE_NODE, self.FACE_NODE, self.BOUNDARY_NODE], + ), + ( + {"contains_edge": True}, + [self.EDGE_NODE, self.FACE_EDGE, self.EDGE_FACE], + ), + ( + {"contains_face": True}, + [ + self.FACE_NODE, + self.FACE_EDGE, + self.FACE_FACE, + self.EDGE_FACE, + ], + ), + ( + {"contains_node": False}, + [self.FACE_EDGE, self.EDGE_FACE, self.FACE_FACE], + ), + ( + {"contains_edge": False}, + [self.FACE_NODE, self.BOUNDARY_NODE, self.FACE_FACE], + ), + ({"contains_face": False}, [self.EDGE_NODE, self.BOUNDARY_NODE]), + ( + {"contains_edge": True, "contains_face": True}, + [self.FACE_EDGE, self.EDGE_FACE], + ), + ( + {"contains_node": False, "contains_edge": False}, + [self.FACE_FACE], + ), + ( + {"contains_node": True, "contains_edge": False}, + [self.FACE_NODE, self.BOUNDARY_NODE], + ), + ( + { + "contains_node": False, + "contains_edge": False, + "contains_face": False, + }, + [], + ), + ) + func = self.mesh.connectivities + for kwargs, expected in kwargs_expected: + expected = {c.cf_role: c for c in expected} + self.assertEqual(expected, func(**kwargs)) + + def test_coords_locations(self): + all_expected = { + "node_x": self.NODE_LON, + "node_y": self.NODE_LAT, + "edge_x": self.EDGE_LON, + "edge_y": self.EDGE_LAT, + "face_x": self.FACE_LON, + "face_y": self.FACE_LAT, + } + + kwargs_expected = ( + ({"axis": "x"}, ["node_x", "edge_x", "face_x"]), + ({"axis": "y"}, ["node_y", "edge_y", "face_y"]), + ({"include_nodes": True}, ["node_x", "node_y"]), + ({"include_edges": True}, ["edge_x", "edge_y"]), + ( + {"include_nodes": False}, + ["edge_x", "edge_y", "face_x", "face_y"], + ), + ( + {"include_edges": False}, + ["node_x", "node_y", "face_x", "face_y"], + ), + ( + {"include_faces": False}, + ["node_x", "node_y", "edge_x", "edge_y"], + ), + ( + {"include_faces": True, "include_edges": True}, + ["edge_x", "edge_y", "face_x", "face_y"], + ), + ( + {"include_faces": False, "include_edges": False}, + ["node_x", "node_y"], + ), + ( + {"include_faces": False, "include_edges": True}, + ["edge_x", "edge_y"], + ), + ) + + func = self.mesh.coords + for kwargs, expected in kwargs_expected: + expected = { + k: all_expected[k] for k in expected if k in all_expected + } + self.assertEqual(expected, func(**kwargs)) + + def test_edge_face(self): + self.assertEqual(self.EDGE_FACE, self.mesh.edge_face_connectivity) + + def test_face_coords(self): + expected = ugrid.MeshFaceCoords(self.FACE_LON, self.FACE_LAT) + self.assertEqual(expected, self.mesh.face_coords) + + def test_face_dimension(self): + self.assertEqual( + self.kwargs["face_dimension"], self.mesh.face_dimension + ) + + def test_face_edge(self): + self.assertEqual(self.FACE_EDGE, self.mesh.face_edge_connectivity) + + def test_face_face(self): + self.assertEqual(self.FACE_FACE, self.mesh.face_face_connectivity) + + def test_face_node(self): + self.assertEqual(self.FACE_NODE, self.mesh.face_node_connectivity) + + +class TestOperations1D(TestMeshCommon): + # Tests that cannot re-use an existing Mesh instance, instead need a new + # one each time. + def setUp(self): + self.mesh = ugrid.Mesh( + topology_dimension=1, + node_coords_and_axes=((self.NODE_LON, "x"), (self.NODE_LAT, "y")), + connectivities=self.EDGE_NODE, + ) + + @staticmethod + def new_connectivity(connectivity, new_len=False): + """Provide a new connectivity recognisably different from the original.""" + # NOTE: assumes non-transposed connectivity (src_dim=0). + if new_len: + shape = (connectivity.shape[0] + 1, connectivity.shape[1]) + else: + shape = connectivity.shape + return connectivity.copy(np.zeros(shape, dtype=int)) + + @staticmethod + def new_coord(coord, new_shape=False): + """Provide a new coordinate recognisably different from the original.""" + if new_shape: + shape = tuple([i + 1 for i in coord.shape]) + else: + shape = coord.shape + return coord.copy(np.zeros(shape)) + + def test___setstate__(self): + false_metadata_manager = "foo" + false_coord_manager = "bar" + false_connectivity_manager = "baz" + self.mesh.__setstate__( + ( + false_metadata_manager, + false_coord_manager, + false_connectivity_manager, + ) + ) + + self.assertEqual(false_metadata_manager, self.mesh._metadata_manager) + self.assertEqual(false_coord_manager, self.mesh._coord_manager) + self.assertEqual( + false_connectivity_manager, self.mesh._connectivity_manager + ) + + def test_add_connectivities(self): + # Cannot test ADD - 1D - nothing extra to add beyond minimum. + + for new_len in (False, True): + # REPLACE connectivities, first with one of the same length, then + # with one of different length. + edge_node = self.new_connectivity(self.EDGE_NODE, new_len) + self.mesh.add_connectivities(edge_node) + self.assertEqual( + ugrid.Mesh1DConnectivities(edge_node), + self.mesh.all_connectivities, + ) + + def test_add_connectivities_duplicates(self): + edge_node_one = self.EDGE_NODE + edge_node_two = self.new_connectivity(self.EDGE_NODE) + self.mesh.add_connectivities(edge_node_one, edge_node_two) + self.assertEqual( + edge_node_two, + self.mesh.edge_node_connectivity, + ) + + def test_add_connectivities_invalid(self): + self.assertRaisesRegex( + TypeError, + "Expected Connectivity.*", + self.mesh.add_connectivities, + "foo", + ) + + face_node = self.FACE_NODE + with self.assertLogs(ugrid.logger, level="DEBUG") as log: + self.mesh.add_connectivities(face_node) + self.assertIn("Not adding connectivity", log.output[0]) + + def test_add_coords(self): + # ADD coords. + edge_kwargs = {"edge_x": self.EDGE_LON, "edge_y": self.EDGE_LAT} + self.mesh.add_coords(**edge_kwargs) + self.assertEqual( + ugrid.MeshEdgeCoords(**edge_kwargs), self.mesh.edge_coords + ) + + for new_shape in (False, True): + # REPLACE coords, first with ones of the same shape, then with ones + # of different shape. + node_kwargs = { + "node_x": self.new_coord(self.NODE_LON, new_shape), + "node_y": self.new_coord(self.NODE_LAT, new_shape), + } + edge_kwargs = { + "edge_x": self.new_coord(self.EDGE_LON, new_shape), + "edge_y": self.new_coord(self.EDGE_LAT, new_shape), + } + self.mesh.add_coords(**node_kwargs, **edge_kwargs) + self.assertEqual( + ugrid.MeshNodeCoords(**node_kwargs), self.mesh.node_coords + ) + self.assertEqual( + ugrid.MeshEdgeCoords(**edge_kwargs), self.mesh.edge_coords + ) + + def test_add_coords_face(self): + self.assertRaises( + TypeError, + self.mesh.add_coords, + face_x=self.FACE_LON, + face_y=self.FACE_LAT, + ) + + def test_add_coords_invalid(self): + func = self.mesh.add_coords + self.assertRaisesRegex( + TypeError, ".*requires to be an 'AuxCoord'.*", func, node_x="foo" + ) + self.assertRaisesRegex( + TypeError, ".*requires a x-axis like.*", func, node_x=self.NODE_LAT + ) + climatological = AuxCoord( + [0], + bounds=[-1, 1], + standard_name="longitude", + climatological=True, + units="Days since 1970", + ) + self.assertRaisesRegex( + TypeError, + ".*cannot be a climatological.*", + func, + node_x=climatological, + ) + wrong_shape = self.NODE_LON.copy([0]) + self.assertRaisesRegex( + ValueError, ".*requires to have shape.*", func, node_x=wrong_shape + ) + + def test_add_coords_single(self): + # ADD coord. + edge_x = self.EDGE_LON + expected = ugrid.MeshEdgeCoords(edge_x=edge_x, edge_y=None) + self.mesh.add_coords(edge_x=edge_x) + self.assertEqual(expected, self.mesh.edge_coords) + + # REPLACE coords. + node_x = self.new_coord(self.NODE_LON) + edge_x = self.new_coord(self.EDGE_LON) + expected_nodes = ugrid.MeshNodeCoords( + node_x=node_x, node_y=self.mesh.node_coords.node_y + ) + expected_edges = ugrid.MeshEdgeCoords(edge_x=edge_x, edge_y=None) + self.mesh.add_coords(node_x=node_x, edge_x=edge_x) + self.assertEqual(expected_nodes, self.mesh.node_coords) + self.assertEqual(expected_edges, self.mesh.edge_coords) + + # Attempt to REPLACE coords with those of DIFFERENT SHAPE. + node_x = self.new_coord(self.NODE_LON, new_shape=True) + edge_x = self.new_coord(self.EDGE_LON, new_shape=True) + node_kwarg = {"node_x": node_x} + edge_kwarg = {"edge_x": edge_x} + both_kwargs = dict(**node_kwarg, **edge_kwarg) + for kwargs in (node_kwarg, edge_kwarg, both_kwargs): + self.assertRaisesRegex( + ValueError, + ".*requires to have shape.*", + self.mesh.add_coords, + **kwargs, + ) + + def test_add_coords_single_face(self): + self.assertRaises( + TypeError, self.mesh.add_coords, face_x=self.FACE_LON + ) + + def test_dimension_names(self): + # Test defaults. + default = ugrid.Mesh1DNames("Mesh1d_node", "Mesh1d_edge") + self.assertEqual(default, self.mesh.dimension_names()) + + with self.assertLogs(ugrid.logger, level="DEBUG") as log: + self.mesh.dimension_names("foo", "bar", "baz") + self.assertIn("Not setting face_dimension", log.output[0]) + self.assertEqual( + ugrid.Mesh1DNames("foo", "bar"), self.mesh.dimension_names() + ) + + self.mesh.dimension_names_reset(True, True, True) + self.assertEqual(default, self.mesh.dimension_names()) + + # Single. + self.mesh.dimension_names(edge="foo") + self.assertEqual("foo", self.mesh.edge_dimension) + self.mesh.dimension_names_reset(edge=True) + self.assertEqual(default, self.mesh.dimension_names()) + + def test_edge_dimension_set(self): + self.mesh.edge_dimension = "foo" + self.assertEqual("foo", self.mesh.edge_dimension) + + def test_face_dimension_set(self): + with self.assertLogs(ugrid.logger, level="DEBUG") as log: + self.mesh.face_dimension = "foo" + self.assertIn("Not setting face_dimension", log.output[0]) + self.assertIsNone(self.mesh.face_dimension) + + def test_node_dimension_set(self): + self.mesh.node_dimension = "foo" + self.assertEqual("foo", self.mesh.node_dimension) + + def test_remove_connectivities(self): + """ + Test that remove() mimics the connectivities() method correctly, + and prevents removal of mandatory connectivities. + + """ + positive_kwargs = ( + {"item": self.EDGE_NODE}, + {"item": "long_name"}, + {"long_name": "long_name"}, + {"var_name": "var_name"}, + {"attributes": {"test": 1}}, + {"cf_role": "edge_node_connectivity"}, + {"contains_node": True}, + {"contains_edge": True}, + {"contains_edge": True, "contains_node": True}, + ) + + fake_connectivity = tests.mock.Mock( + __class__=ugrid.Connectivity, cf_role="fake" + ) + negative_kwargs = ( + {"item": fake_connectivity}, + {"item": "foo"}, + {"standard_name": "air_temperature"}, + {"long_name": "foo"}, + {"var_name": "foo"}, + {"attributes": {"test": 2}}, + {"cf_role": "foo"}, + {"contains_node": False}, + {"contains_edge": False}, + {"contains_edge": True, "contains_node": False}, + {"contains_edge": False, "contains_node": False}, + ) + + for kwargs in positive_kwargs: + with self.assertLogs(ugrid.logger, level="DEBUG") as log: + self.mesh.remove_connectivities(**kwargs) + self.assertIn("Ignoring request to remove", log.output[0]) + self.assertEqual(self.EDGE_NODE, self.mesh.edge_node_connectivity) + for kwargs in negative_kwargs: + with self.assertLogs(ugrid.logger, level="DEBUG") as log: + # Check that the only debug log is the one we inserted. + ugrid.logger.debug("foo") + self.mesh.remove_connectivities(**kwargs) + self.assertEqual(1, len(log.output)) + self.assertEqual(self.EDGE_NODE, self.mesh.edge_node_connectivity) + + def test_remove_coords(self): + # Test that remove() mimics the coords() method correctly, + # and prevents removal of mandatory coords. + positive_kwargs = ( + {"item": self.NODE_LON}, + {"item": "longitude"}, + {"standard_name": "longitude"}, + {"long_name": "long_name"}, + {"var_name": "node_lon"}, + {"attributes": {"test": 1}}, + ) + + fake_coord = AuxCoord([0]) + negative_kwargs = ( + {"item": fake_coord}, + {"item": "foo"}, + {"standard_name": "air_temperature"}, + {"long_name": "foo"}, + {"var_name": "foo"}, + {"attributes": {"test": 2}}, + ) + + for kwargs in positive_kwargs: + with self.assertLogs(ugrid.logger, level="DEBUG") as log: + self.mesh.remove_coords(**kwargs) + self.assertIn("Ignoring request to remove", log.output[0]) + self.assertEqual(self.NODE_LON, self.mesh.node_coords.node_x) + for kwargs in negative_kwargs: + with self.assertLogs(ugrid.logger, level="DEBUG") as log: + # Check that the only debug log is the one we inserted. + ugrid.logger.debug("foo") + self.mesh.remove_coords(**kwargs) + self.assertEqual(1, len(log.output)) + self.assertEqual(self.NODE_LON, self.mesh.node_coords.node_x) + + # Test removal of optional connectivity. + self.mesh.add_coords(edge_x=self.EDGE_LON) + # Attempt to remove a non-existent coord. + self.mesh.remove_coords(self.EDGE_LAT) + # Confirm that EDGE_LON is still there. + self.assertEqual(self.EDGE_LON, self.mesh.edge_coords.edge_x) + # Remove EDGE_LON and confirm success. + self.mesh.remove_coords(self.EDGE_LON) + self.assertEqual(None, self.mesh.edge_coords.edge_x) + + +class TestOperations2D(TestOperations1D): + # Additional/specialised tests for topology_dimension=2. + def setUp(self): + self.mesh = ugrid.Mesh( + topology_dimension=2, + node_coords_and_axes=((self.NODE_LON, "x"), (self.NODE_LAT, "y")), + connectivities=(self.FACE_NODE), + ) + + def test_add_connectivities(self): + # ADD connectivities. + kwargs = { + "edge_node": self.EDGE_NODE, + "face_edge": self.FACE_EDGE, + "face_face": self.FACE_FACE, + "edge_face": self.EDGE_FACE, + "boundary_node": self.BOUNDARY_NODE, + } + expected = ugrid.Mesh2DConnectivities( + face_node=self.mesh.face_node_connectivity, **kwargs + ) + self.mesh.add_connectivities(*kwargs.values()) + self.assertEqual(expected, self.mesh.all_connectivities) + + # REPLACE connectivities. + kwargs["face_node"] = self.FACE_NODE + for new_len in (False, True): + # First replace with ones of same length, then with ones of + # different length. + kwargs = { + k: self.new_connectivity(v, new_len) for k, v in kwargs.items() + } + self.mesh.add_connectivities(*kwargs.values()) + self.assertEqual( + ugrid.Mesh2DConnectivities(**kwargs), + self.mesh.all_connectivities, + ) + + def test_add_connectivities_inconsistent(self): + # ADD Connectivities. + self.mesh.add_connectivities(self.EDGE_NODE) + face_edge = self.new_connectivity(self.FACE_EDGE, new_len=True) + edge_face = self.new_connectivity(self.EDGE_FACE, new_len=True) + for args in ([face_edge], [edge_face], [face_edge, edge_face]): + self.assertRaisesRegex( + ValueError, + "inconsistent .* counts.", + self.mesh.add_connectivities, + *args, + ) + + # REPLACE Connectivities + self.mesh.add_connectivities(self.FACE_EDGE, self.EDGE_FACE) + for args in ([face_edge], [edge_face], [face_edge, edge_face]): + self.assertRaisesRegex( + ValueError, + "inconsistent .* counts.", + self.mesh.add_connectivities, + *args, + ) + + def test_add_connectivities_invalid(self): + fake_cf_role = tests.mock.Mock( + __class__=ugrid.Connectivity, cf_role="foo" + ) + with self.assertLogs(ugrid.logger, level="DEBUG") as log: + self.mesh.add_connectivities(fake_cf_role) + self.assertIn("Not adding connectivity", log.output[0]) + + def test_add_coords_face(self): + # ADD coords. + kwargs = {"face_x": self.FACE_LON, "face_y": self.FACE_LAT} + self.mesh.add_coords(**kwargs) + self.assertEqual(ugrid.MeshFaceCoords(**kwargs), self.mesh.face_coords) + + for new_shape in (False, True): + # REPLACE coords, first with ones of the same shape, then with ones + # of different shape. + kwargs = { + "face_x": self.new_coord(self.FACE_LON, new_shape), + "face_y": self.new_coord(self.FACE_LAT, new_shape), + } + self.mesh.add_coords(**kwargs) + self.assertEqual( + ugrid.MeshFaceCoords(**kwargs), self.mesh.face_coords + ) + + def test_add_coords_single_face(self): + # ADD coord. + face_x = self.FACE_LON + expected = ugrid.MeshFaceCoords(face_x=face_x, face_y=None) + self.mesh.add_coords(face_x=face_x) + self.assertEqual(expected, self.mesh.face_coords) + + # REPLACE coord. + face_x = self.new_coord(self.FACE_LON) + expected = ugrid.MeshFaceCoords(face_x=face_x, face_y=None) + self.mesh.add_coords(face_x=face_x) + self.assertEqual(expected, self.mesh.face_coords) + + # Attempt to REPLACE coord with that of DIFFERENT SHAPE. + face_x = self.new_coord(self.FACE_LON, new_shape=True) + self.assertRaisesRegex( + ValueError, + ".*requires to have shape.*", + self.mesh.add_coords, + face_x=face_x, + ) + + def test_dimension_names(self): + # Test defaults. + default = ugrid.Mesh2DNames( + "Mesh2d_node", "Mesh2d_edge", "Mesh2d_face" + ) + self.assertEqual(default, self.mesh.dimension_names()) + + self.mesh.dimension_names("foo", "bar", "baz") + self.assertEqual( + ugrid.Mesh2DNames("foo", "bar", "baz"), self.mesh.dimension_names() + ) + + self.mesh.dimension_names_reset(True, True, True) + self.assertEqual(default, self.mesh.dimension_names()) + + # Single. + self.mesh.dimension_names(face="foo") + self.assertEqual("foo", self.mesh.face_dimension) + self.mesh.dimension_names_reset(face=True) + self.assertEqual(default, self.mesh.dimension_names()) + + def test_face_dimension_set(self): + self.mesh.face_dimension = "foo" + self.assertEqual("foo", self.mesh.face_dimension) + + def test_remove_connectivities(self): + """Do what 1D test could not - test removal of optional connectivity.""" + + # Add an optional connectivity. + self.mesh.add_connectivities(self.FACE_FACE) + # Attempt to remove a non-existent connectivity. + self.mesh.remove_connectivities(self.EDGE_NODE) + # Confirm that FACE_FACE is still there. + self.assertEqual(self.FACE_FACE, self.mesh.face_face_connectivity) + # Remove FACE_FACE and confirm success. + self.mesh.remove_connectivities(contains_face=True) + self.assertEqual(None, self.mesh.face_face_connectivity) + + def test_remove_coords(self): + """Test the face argument.""" + super().test_remove_coords() + self.mesh.add_coords(face_x=self.FACE_LON) + self.assertEqual(self.FACE_LON, self.mesh.face_coords.face_x) + self.mesh.remove_coords(include_faces=True) + self.assertEqual(None, self.mesh.face_coords.face_x) + + +class InitValidation(TestMeshCommon): + def test_invalid_topology(self): + kwargs = { + "topology_dimension": 0, + "node_coords_and_axes": ( + (self.NODE_LON, "x"), + (self.NODE_LAT, "y"), + ), + "connectivities": self.EDGE_NODE, + } + self.assertRaisesRegex( + ValueError, "Expected 'topology_dimension'.*", ugrid.Mesh, **kwargs + ) + + def test_invalid_axes(self): + kwargs = { + "topology_dimension": 2, + "connectivities": self.FACE_NODE, + } + self.assertRaisesRegex( + ValueError, + "Invalid axis specified for node.*", + ugrid.Mesh, + node_coords_and_axes=( + (self.NODE_LON, "foo"), + (self.NODE_LAT, "y"), + ), + **kwargs, + ) + kwargs["node_coords_and_axes"] = ( + (self.NODE_LON, "x"), + (self.NODE_LAT, "y"), + ) + self.assertRaisesRegex( + ValueError, + "Invalid axis specified for edge.*", + ugrid.Mesh, + edge_coords_and_axes=((self.EDGE_LON, "foo"),), + **kwargs, + ) + self.assertRaisesRegex( + ValueError, + "Invalid axis specified for face.*", + ugrid.Mesh, + face_coords_and_axes=((self.FACE_LON, "foo"),), + **kwargs, + ) + + # Several arg safety checks in __init__ currently unreachable given earlier checks. + + def test_minimum_connectivities(self): + # Further validations are tested in add_connectivity tests. + kwargs = { + "topology_dimension": 1, + "node_coords_and_axes": ( + (self.NODE_LON, "x"), + (self.NODE_LAT, "y"), + ), + "connectivities": (self.FACE_NODE,), + } + self.assertRaisesRegex( + ValueError, + ".*requires a edge_node_connectivity.*", + ugrid.Mesh, + **kwargs, + ) + + def test_minimum_coords(self): + # Further validations are tested in add_coord tests. + kwargs = { + "topology_dimension": 1, + "node_coords_and_axes": ((self.NODE_LON, "x"), (None, "y")), + "connectivities": (self.FACE_NODE,), + } + self.assertRaisesRegex( + ValueError, ".*is a required coordinate.*", ugrid.Mesh, **kwargs + )