diff options
author | 2023-11-25 21:04:48 +0100 | |
---|---|---|
committer | 2023-12-10 10:23:54 -0500 | |
commit | 26dad5cf4685a3f59b2c5569b102b5e22bdddcb6 (patch) | |
tree | b50bea298eb527d8f1816305c7a5d691bc1b6834 | |
parent | pmt: Disallow None for pmt_t arguments in Python (diff) | |
download | gnuradio-26dad5cf4685a3f59b2c5569b102b5e22bdddcb6.tar.xz gnuradio-26dad5cf4685a3f59b2c5569b102b5e22bdddcb6.zip |
blocks: msg_pair_to_var: Log with block logger, not gr.log (#6950)
* blocks: msg_pair_to_var: Log with block logger, not gr.log
This also adds the missing unit test.
it seems that 0xdeadbeef is not `pmt.from_long`able on windows/conda CI
Signed-off-by: Marcus Müller <mmueller@gnuradio.org>
(cherry picked from commit d7f887227add25431b948d3a18471efd56bb1945)
Signed-off-by: Jeff Long <willcode4@gmail.com>
-rw-r--r-- | gr-blocks/python/blocks/msg_pair_to_var.py | 14 | ||||
-rwxr-xr-x | gr-blocks/python/blocks/qa_msg_pair_to_var.py | 81 |
2 files changed, 89 insertions, 6 deletions
diff --git a/gr-blocks/python/blocks/msg_pair_to_var.py b/gr-blocks/python/blocks/msg_pair_to_var.py index e282af058..d01feca75 100644 --- a/gr-blocks/python/blocks/msg_pair_to_var.py +++ b/gr-blocks/python/blocks/msg_pair_to_var.py @@ -20,27 +20,29 @@ class msg_pair_to_var(gr.sync_block): unpredictable. """ + IN_PORT = pmt.intern("inpair") + def __init__(self, callback): gr.sync_block.__init__( self, name="msg_pair_to_var", in_sig=None, out_sig=None) self.callback = callback - self.message_port_register_in(pmt.intern("inpair")) - self.set_msg_handler(pmt.intern("inpair"), self.msg_handler) + self.message_port_register_in(self.IN_PORT) + self.set_msg_handler(self.IN_PORT, self.msg_handler) def msg_handler(self, msg): if not pmt.is_pair(msg) or pmt.is_dict(msg) or pmt.is_pdu(msg): - gr.log.warn( - "Input message %s is not a simple pair, dropping" % repr(msg)) + self.logger.warn( + f"Input message {msg} is not a simple pair, dropping") return new_val = pmt.to_python(pmt.cdr(msg)) try: self.callback(new_val) except Exception as e: - gr.log.error("Error when calling " + repr(self.callback) + " with " + - repr(new_val) + " (reason: %s)" % repr(e)) + self.logger.error( + f"Error when calling {self.callback} with {new_val} (reason: {e})") def stop(self): return True diff --git a/gr-blocks/python/blocks/qa_msg_pair_to_var.py b/gr-blocks/python/blocks/qa_msg_pair_to_var.py new file mode 100755 index 000000000..605160f31 --- /dev/null +++ b/gr-blocks/python/blocks/qa_msg_pair_to_var.py @@ -0,0 +1,81 @@ +#!/usr/bin/env python +# +# Copyright 2023 Marcus Müller +# +# This file is part of GNU Radio +# +# SPDX-License-Identifier: GPL-3.0-or-later +# +# + + +from gnuradio import blocks +from gnuradio import gr_unittest +from gnuradio import gr +import pmt +import sys +import time + + +class test_msg_pair_to_var(gr_unittest.TestCase): + + def setUp(self): + self.backend = gr.dictionary_logger_backend() + gr.logging().add_default_sink(self.backend) + self.tb = gr.top_block(catch_exceptions=False) + + def tearDown(self): + self.tb = None + + def test_calling(self): + canary = 0xDEAFBEEF + self.test_val = 0 + + def test_f(value: int = 0): + self.test_val = value + + dut = blocks.msg_pair_to_var(test_f) + test_msg = pmt.cons(pmt.intern("foo"), pmt.from_uint64(canary)) + src = blocks.message_strobe(test_msg, 10) + self.tb.msg_connect(*(src, pmt.intern("strobe")), + *(dut, blocks.msg_pair_to_var.IN_PORT)) + self.tb.start() + time.sleep(15e-3) + self.tb.stop() + self.tb.wait() + + self.assertEqual(self.test_val, canary, + f"Test value {hex(self.test_val)} not found to be canary {hex(canary)}") + + def test_exceptions(self): + lark = 0xDEAFBEEF + nightingale = 0xDEAFB16D + self.test_val = 0 + + def test_f(value: int = 0): + if value == nightingale and value != lark: + raise Exception("It was the nightingale, and not the lark") + + dut = blocks.msg_pair_to_var(test_f) + test_msg = pmt.cons(pmt.intern("birb"), pmt.from_uint64(nightingale)) + src = blocks.message_strobe(test_msg, 10) + self.tb.msg_connect(*(src, pmt.intern("strobe")), + *(dut, blocks.msg_pair_to_var.IN_PORT)) + self.tb.start() + time.sleep(50e-3) + self.tb.stop() + self.tb.wait() + + logged = self.backend.get_map() + self.assertIn("msg_pair_to_var", logged, "should have logged; didn't.") + log_entries = logged["msg_pair_to_var"] + for timestamp, entry in log_entries: + self.assertRegex( + entry, + r"Error when calling <function .*\.test_exceptions\..*.test_f.*> with 3736056173" + + r".*reason: It was the nightingale, and not the lark.*", + f"Log entry '{entry}' from {timestamp} doesn't contain Shakespeare") + + +if __name__ == '__main__': + gr_unittest.run(test_msg_pair_to_var) |