How to use the eccodes.BufrFile function in eccodes

To help you get started, we’ve selected a few eccodes examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github ecmwf / eccodes / examples / python / high_level_api.py View on Github external
def test_memory_management(self):
        """Messages in BufrFile can be opened and closed properly."""
        with BufrFile(TESTBUFR) as bufr_file:
            self.assertEqual(len(bufr_file), 3)
            for i in range(len(bufr_file)):
                msg = BufrMessage(bufr_file)
                self.assertEqual(msg["bufrHeaderCentre"], 98)
                self.assertEqual(msg['count'], i + 1)
            self.assertEqual(len(bufr_file.open_messages), 3)
        self.assertEqual(len(bufr_file.open_messages), 0)
github GribApiDotNet / GribApi.XP / grib_api / examples / python / high_level_api.py View on Github external
def test_metadata(self):
        """Metadata is read correctly from BufrMessage."""
        with BufrFile(TESTBUFR) as bufr_file:
            msg = BufrMessage(bufr_file)
            msg.unpack()
            msg_keys = msg.keys()
            self.assertEqual(len(msg_keys), 140)
            for key in KNOWN_BUFR_KEYS:
                assert key in msg_keys
            # Size of message in bytes
            self.assertEqual(msg.size(), 220)
            self.assertEqual(len(msg.keys()), len(msg))
github GribApiDotNet / GribApi.XP / grib_api / examples / python / high_level_api.py View on Github external
def test_clone(self):
        """Messages can be used to produce clone Messages."""
        with BufrFile(TESTBUFR) as bufr_file:
            msg = BufrMessage(bufr_file)
            msg2 = BufrMessage(clone=msg)
            self.assertSequenceEqual(msg.keys(), msg2.keys())
github ecmwf / eccodes / examples / python / high_level_api.py View on Github external
def test_read_invalid_file(self):
        """Trying to open message on nonexistent file raises IOError."""
        with NamedTemporaryFile(mode='r') as f:
            with BufrFile(f.name) as bufr_file:
                self.assertRaises(IOError, lambda: BufrMessage(bufr_file))
github ecmwf / eccodes / examples / python / high_level_api.py View on Github external
def test_value_setting(self):
        """Keys can be set properly."""
        with BufrFile(TESTBUFR) as bufr_file:
            msg = BufrMessage(bufr_file)
            key, val = "localLongitude", 5
            msg[key] = val
            self.assertEqual(msg[key], val)
github GribApiDotNet / GribApi.XP / grib_api / examples / python / high_level_api.py View on Github external
def test_iterator_protocol(self):
        """The BufrFile allows pythonic iteration over all messages."""
        latitudes = []
        with BufrFile(TESTBUFR) as bufr_file:
            for msg in bufr_file:
                latitudes.append(msg["localLatitude"])
        self.assertSequenceEqual(latitudes, [70.93, 77, 78.92])
github GribApiDotNet / GribApi.XP / grib_api / examples / python / high_level_api.py View on Github external
def test_read_past_last_message(self):
        """Trying to open message on exhausted BUFR file raises IOError."""
        with BufrFile(TESTBUFR) as bufr_file:
            for _ in range(len(bufr_file)):
                BufrMessage(bufr_file)
            self.assertRaises(IOError, lambda: BufrMessage(bufr_file))
github ecmwf / eccodes / examples / python / high_level_api.py View on Github external
def test_read_past_last_message(self):
        """Trying to open message on exhausted BUFR file raises IOError."""
        with BufrFile(TESTBUFR) as bufr_file:
            for _ in range(len(bufr_file)):
                BufrMessage(bufr_file)
            self.assertRaises(IOError, lambda: BufrMessage(bufr_file))
github GribApiDotNet / GribApi.XP / grib_api / examples / python / high_level_api.py View on Github external
def test_value_setting(self):
        """Keys can be set properly."""
        with BufrFile(TESTBUFR) as bufr_file:
            msg = BufrMessage(bufr_file)
            key, val = "localLongitude", 5
            msg[key] = val
            self.assertEqual(msg[key], val)
github ecmwf / eccodes / examples / python / high_level_api.py View on Github external
def test_serialize(self):
        """Message can be serialized to file."""
        with BufrFile(TESTBUFR) as bufr_file:
            msg = BufrMessage(bufr_file)
            with open(TEST_OUTPUT, "w") as test:
                msg.write(test)
        os.unlink(TEST_OUTPUT)