Skip to content

mbusreader API Documentation

i18n

Created on 2025-01-22

@author: wf

I18n

Simple internationalization class for message handling

Source code in mbusread/i18n.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
@lod_storable
class I18n:
    """Simple internationalization class for message handling"""

    language: str = "en"
    messages: Dict[str, Dict[str, str]] = field(default_factory=dict)

    @classmethod
    def default(cls) -> "I18n":
        yaml_file = MBusConfig.examples_path() + "/i18n.yaml"
        i18n = cls.load_from_yaml_file(yaml_file)
        return i18n

    def get(self, key: str, *args) -> str:
        """Get localized message with optional formatting"""
        if self.language not in self.messages:
            self.language = "en"
        message = self.messages[self.language].get(key, key)
        formatted_message = message.format(*args) if args else message
        return formatted_message

get(key, *args)

Get localized message with optional formatting

Source code in mbusread/i18n.py 
28
29
30
31
32
33
34
def get(self, key: str, *args) -> str:
    """Get localized message with optional formatting"""
    if self.language not in self.messages:
        self.language = "en"
    message = self.messages[self.language].get(key, key)
    formatted_message = message.format(*args) if args else message
    return formatted_message

logger

Created on 2025-01-25

@author: wf

Logger

Logger singleton for M-Bus reader

Source code in mbusread/logger.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
class Logger:
    """Logger singleton for M-Bus reader"""
    _logger = None

    @classmethod
    def setup_logger(cls, debug: bool = False) -> logging.Logger:
        if cls._logger is None:
            cls._logger = logging.getLogger("MBusReader")
            if debug:
                cls._logger.setLevel(logging.DEBUG)
            handler = logging.StreamHandler()
            formatter = logging.Formatter(
                "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
            )
            handler.setFormatter(formatter)
            cls._logger.addHandler(handler)
        return cls._logger

mbus_config

Created on 2025-01-22 M-Bus configuration Classes to be read from configuration files e.g. YAML encoded @author: wf

Device

A device class for M-Bus devices storing manufacturer reference

Note on wakeup timing: The M-Bus standard formula suggests 33 bytes per 300 baud with start+8data+stop = 10 bits. However, we use configured pattern repetitions: - Default is 528 times (0x55) - Ultramaxx needs 1056 times (66 lines * 16 bytes) The total time includes sending these patterns at the given baudrate plus a fixed delay time.

Source code in mbusread/mbus_config.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
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
@lod_storable
class Device:
    """
    A device class for M-Bus devices storing manufacturer reference

    Note on wakeup timing:
    The M-Bus standard formula suggests 33 bytes per 300 baud with
    start+8data+stop = 10 bits. However, we use configured pattern repetitions:
    - Default is 528 times (0x55)
    - Ultramaxx needs 1056 times (66 lines * 16 bytes)
    The total time includes sending these patterns at the given baudrate
    plus a fixed delay time.
    """

    model: str
    title: str = ""
    url: str = ""
    doc_url: str = ""
    has_echo: bool = False
    wakeup_pattern: str = "55"
    wakeup_times: int = 528  # Number of pattern repetitions
    wakeup_delay: float = 0.35  # secs
    messages: Dict[str, MBusMessage] = field(default_factory=dict)

    def wakeup_time(self, baudrate: int = 2400) -> float:
        """Calculate total wakeup time based on pattern repetitions"""
        secs = (self.wakeup_times * len(bytes.fromhex(self.wakeup_pattern))) / (
            baudrate / 10
        ) + self.wakeup_delay
        return secs

    def as_html(self) -> str:
        """Generate HTML representation of the device including wakeup info"""
        title = self.title if self.title else self.model
        device_link = (
            Link.create(url=self.url, text=title, target="_blank")
            if self.doc_url
            else title
        )
        doc_link = (
            Link.create(url=self.doc_url, text="📄", target="_blank")
            if self.doc_url
            else ""
        )
        mfr_html = (
            self.manufacturer.as_html() if hasattr(self, "manufacturer") else self.mid
        )
        wakeup_html = f"""wakeup: {self.wakeup_pattern} = {self.wakeup_times}×0x{self.wakeup_pattern} ({self.wakeup_time(2400):.2f}s incl. {self.wakeup_delay}s delay)"""
        markup = f"""{mfr_html} → {device_link}{doc_link}<br>
{wakeup_html}"""
        return markup

as_html()

Generate HTML representation of the device including wakeup info

Source code in mbusread/mbus_config.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
    def as_html(self) -> str:
        """Generate HTML representation of the device including wakeup info"""
        title = self.title if self.title else self.model
        device_link = (
            Link.create(url=self.url, text=title, target="_blank")
            if self.doc_url
            else title
        )
        doc_link = (
            Link.create(url=self.doc_url, text="📄", target="_blank")
            if self.doc_url
            else ""
        )
        mfr_html = (
            self.manufacturer.as_html() if hasattr(self, "manufacturer") else self.mid
        )
        wakeup_html = f"""wakeup: {self.wakeup_pattern} = {self.wakeup_times}×0x{self.wakeup_pattern} ({self.wakeup_time(2400):.2f}s incl. {self.wakeup_delay}s delay)"""
        markup = f"""{mfr_html} → {device_link}{doc_link}<br>
{wakeup_html}"""
        return markup

wakeup_time(baudrate=2400)

Calculate total wakeup time based on pattern repetitions

Source code in mbusread/mbus_config.py 
118
119
120
121
122
123
def wakeup_time(self, baudrate: int = 2400) -> float:
    """Calculate total wakeup time based on pattern repetitions"""
    secs = (self.wakeup_times * len(bytes.fromhex(self.wakeup_pattern))) / (
        baudrate / 10
    ) + self.wakeup_delay
    return secs

a link

Source code in mbusread/mbus_config.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
class Link:
    """
    a link
    """

    red = "color: red;text-decoration: underline;"
    blue = "color: blue;text-decoration: underline;"

    @staticmethod
    def create(
        url, text, tooltip=None, target=None, style: str = None, url_encode=False
    ):
        """
        Create a link for the given URL and text, with optional URL encoding.

        Args:
            url (str): The URL.
            text (str): The link text.
            tooltip (str): An optional tooltip.
            target (str): Target attribute, e.g., _blank for opening the link in a new tab.
            style (str): CSS style to be applied.
            url_encode (bool): Flag to indicate if the URL needs encoding. default: False

        Returns:
            str: HTML anchor tag as a string.
        """
        if url_encode:
            url = quote(url)

        title = "" if tooltip is None else f" title='{tooltip}'"
        target = "" if target is None else f" target='{target}'"
        if style is None:
            style = Link.blue
        style = f" style='{style}'"
        link = f"<a href='{url}'{title}{target}{style}>{text}</a>"
        return link

create(url, text, tooltip=None, target=None, style=None, url_encode=False) staticmethod

Create a link for the given URL and text, with optional URL encoding.

Parameters:

Name Type Description Default
url str

The URL.

 required
text str

The link text.

 required
tooltip str

An optional tooltip.

 None
target str

Target attribute, e.g., _blank for opening the link in a new tab.

 None
style str

CSS style to be applied.

 None
url_encode bool

Flag to indicate if the URL needs encoding. default: False

 False

Returns:

Name Type Description
str

HTML anchor tag as a string.
Source code in mbusread/mbus_config.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
@staticmethod
def create(
    url, text, tooltip=None, target=None, style: str = None, url_encode=False
):
    """
    Create a link for the given URL and text, with optional URL encoding.

    Args:
        url (str): The URL.
        text (str): The link text.
        tooltip (str): An optional tooltip.
        target (str): Target attribute, e.g., _blank for opening the link in a new tab.
        style (str): CSS style to be applied.
        url_encode (bool): Flag to indicate if the URL needs encoding. default: False

    Returns:
        str: HTML anchor tag as a string.
    """
    if url_encode:
        url = quote(url)

    title = "" if tooltip is None else f" title='{tooltip}'"
    target = "" if target is None else f" target='{target}'"
    if style is None:
        style = Link.blue
    style = f" style='{style}'"
    link = f"<a href='{url}'{title}{target}{style}>{text}</a>"
    return link

MBusConfig

Manages M-Bus manufacture/devices/message hierarchy

Source code in mbusread/mbus_config.py 
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
@lod_storable
class MBusConfig:
    """
    Manages M-Bus manufacture/devices/message hierarchy
    """

    manufacturers: Dict[str, Manufacturer] = field(default_factory=dict)

    @classmethod
    def get(cls, yaml_path: str = None) -> "MBusConfig":
        if yaml_path is None:
            yaml_path = cls.examples_path() + "/mbus_config.yaml"

        # Load raw YAML data
        mbus_config = cls.load_from_yaml_file(yaml_path)
        mbus_config.relink()
        return mbus_config

    def relink(self):
        """
        Link objects in the manufacturer/device/message hierarchy
        """
        for manufacturer in self.manufacturers.values():
            for _device_id, device in manufacturer.devices.items():
                device.manufacturer = manufacturer
                for _message_id, message in device.messages.items():
                    message.device = device

    @classmethod
    def examples_path(cls) -> str:
        # the root directory (default: examples)
        path = os.path.join(os.path.dirname(__file__), "../mbusread_examples")
        path = os.path.abspath(path)
        return path

Link objects in the manufacturer/device/message hierarchy

Source code in mbusread/mbus_config.py 
184
185
186
187
188
189
190
191
192
def relink(self):
    """
    Link objects in the manufacturer/device/message hierarchy
    """
    for manufacturer in self.manufacturers.values():
        for _device_id, device in manufacturer.devices.items():
            device.manufacturer = manufacturer
            for _message_id, message in device.messages.items():
                message.device = device

MBusIoConfig

Configuration data class for M-Bus reader

Source code in mbusread/mbus_config.py 
57
58
59
60
61
62
63
@lod_storable
class MBusIoConfig:
    """Configuration data class for M-Bus reader"""

    serial_device: str = "/dev/ttyUSB0"
    initial_baudrate: int = 2400
    timeout: float = 10.0

MBusMessage

An M-Bus message

Source code in mbusread/mbus_config.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
@lod_storable
class MBusMessage:
    """
    An M-Bus message
    """

    name: str
    title: str
    hex: str
    valid: bool = False

    def as_html(self) -> str:
        device_html = self.device.as_html() if hasattr(self, "device") else self.did
        example_text = f"{self.name}: {self.title}" if self.title else self.name
        return f"{device_html} → {example_text}"

Manufacturer

A manufacturer of M-Bus devices

Source code in mbusread/mbus_config.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
@lod_storable
class Manufacturer:
    """
    A manufacturer of M-Bus devices
    """

    name: str
    url: str
    country: str = "Germany"  # Most M-Bus manufacturers are German
    devices: Dict[str, Device] = field(default_factory=dict)

    def as_html(self) -> str:
        return (
            Link.create(url=self.url, text=self.name, target="_blank")
            if self.url
            else self.name
        )

MqttConfig

MQTT configuration

Source code in mbusread/mbus_config.py 
66
67
68
69
70
71
72
73
74
@lod_storable
class MqttConfig:
    """MQTT configuration"""

    broker: str = "localhost"
    port: int = 1883
    username: str = None
    password: str = None
    topic: str = "mbus/data"

mbus_mqtt

Created on 2025-01-24 based on https://github.com/ganehag/pyMeterBus/discussions/40

@author: Thorsten1982,wf

MBusMqtt

MQTT handler for M-Bus data

Source code in mbusread/mbus_mqtt.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
class MBusMqtt:
    """MQTT handler for M-Bus data"""

    def __init__(self, config: MqttConfig):
        self.config = config
        self.logger = logging.getLogger("MBusMqtt")

    @classmethod
    def from_yaml(cls, yaml_path: str) -> "MBusMqtt":
        config = cls.load_from_yaml_file(yaml_path)
        mqtt
        return mqtt

    def publish(self, record:Dict):
        """Publish M-Bus data via MQTT"""
        client = mqtt.Client()
        if self.config.username:
            client.username_pw_set(self.config.username, self.config.password)

        try:
            client.connect(self.config.broker, self.config.port, 60)
            client.loop_start()
            json_str = json.dumps(record, indent=2)
            client.publish(self.config.topic, json_str)
            time.sleep(1)
            client.loop_stop()
            client.disconnect()
        except Exception as e:
            self.logger.error(f"MQTT error: {str(e)}")

publish(record)

Publish M-Bus data via MQTT

Source code in mbusread/mbus_mqtt.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
def publish(self, record:Dict):
    """Publish M-Bus data via MQTT"""
    client = mqtt.Client()
    if self.config.username:
        client.username_pw_set(self.config.username, self.config.password)

    try:
        client.connect(self.config.broker, self.config.port, 60)
        client.loop_start()
        json_str = json.dumps(record, indent=2)
        client.publish(self.config.topic, json_str)
        time.sleep(1)
        client.loop_stop()
        client.disconnect()
    except Exception as e:
        self.logger.error(f"MQTT error: {str(e)}")

mbus_parser

Created on 2025-01-22 see also https://github.com/ganehag/pyMeterBus/discussions/40 @author: Thorsten1982, wf

MBusParser

parse MBus data

Source code in mbusread/mbus_parser.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
class MBusParser:
    """
    parse MBus data
    """

    def __init__(self, debug: bool = False):
        self.debug = debug
        self.logger = Logger.setup_logger(debug=debug)

    def fromhex(self, x, base=16):
        """Convert hex string to integer"""
        return int(x, base)

    def get_frame_json(self, frame):
        """
        Workarounds for JSON bugs in pyMeterBus
        """
        if isinstance(frame, TelegramShort):
            # Handle serialization explicitly for TelegramShort
            interpreted_data = frame.interpreted
            json_str = json.dumps(
                interpreted_data, sort_keys=True, indent=2, default=str
            )
        elif hasattr(frame, "to_JSON"):
            json_str = frame.to_JSON()
        else:
            # Fallback to basic frame info
            data = {
                "header": {
                    "start": frame.header.startField.parts[0],
                    "length": len(frame.body.bodyHeader.ci_field.parts) + 2,
                    "control": frame.header.cField.parts[0],
                    "address": frame.header.aField.parts[0],
                },
                "body": {"ci_field": frame.body.bodyHeader.ci_field.parts[0]},
            }
            json_str = json.dumps(data)
        return json_str

    def parse_mbus_frame(self, hex_data):
        """
        Parse M-Bus hex data and return mbus frame
        Returns tuple of (error_msg, mbus_frame)
        """
        frame = None
        error_msg = None
        try:
            filtered_data = "".join(char for char in hex_data if char.isalnum())
            data = list(map(self.fromhex, re.findall("..", filtered_data)))
            frame = meterbus.load(data)
        except Exception as ex:
            error_type = type(ex).__name__
            error_msg = f"Error parsing M-Bus data: {error_type}: {str(ex)}"
            if self.debug:
                traceback.format_exception(ex)
        return error_msg, frame

    def extract_frame(self, data: bytes) -> Optional[bytes]:
        """Extract valid M-Bus frame between start (0x68) and end (0x16) bytes"""
        start_byte = b"\x68"
        end_byte = b"\x16"
        result = None
        status = "❌"

        if data:
            try:
                start_idx = data.index(start_byte)
                end_idx = data.find(end_byte, start_idx + 1)
                if end_idx != -1:
                    result = data[start_idx : end_idx + 1]
                    status = "✅"
                else:
                    status = "⚠️"
            except ValueError:
                pass

        self.logger.debug(
            f"Frame extraction {status}: {result.hex() if result else 'None'}"
        )
        return result

extract_frame(data)

Extract valid M-Bus frame between start (0x68) and end (0x16) bytes

Source code in mbusread/mbus_parser.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def extract_frame(self, data: bytes) -> Optional[bytes]:
    """Extract valid M-Bus frame between start (0x68) and end (0x16) bytes"""
    start_byte = b"\x68"
    end_byte = b"\x16"
    result = None
    status = "❌"

    if data:
        try:
            start_idx = data.index(start_byte)
            end_idx = data.find(end_byte, start_idx + 1)
            if end_idx != -1:
                result = data[start_idx : end_idx + 1]
                status = "✅"
            else:
                status = "⚠️"
        except ValueError:
            pass

    self.logger.debug(
        f"Frame extraction {status}: {result.hex() if result else 'None'}"
    )
    return result

fromhex(x, base=16)

Convert hex string to integer

Source code in mbusread/mbus_parser.py 
28
29
30
def fromhex(self, x, base=16):
    """Convert hex string to integer"""
    return int(x, base)

get_frame_json(frame)

Workarounds for JSON bugs in pyMeterBus

Source code in mbusread/mbus_parser.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def get_frame_json(self, frame):
    """
    Workarounds for JSON bugs in pyMeterBus
    """
    if isinstance(frame, TelegramShort):
        # Handle serialization explicitly for TelegramShort
        interpreted_data = frame.interpreted
        json_str = json.dumps(
            interpreted_data, sort_keys=True, indent=2, default=str
        )
    elif hasattr(frame, "to_JSON"):
        json_str = frame.to_JSON()
    else:
        # Fallback to basic frame info
        data = {
            "header": {
                "start": frame.header.startField.parts[0],
                "length": len(frame.body.bodyHeader.ci_field.parts) + 2,
                "control": frame.header.cField.parts[0],
                "address": frame.header.aField.parts[0],
            },
            "body": {"ci_field": frame.body.bodyHeader.ci_field.parts[0]},
        }
        json_str = json.dumps(data)
    return json_str

parse_mbus_frame(hex_data)

Parse M-Bus hex data and return mbus frame Returns tuple of (error_msg, mbus_frame)

Source code in mbusread/mbus_parser.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def parse_mbus_frame(self, hex_data):
    """
    Parse M-Bus hex data and return mbus frame
    Returns tuple of (error_msg, mbus_frame)
    """
    frame = None
    error_msg = None
    try:
        filtered_data = "".join(char for char in hex_data if char.isalnum())
        data = list(map(self.fromhex, re.findall("..", filtered_data)))
        frame = meterbus.load(data)
    except Exception as ex:
        error_type = type(ex).__name__
        error_msg = f"Error parsing M-Bus data: {error_type}: {str(ex)}"
        if self.debug:
            traceback.format_exception(ex)
    return error_msg, frame

mbus_reader

Created on 2025-01-24 based on https://github.com/ganehag/pyMeterBus/discussions/40

@author: Thorsten1982,wf

MBusReader

Reader for Meter Bus data

Source code in mbusread/mbus_reader.py 
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
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
class MBusReader:
    """Reader for Meter Bus data"""

    def __init__(
        self,
        device: Device,
        io_config: Optional[MBusIoConfig] = None,
        i18n: I18n = None,
        debug: bool = False,
    ):
        """
        Initialize MBusReader with configuration
        """
        self.debug = debug
        self.device = device
        self.logger = Logger.setup_logger(debug=debug)
        self.io_config = io_config or MBusIoConfig
        if i18n is None:
            i18n = I18n.default()
        self.i18n = i18n
        self.ser = self._setup_serial()

    def _setup_serial(self) -> serial.Serial:
        """Initialize serial connection"""
        ser = serial.Serial(
            port=self.io_config.serial_device,
            baudrate=self.io_config.initial_baudrate,
            bytesize=8,
            parity=serial.PARITY_NONE,
            stopbits=1,
            timeout=self.io_config.timeout,
        )
        return ser

    def show_echo(self, msg: str, echo: str, echo_display_len: int = 32):
        if echo != msg:
            # Truncate to first echo_display_len bytes for readability
            sent_hex = msg[:echo_display_len].hex()
            echo_hex = echo[:echo_display_len].hex()
            warn_msg = f"""Echo mismatch!  Sent {len(msg)} Repl {len(echo)}
Sent={sent_hex}
Repl={echo_hex}"""
            self.logger.warning(warn_msg)
        else:
            self.logger.debug(f"Echo matched: {len(echo)} bytes")

    def ser_write(self, msg: bytes, info: str, echo_display_len: int = 16) -> None:
        """
        Writes a message to the serial port and validates the echo.

        Args:
            msg (bytes): The message to write as a byte string.
            info (str): The log message key for identifying the operation.

        Logs:
            A warning if the echo does not match the sent message.
            A debug message if the echo matches.
        """
        self.logger.info(self.i18n.get(info))
        self.ser.write(msg)
        self.ser.flush()
        if self.device.has_echo:
            # Check and validate echo
            echo = self.ser.read(len(msg))
            self.show_echo(msg, echo, echo_display_len)

    def wake_up(self, device: Device) -> None:
        """Perform the wakeup sequence based on device configuration"""
        try:
            pattern = bytes.fromhex(device.wakeup_pattern)
            times = device.wakeup_times
            sleep_time = device.wakeup_delay

            self.ser_write(pattern * times, "wake_up_started")
            time.sleep(sleep_time)
            self.ser.parity = serial.PARITY_EVEN
            self.logger.info(self.i18n.get("wake_up_complete"))
        except serial.SerialException as e:
            self.logger.error(self.i18n.get("serial_error", "wake_up", str(e)))

    def get_data(self, read_data_msg_key: str = "read_data") -> Optional[bytes]:
        """Get data from the M-Bus device"""
        try:
            if read_data_msg_key not in self.device.messages:
                raise ValueError(f"Message {read_data_msg_key} not found")

            self.wake_up(self.device)
            read_data = bytes.fromhex(self.device.messages[read_data_msg_key].hex)
            self.ser_write(read_data, "reading_data")

            result = self.ser.read(620)
            if not result:
                self.logger.warning(self.i18n.get("no_data_received"))
                return None

            byte_array_hex = binascii.hexlify(result)
            self.logger.info(self.i18n.get("read_data_hex", byte_array_hex.decode()))
            return result

        except serial.SerialException as e:
            self.logger.error(self.i18n.get("serial_error", "get_data", str(e)))
            return None

    def send_mbus_request(self, msg_id: str) -> None:
        """Send an M-Bus request to the device"""
        try:
            if msg_id not in self.device.messages:
                raise ValueError(f"Message {msg_id} not found in device configuration")

            request = bytes.fromhex(self.device.messages[msg_id].hex)
            self.logger.info(self.i18n.get("sending_request"))
            self.ser.write(request)
        except serial.SerialException as e:
            self.logger.error(
                self.i18n.get("serial_error", "send_mbus_request", str(e))
            )

    def read_response(self, buffer_size: int = 256) -> Optional[bytes]:
        """Read the response from the device"""
        try:
            response = self.ser.read(buffer_size)
            if response:
                hex_response = " ".join(format(b, "02x") for b in response)
                self.logger.info(self.i18n.get("response_received", hex_response))
                return response
            return None
        except serial.SerialException as e:
            self.logger.error(self.i18n.get("serial_error", "read_response", str(e)))
            return None

    def close(self) -> None:
        """Close the serial connection"""
        if self.ser and self.ser.is_open:
            self.ser.close()

__init__(device, io_config=None, i18n=None, debug=False)

Initialize MBusReader with configuration

Source code in mbusread/mbus_reader.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
def __init__(
    self,
    device: Device,
    io_config: Optional[MBusIoConfig] = None,
    i18n: I18n = None,
    debug: bool = False,
):
    """
    Initialize MBusReader with configuration
    """
    self.debug = debug
    self.device = device
    self.logger = Logger.setup_logger(debug=debug)
    self.io_config = io_config or MBusIoConfig
    if i18n is None:
        i18n = I18n.default()
    self.i18n = i18n
    self.ser = self._setup_serial()

close()

Close the serial connection

Source code in mbusread/mbus_reader.py 
150
151
152
153
def close(self) -> None:
    """Close the serial connection"""
    if self.ser and self.ser.is_open:
        self.ser.close()

get_data(read_data_msg_key='read_data')

Get data from the M-Bus device

Source code in mbusread/mbus_reader.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
def get_data(self, read_data_msg_key: str = "read_data") -> Optional[bytes]:
    """Get data from the M-Bus device"""
    try:
        if read_data_msg_key not in self.device.messages:
            raise ValueError(f"Message {read_data_msg_key} not found")

        self.wake_up(self.device)
        read_data = bytes.fromhex(self.device.messages[read_data_msg_key].hex)
        self.ser_write(read_data, "reading_data")

        result = self.ser.read(620)
        if not result:
            self.logger.warning(self.i18n.get("no_data_received"))
            return None

        byte_array_hex = binascii.hexlify(result)
        self.logger.info(self.i18n.get("read_data_hex", byte_array_hex.decode()))
        return result

    except serial.SerialException as e:
        self.logger.error(self.i18n.get("serial_error", "get_data", str(e)))
        return None

read_response(buffer_size=256)

Read the response from the device

Source code in mbusread/mbus_reader.py 
137
138
139
140
141
142
143
144
145
146
147
148
def read_response(self, buffer_size: int = 256) -> Optional[bytes]:
    """Read the response from the device"""
    try:
        response = self.ser.read(buffer_size)
        if response:
            hex_response = " ".join(format(b, "02x") for b in response)
            self.logger.info(self.i18n.get("response_received", hex_response))
            return response
        return None
    except serial.SerialException as e:
        self.logger.error(self.i18n.get("serial_error", "read_response", str(e)))
        return None

send_mbus_request(msg_id)

Send an M-Bus request to the device

Source code in mbusread/mbus_reader.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
def send_mbus_request(self, msg_id: str) -> None:
    """Send an M-Bus request to the device"""
    try:
        if msg_id not in self.device.messages:
            raise ValueError(f"Message {msg_id} not found in device configuration")

        request = bytes.fromhex(self.device.messages[msg_id].hex)
        self.logger.info(self.i18n.get("sending_request"))
        self.ser.write(request)
    except serial.SerialException as e:
        self.logger.error(
            self.i18n.get("serial_error", "send_mbus_request", str(e))
        )

ser_write(msg, info, echo_display_len=16)

Writes a message to the serial port and validates the echo.

Parameters:

Name Type Description Default
msg bytes

The message to write as a byte string.

 required
info str

The log message key for identifying the operation.

 required
Logs

A warning if the echo does not match the sent message. A debug message if the echo matches.

Source code in mbusread/mbus_reader.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def ser_write(self, msg: bytes, info: str, echo_display_len: int = 16) -> None:
    """
    Writes a message to the serial port and validates the echo.

    Args:
        msg (bytes): The message to write as a byte string.
        info (str): The log message key for identifying the operation.

    Logs:
        A warning if the echo does not match the sent message.
        A debug message if the echo matches.
    """
    self.logger.info(self.i18n.get(info))
    self.ser.write(msg)
    self.ser.flush()
    if self.device.has_echo:
        # Check and validate echo
        echo = self.ser.read(len(msg))
        self.show_echo(msg, echo, echo_display_len)

wake_up(device)

Perform the wakeup sequence based on device configuration

Source code in mbusread/mbus_reader.py 
86
87
88
89
90
91
92
93
94
95
96
97
98
def wake_up(self, device: Device) -> None:
    """Perform the wakeup sequence based on device configuration"""
    try:
        pattern = bytes.fromhex(device.wakeup_pattern)
        times = device.wakeup_times
        sleep_time = device.wakeup_delay

        self.ser_write(pattern * times, "wake_up_started")
        time.sleep(sleep_time)
        self.ser.parity = serial.PARITY_EVEN
        self.logger.info(self.i18n.get("wake_up_complete"))
    except serial.SerialException as e:
        self.logger.error(self.i18n.get("serial_error", "wake_up", str(e)))

mbus_reader_cmd

MBusCommunicator

communicate with an M-Bus device

Source code in mbusread/mbus_reader_cmd.py 
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
class MBusCommunicator:
    """
    communicate with an M-Bus device
    """

    def __init__(self, args: argparse.Namespace):
        self.args=args
        self.logger = Logger.setup_logger(args.debug)
        i18n = I18n.default()
        i18n.language = args.lang
        mbus_config = MBusConfig.get(args.config)
        io_config = MBusIoConfig.load_from_yaml_file(args.io_config)
        device = mbus_config.manufacturers[args.manufacturer].devices[args.device]
        self.mqtt_config = (
            MqttConfig.load_from_yaml_file(args.mqtt_config)
            if args.mqtt_config
            else None
        )

        self.reader = MBusReader(
            device=device, io_config=io_config, i18n=i18n, debug=args.debug
        )
        self.parser = MBusParser(debug=args.debug)

        pass

    @classmethod
    def get_parser(cls) -> argparse.ArgumentParser:
        parser = argparse.ArgumentParser(description="M-Bus Reader")
        parser.add_argument(
            "-c",
            "--config",
            default=MBusConfig.examples_path() + "/mbus_config.yaml",
            help="Config file path [default: %(default)s]",
        )
        parser.add_argument(
            "-i",
            "--io_config",
            default=MBusConfig.examples_path() + "/mbus_io_config.yaml",
            help="IO config file path [default: %(default)s]",
        )
        parser.add_argument(
            "-q",
            "--mqtt_config",
            default=MBusConfig.examples_path() + "/mqtt_config.yaml",
            help="MQTT config file path [default: %(default)s]",
        )

        parser.add_argument(
            "-D",
            "--device",
            default="cf_echo_ii",
            help="Device type [default: %(default)s]",
        )
        parser.add_argument(
            "-m", "--message", help="Message ID to send [default: %(default)s]"
        )
        parser.add_argument(
            "-M",
            "--manufacturer",
            default="allmess",
            help="Manufacturer ID [default: %(default)s]",
        )
        parser.add_argument(
            "--mqtt", action="store_true", help="Enable MQTT publishing"
        )
        parser.add_argument(
            "--lang",
            choices=["en", "de"],
            default="en",
            help="Language for messages (default: en)",
        )
        parser.add_argument("--debug", action="store_true", help="Enable debug logging")
        return parser

    def work(self):
        try:
            if self.args.message:
                self.reader.send_mbus_request(self.args.message)
                raw_data = self.reader.read_response()
            else:
                raw_data = self.reader.get_data()
            frame = self.parser.extract_frame(raw_data)
            if not frame:
                self.logger.warning("No valid frame found in data")
                return None

            error_msg, mbus_frame = self.parser.parse_mbus_frame(frame.hex())
            if error_msg:
                self.logger.error(f"Frame parsing error: {error_msg}")
                return None

            json_str = self.parser.get_frame_json(mbus_frame)
            record = json.loads(json_str)
            pretty_json=json.dumps(record, indent=2, default=str)
            if self.args.debug:
                print(pretty_json)

            if self.args.mqtt and self.mqtt_config:
                mqtt_handler = MBusMqtt(self.mqtt_config)
                mqtt_handler.publish(record)
        finally:
            self.reader.close()

mbus_viewer

MBusViewer

Bases: MBusParser

Enhanced M-Bus message viewer with improved error handling and UI organization

Source code in mbusread/mbus_viewer.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
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
166
167
168
169
170
171
172
173
174
175
176
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
class MBusViewer(MBusParser):
    """Enhanced M-Bus message viewer with improved error handling and UI organization"""

    def __init__(self, solution=None):
        super().__init__()
        self.solution = solution
        self.config = MBusConfig.get()

        # Initialize UI components
        self.hex_input: Optional[ui.textarea] = None
        self.json_view: Optional[ui.code] = None
        self.details_view: Optional[ui.html] = None
        self.error_view: Optional[ui.html] = None

    def create_textarea(
        self, label: str, placeholder: Optional[str] = None, height: str = "h-32"
    ) -> ui.textarea:
        """Create a consistent textarea with error handling"""
        return (
            ui.textarea(label=label, placeholder=placeholder)
            .classes(f"w-full {height}")
            .props("clearable outlined")
        )

    def setup_ui(self) -> None:
        """Create the main UI layout with two-column design"""
        try:
            ui.label("M-Bus Message Parser").classes("text-2xl font-bold mb-4")

            with ui.row().classes("w-full gap-4"):
                # Left column
                with ui.column().classes("flex-1"):
                    with ui.card().classes("w-full"):
                        with ui.row().classes("gap-8"):
                            self.manufacturer_select = RadioSelection[Manufacturer](
                                "Manufacturer",
                                "name",
                                selection=self.config.manufacturers,
                            ).setup(self._on_manufacturer_change)

                            self.device_select = RadioSelection[Device](
                                "Device",
                                "model",
                                selection=self.manufacturer_select.item.devices,
                            ).setup(self._on_device_change)

                        # Device details
                        with ui.card().classes("w-full"):
                            ui.label("Device Details").classes("text-lg font-bold mb-2")
                            self.details_view = ui.html()

                        with ui.row().classes("mt-4"):
                            self.message_select = RadioSelection[MBusMessage](
                                "Message",
                                "name",
                                selection=self.device_select.item.messages,
                            ).setup(self._on_message_change)

                    # Input area
                    with ui.card().classes("w-full mt-4"):
                        self.hex_input = self.create_textarea(
                            "Enter M-Bus hex message",
                            "e.g. 68 4d 4d 68 08 00 72 26 54 83 22 77...",
                        )
                        ui.button(
                            "Parse Message", on_click=self._parse_message
                        ).classes("mt-4")

                # Right column
                with ui.column().classes("flex-1"):
                    # Results area
                    with ui.row().classes("w-full"):
                        self.error_view = ui.html().classes("text-red-500")
                        self.json_view = ui.code("", language="json").classes(
                            "w-full h-96"
                        )

        except Exception as ex:
            self._handle_error("Error setting up UI", ex)

    def _on_manufacturer_change(self, manufacturer: Manufacturer) -> None:
        """Update device options when manufacturer changes"""
        try:
            self.device_select.selection = manufacturer.devices
            self.device_select._update_options()
        except Exception as ex:
            self._handle_error("Error updating devices", ex)

    def _on_device_change(self, device: Device) -> None:
        """Update message options when device changes"""
        try:
            self.message_select.selection = device.messages
            self.message_select._update_options()
            self.details_view.content = device.as_html()
        except Exception as ex:
            self._handle_error("Error updating messages", ex)

    def _on_message_change(self, message: MBusMessage) -> None:
        """Update display when message changes"""
        try:
            self.hex_input.value = message.hex
            self.details_view.content = message.device.as_html()
            self._parse_message()
        except Exception as ex:
            self._handle_error("Error updating message display", ex)

    def _parse_message(self) -> None:
        """Parse the M-Bus message with comprehensive error handling"""
        try:
            self.json_view.content = ""
            self.error_view.content = ""

            hex_str = self.hex_input.value
            if not hex_str:
                raise ValueError("Please enter a hex message")

            error_msg, frame = self.parse_mbus_frame(hex_str)
            if error_msg:
                raise ValueError(error_msg)

            json_str = self.get_frame_json(frame)
            self.json_view.content = json_str

        except Exception as ex:
            self._handle_error("Error parsing message", ex)

    def _handle_error(self, context: str, error: Exception) -> None:
        """Centralized error handling with user feedback"""
        error_msg = f"{context}: {str(error)}"
        self.error_view.content = f"<div class='text-red-500'>{error_msg}</div>"
        ui.notify(error_msg, type="negative")

        if self.solution:
            self.solution.handle_exception(error)

create_textarea(label, placeholder=None, height='h-32')

Create a consistent textarea with error handling

Source code in mbusread/mbus_viewer.py 
90
91
92
93
94
95
96
97
98
def create_textarea(
    self, label: str, placeholder: Optional[str] = None, height: str = "h-32"
) -> ui.textarea:
    """Create a consistent textarea with error handling"""
    return (
        ui.textarea(label=label, placeholder=placeholder)
        .classes(f"w-full {height}")
        .props("clearable outlined")
    )

setup_ui()

Create the main UI layout with two-column design

Source code in mbusread/mbus_viewer.py 
100
101
102
103
104
105
106
107
108
109
110
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
def setup_ui(self) -> None:
    """Create the main UI layout with two-column design"""
    try:
        ui.label("M-Bus Message Parser").classes("text-2xl font-bold mb-4")

        with ui.row().classes("w-full gap-4"):
            # Left column
            with ui.column().classes("flex-1"):
                with ui.card().classes("w-full"):
                    with ui.row().classes("gap-8"):
                        self.manufacturer_select = RadioSelection[Manufacturer](
                            "Manufacturer",
                            "name",
                            selection=self.config.manufacturers,
                        ).setup(self._on_manufacturer_change)

                        self.device_select = RadioSelection[Device](
                            "Device",
                            "model",
                            selection=self.manufacturer_select.item.devices,
                        ).setup(self._on_device_change)

                    # Device details
                    with ui.card().classes("w-full"):
                        ui.label("Device Details").classes("text-lg font-bold mb-2")
                        self.details_view = ui.html()

                    with ui.row().classes("mt-4"):
                        self.message_select = RadioSelection[MBusMessage](
                            "Message",
                            "name",
                            selection=self.device_select.item.messages,
                        ).setup(self._on_message_change)

                # Input area
                with ui.card().classes("w-full mt-4"):
                    self.hex_input = self.create_textarea(
                        "Enter M-Bus hex message",
                        "e.g. 68 4d 4d 68 08 00 72 26 54 83 22 77...",
                    )
                    ui.button(
                        "Parse Message", on_click=self._parse_message
                    ).classes("mt-4")

            # Right column
            with ui.column().classes("flex-1"):
                # Results area
                with ui.row().classes("w-full"):
                    self.error_view = ui.html().classes("text-red-500")
                    self.json_view = ui.code("", language="json").classes(
                        "w-full h-96"
                    )

    except Exception as ex:
        self._handle_error("Error setting up UI", ex)

RadioSelection dataclass

Bases: Generic[T]

Generic radio button selection with type hints and improved structure

Source code in mbusread/mbus_viewer.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
@dataclass
class RadioSelection(Generic[T]):
    """Generic radio button selection with type hints and improved structure"""

    title: str
    key_attr: str
    selection: Dict[str, T] = field(default_factory=dict)
    value: Optional[str] = None
    item: Optional[T] = None
    on_change: Optional[Callable[[T], None]] = None

    def __post_init__(self):
        self.radio = None
        self.label = None
        self.options = observables.ObservableDict()

    def setup(
        self, on_change: Optional[Callable[[T], None]] = None
    ) -> "RadioSelection[T]":
        """Initialize the radio selection UI components"""
        self.on_change = on_change
        with ui.column():
            self.label = ui.label(self.title).classes("font-bold text-lg")
            self.options = observables.ObservableDict()
            self._update_options()
            self.radio = ui.radio(
                options=list(self.options.keys()),
                on_change=self._handle_change,
                value=self.value,
            ).props("inline dense")
        return self

    def _update_options(self) -> None:
        """Update radio options with error handling"""
        try:
            self.options.clear()
            for i, (key, item) in enumerate(self.selection.items()):
                value = getattr(item, self.key_attr)
                self.options[value] = key
                if i == 0:  # Set initial selection
                    self.value = value
                    self.item = item

            if self.radio:
                self.radio.clear()
                self.radio.options = list(self.options.keys())
                self.radio.update()

        except AttributeError as e:
            ui.notify(f"Error updating options: {str(e)}", type="negative")

    def _handle_change(self, event) -> None:
        """Handle radio selection changes with validation"""
        try:
            self.value = event.value
            key = self.options.get(self.value)
            if key is not None:
                self.item = self.selection[key]
                if self.on_change:
                    self.on_change(self.item)
        except KeyError as e:
            ui.notify(f"Invalid selection: {str(e)}", type="negative")

setup(on_change=None)

Initialize the radio selection UI components

Source code in mbusread/mbus_viewer.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
def setup(
    self, on_change: Optional[Callable[[T], None]] = None
) -> "RadioSelection[T]":
    """Initialize the radio selection UI components"""
    self.on_change = on_change
    with ui.column():
        self.label = ui.label(self.title).classes("font-bold text-lg")
        self.options = observables.ObservableDict()
        self._update_options()
        self.radio = ui.radio(
            options=list(self.options.keys()),
            on_change=self._handle_change,
            value=self.value,
        ).props("inline dense")
    return self

mbus_viewer_cmd

Created on 2025-01-22

@author: wf

NiceMBusCmd

Bases: WebserverCmd

command line handling for ngwidgets

Source code in mbusread/mbus_viewer_cmd.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
class NiceMBusCmd(WebserverCmd):
    """
    command line handling for ngwidgets
    """

    def getArgParser(self, description: str, version_msg) -> ArgumentParser:
        """
        override the default argparser call
        """
        parser = super().getArgParser(description, version_msg)
        parser.add_argument(
            "-rp",
            "--root_path",
            default=MBusConfig.examples_path(),
            help="path to mbux hex files [default: %(default)s]",
        )
        return parser

getArgParser(description, version_msg)

override the default argparser call

Source code in mbusread/mbus_viewer_cmd.py 
21
22
23
24
25
26
27
28
29
30
31
32
def getArgParser(self, description: str, version_msg) -> ArgumentParser:
    """
    override the default argparser call
    """
    parser = super().getArgParser(description, version_msg)
    parser.add_argument(
        "-rp",
        "--root_path",
        default=MBusConfig.examples_path(),
        help="path to mbux hex files [default: %(default)s]",
    )
    return parser

main(argv=None)

main call

Source code in mbusread/mbus_viewer_cmd.py 
35
36
37
38
39
40
41
42
43
44
def main(argv: list = None):
    """
    main call
    """
    cmd = NiceMBusCmd(
        config=NiceMBusWebserver.get_config(),
        webserver_cls=NiceMBusWebserver,
    )
    exit_code = cmd.cmd_main(argv)
    return exit_code

mbus_viewer_server

Created on 22.01.2025

@author: wf

NiceMBus

Bases: InputWebSolution

Source code in mbusread/mbus_viewer_server.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
class NiceMBus(InputWebSolution):
    """ """

    def __init__(self, webserver: "NiceMBusWebserver", client: Client):
        super().__init__(webserver, client)

    async def home(self):
        """
        provide the main content page
        """

        def setup_home():
            viewer = MBusViewer(solution=self)
            viewer.setup_ui()

        await self.setup_content_div(setup_home)

home() async

provide the main content page

Source code in mbusread/mbus_viewer_server.py 
61
62
63
64
65
66
67
68
69
70
async def home(self):
    """
    provide the main content page
    """

    def setup_home():
        viewer = MBusViewer(solution=self)
        viewer.setup_ui()

    await self.setup_content_div(setup_home)

NiceMBusWebserver

Bases: InputWebserver

webserver to demonstrate ngwidgets capabilities

Source code in mbusread/mbus_viewer_server.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
class NiceMBusWebserver(InputWebserver):
    """
    webserver to demonstrate ngwidgets capabilities
    """

    @classmethod
    def get_config(cls) -> WebserverConfig:
        copy_right = "(c)2025 Wolfgang Fahl"
        config = WebserverConfig(
            short_name="mbus_viewer",
            timeout=6.0,
            copy_right=copy_right,
            version=Version(),
            default_port=9996,
        )
        server_config = WebserverConfig.get(config)
        server_config.solution_class = NiceMBus
        return server_config

    def __init__(self):
        """
        Constructor
        """
        InputWebserver.__init__(self, config=NiceMBusWebserver.get_config())
        pass

    def configure_run(self):
        root_path = (
            self.args.root_path if self.args.root_path else MBusConfig.examples_path()
        )
        self.root_path = os.path.abspath(root_path)
        self.allowed_urls = [
            "https://raw.githubusercontent.com/WolfgangFahl/nicescad/main/examples/",
            "https://raw.githubusercontent.com/openscad/openscad/master/examples/",
            self.root_path,
        ]

__init__()

Constructor

Source code in mbusread/mbus_viewer_server.py 
36
37
38
39
40
41
def __init__(self):
    """
    Constructor
    """
    InputWebserver.__init__(self, config=NiceMBusWebserver.get_config())
    pass

version

Created on 22.01.2025

@author: wf

Version dataclass

Version handling for nicegui widgets

Source code in mbusread/version.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
@dataclass
class Version:
    """
    Version handling for nicegui widgets
    """

    name = "mbusreader"
    version = mbusread.__version__
    date = "2025-01-22"
    updated = "2025-01-25"
    description = "MBus message parser and JSON result viewer"

    authors = "Wolfgang Fahl"

    doc_url = "https://wiki.bitplan.com/index.php/MBus_Reader"
    chat_url = "https://github.com/WolfgangFahl/mbusreader/discussions"
    cm_url = "https://github.com/WolfgangFahl/mbusreader"

    license = f"""Copyright 2025 contributors. All rights reserved.

  Licensed under the Apache License 2.0
  http://www.apache.org/licenses/LICENSE-2.0

  Distributed on an "AS IS" basis without warranties
  or conditions of any kind, either express or implied."""

    longDescription = f"""{name} version {version}
{description}

  Created by {authors} on {date} last updated {updated}"""

yamlable

Created on 2023-12-08, Extended on 2023-16-12 and 2024-01-25

@author: wf, ChatGPT

Prompts for the development and extension of the 'YamlAble' class within the 'yamable' module:

  1. Develop 'YamlAble' class in 'yamable' module. It should convert dataclass instances to/from YAML.
  2. Implement methods for YAML block scalar style and exclude None values in 'YamlAble' class.
  3. Add functionality to remove None values from dataclass instances before YAML conversion.
  4. Ensure 'YamlAble' processes only dataclass instances, with error handling for non-dataclass objects.
  5. Extend 'YamlAble' for JSON serialization and deserialization.
  6. Add methods for saving/loading dataclass instances to/from YAML and JSON files in 'YamlAble'.
  7. Implement loading of dataclass instances from URLs for both YAML and JSON in 'YamlAble'.
  8. Write tests for 'YamlAble' within the pyLodStorage context. Use 'samples 2' example from pyLoDStorage https://github.com/WolfgangFahl/pyLoDStorage/blob/master/lodstorage/sample2.py as a reference.
  9. Ensure tests cover YAML/JSON serialization, deserialization, and file I/O operations, using the sample-based approach..
  10. Use Google-style docstrings, comments, and type hints in 'YamlAble' class and tests.
  11. Adhere to instructions and seek clarification for any uncertainties.
  12. Add @lod_storable annotation support that will automatically YamlAble support and add @dataclass and @dataclass_json prerequisite behavior to a class

DateConvert

date converter

Source code in mbusread/yamlable.py 
76
77
78
79
80
81
82
83
84
class DateConvert:
    """
    date converter
    """

    @classmethod
    def iso_date_to_datetime(cls, iso_date: str) -> datetime.date:
        date = datetime.strptime(iso_date, "%Y-%m-%d").date() if iso_date else None
        return date

YamlAble

Bases: Generic[T]

An extended YAML handler class for converting dataclass objects to and from YAML format, and handling loading from and saving to files and URLs.

Source code in mbusread/yamlable.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
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
166
167
168
169
170
171
172
173
174
175
176
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
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
class YamlAble(Generic[T]):
    """
    An extended YAML handler class for converting dataclass objects to and from YAML format,
    and handling loading from and saving to files and URLs.
    """

    def _yaml_setup(self):
        """
        Initializes the YamAble handler, setting up custom representers and preparing it for various operations.
        """
        if not is_dataclass(self):
            raise ValueError("I must be a dataclass instance.")
        if not hasattr(self, "_yaml_dumper"):
            self._yaml_dumper = yaml.Dumper
            self._yaml_dumper.ignore_aliases = lambda *_args: True
            self._yaml_dumper.add_representer(type(None), self.represent_none)
            self._yaml_dumper.add_representer(str, self.represent_literal)

    def represent_none(self, _, __) -> yaml.Node:
        """
        Custom representer for ignoring None values in the YAML output.
        """
        return self._yaml_dumper.represent_scalar("tag:yaml.org,2002:null", "")

    def represent_literal(self, dumper: yaml.Dumper, data: str) -> yaml.Node:
        """
        Custom representer for block scalar style for strings.
        """
        if "\n" in data:
            return dumper.represent_scalar("tag:yaml.org,2002:str", data, style="|")
        return dumper.represent_scalar("tag:yaml.org,2002:str", data)

    def to_yaml(
        self,
        ignore_none: bool = True,
        ignore_underscore: bool = True,
        allow_unicode: bool = True,
        sort_keys: bool = False,
    ) -> str:
        """
        Converts this dataclass object to a YAML string, with options to omit None values and/or underscore-prefixed variables,
        and using block scalar style for strings.

        Args:
            ignore_none: Flag to indicate whether None values should be removed from the YAML output.
            ignore_underscore: Flag to indicate whether attributes starting with an underscore should be excluded from the YAML output.
            allow_unicode: Flag to indicate whether to allow unicode characters in the output.
            sort_keys: Flag to indicate whether to sort the dictionary keys in the output.

        Returns:
            A string representation of the dataclass object in YAML format.
        """
        obj_dict = asdict(self)
        self._yaml_setup()
        clean_dict = self.remove_ignored_values(
            obj_dict, ignore_none, ignore_underscore
        )
        yaml_str = yaml.dump(
            clean_dict,
            Dumper=self._yaml_dumper,
            default_flow_style=False,
            allow_unicode=allow_unicode,
            sort_keys=sort_keys,
        )
        return yaml_str

    @classmethod
    def from_yaml(cls: Type[T], yaml_str: str) -> T:
        """
        Deserializes a YAML string to a dataclass instance.

        Args:
            yaml_str (str): A string containing YAML formatted data.

        Returns:
            T: An instance of the dataclass.
        """
        data: dict[str, Any] = yaml.safe_load(yaml_str)
        instance: T = cls.from_dict(data)
        return instance

    @classmethod
    def load_from_yaml_file(cls: Type[T], filename: str) -> T:
        """
        Loads a dataclass instance from a YAML file.

        Args:
            filename (str): The path to the YAML file.

        Returns:
            T: An instance of the dataclass.
        """
        with open(filename, "r") as file:
            yaml_str: str = file.read()
        instance: T = cls.from_yaml(yaml_str)
        return instance

    @classmethod
    def load_from_yaml_url(cls: Type[T], url: str) -> T:
        """
        Loads a dataclass instance from a YAML string obtained from a URL.

        Args:
            url (str): The URL pointing to the YAML data.

        Returns:
            T: An instance of the dataclass.
        """
        yaml_str: str = cls.read_from_url(url)
        instance: T = cls.from_yaml(yaml_str)
        return instance

    def save_to_yaml_file(self, filename: str):
        """
        Saves the current dataclass instance to a YAML file.

        Args:
            filename (str): The path where the YAML file will be saved.
        """
        yaml_content: str = self.to_yaml()
        with open(filename, "w") as file:
            file.write(yaml_content)

    @classmethod
    def load_from_json_file(cls: Type[T], filename: str) -> T:
        """
        Loads a dataclass instance from a JSON file.

        Args:
            filename (str): The path to the JSON file.

        Returns:
            T: An instance of the dataclass.
        """
        with open(filename, "r") as file:
            json_str: str = file.read()
        instance: T = cls.from_json(json_str)
        return instance

    @classmethod
    def load_from_json_url(cls: Type[T], url: str) -> T:
        """
        Loads a dataclass instance from a JSON string obtained from a URL.

        Args:
            url (str): The URL pointing to the JSON data.

        Returns:
            T: An instance of the dataclass.
        """
        json_str: str = cls.read_from_url(url)
        instance: T = cls.from_json(json_str)
        return instance

    def save_to_json_file(self, filename: str, **kwargs):
        """
        Saves the current dataclass instance to a JSON file.

        Args:
            filename (str): The path where the JSON file will be saved.
            **kwargs: Additional keyword arguments for the `to_json` method.
        """
        json_content: str = self.to_json(**kwargs)
        with open(filename, "w") as file:
            file.write(json_content)

    @classmethod
    def read_from_url(cls, url: str) -> str:
        """
        Helper method to fetch content from a URL.
        """
        with urllib.request.urlopen(url) as response:
            if response.status == 200:
                return response.read().decode()
            else:
                raise Exception(f"Unable to load data from URL: {url}")

    @classmethod
    def remove_ignored_values(
        cls,
        value: Any,
        ignore_none: bool = True,
        ignore_underscore: bool = False,
        ignore_empty: bool = True,
    ) -> Any:
        """
        Recursively removes specified types of values from a dictionary or list.
        By default, it removes keys with None values. Optionally, it can also remove keys starting with an underscore.

        Args:
            value: The value to process (dictionary, list, or other).
            ignore_none: Flag to indicate whether None values should be removed.
            ignore_underscore: Flag to indicate whether keys starting with an underscore should be removed.
            ignore_empty: Flag to indicate whether empty collections should be removed.
        """

        def is_valid(v):
            """Check if the value is valid based on the specified flags."""
            if ignore_none and v is None:
                return False
            if ignore_empty:
                if isinstance(v, Mapping) and not v:
                    return False  # Empty dictionary
                if (
                    isinstance(v, Iterable)
                    and not isinstance(v, (str, bytes))
                    and not v
                ):
                    return (
                        False  # Empty list, set, tuple, etc., but not string or bytes
                    )
            return True

        if isinstance(value, Mapping):
            value = {
                k: YamlAble.remove_ignored_values(
                    v, ignore_none, ignore_underscore, ignore_empty
                )
                for k, v in value.items()
                if is_valid(v) and (not ignore_underscore or not k.startswith("_"))
            }
        elif isinstance(value, Iterable) and not isinstance(value, (str, bytes)):
            value = [
                YamlAble.remove_ignored_values(
                    v, ignore_none, ignore_underscore, ignore_empty
                )
                for v in value
                if is_valid(v)
            ]
        return value

    @classmethod
    def from_dict2(cls: Type[T], data: dict) -> T:
        """
        Creates an instance of a dataclass from a dictionary, typically used in deserialization.
        """
        if not data:
            return None
        instance = from_dict(data_class=cls, data=data)
        return instance

from_dict2(data) classmethod

Creates an instance of a dataclass from a dictionary, typically used in deserialization.

Source code in mbusread/yamlable.py 
318
319
320
321
322
323
324
325
326
@classmethod
def from_dict2(cls: Type[T], data: dict) -> T:
    """
    Creates an instance of a dataclass from a dictionary, typically used in deserialization.
    """
    if not data:
        return None
    instance = from_dict(data_class=cls, data=data)
    return instance

from_yaml(yaml_str) classmethod

Deserializes a YAML string to a dataclass instance.

Parameters:

Name Type Description Default
yaml_str str

A string containing YAML formatted data.

 required

Returns:

Name Type Description
T T

An instance of the dataclass.
Source code in mbusread/yamlable.py 
153
154
155
156
157
158
159
160
161
162
163
164
165
166
@classmethod
def from_yaml(cls: Type[T], yaml_str: str) -> T:
    """
    Deserializes a YAML string to a dataclass instance.

    Args:
        yaml_str (str): A string containing YAML formatted data.

    Returns:
        T: An instance of the dataclass.
    """
    data: dict[str, Any] = yaml.safe_load(yaml_str)
    instance: T = cls.from_dict(data)
    return instance

load_from_json_file(filename) classmethod

Loads a dataclass instance from a JSON file.

Parameters:

Name Type Description Default
filename str

The path to the JSON file.

 required

Returns:

Name Type Description
T T

An instance of the dataclass.
Source code in mbusread/yamlable.py 
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
@classmethod
def load_from_json_file(cls: Type[T], filename: str) -> T:
    """
    Loads a dataclass instance from a JSON file.

    Args:
        filename (str): The path to the JSON file.

    Returns:
        T: An instance of the dataclass.
    """
    with open(filename, "r") as file:
        json_str: str = file.read()
    instance: T = cls.from_json(json_str)
    return instance

load_from_json_url(url) classmethod

Loads a dataclass instance from a JSON string obtained from a URL.

Parameters:

Name Type Description Default
url str

The URL pointing to the JSON data.

 required

Returns:

Name Type Description
T T

An instance of the dataclass.
Source code in mbusread/yamlable.py 
226
227
228
229
230
231
232
233
234
235
236
237
238
239
@classmethod
def load_from_json_url(cls: Type[T], url: str) -> T:
    """
    Loads a dataclass instance from a JSON string obtained from a URL.

    Args:
        url (str): The URL pointing to the JSON data.

    Returns:
        T: An instance of the dataclass.
    """
    json_str: str = cls.read_from_url(url)
    instance: T = cls.from_json(json_str)
    return instance

load_from_yaml_file(filename) classmethod

Loads a dataclass instance from a YAML file.

Parameters:

Name Type Description Default
filename str

The path to the YAML file.

 required

Returns:

Name Type Description
T T

An instance of the dataclass.
Source code in mbusread/yamlable.py 
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
@classmethod
def load_from_yaml_file(cls: Type[T], filename: str) -> T:
    """
    Loads a dataclass instance from a YAML file.

    Args:
        filename (str): The path to the YAML file.

    Returns:
        T: An instance of the dataclass.
    """
    with open(filename, "r") as file:
        yaml_str: str = file.read()
    instance: T = cls.from_yaml(yaml_str)
    return instance

load_from_yaml_url(url) classmethod

Loads a dataclass instance from a YAML string obtained from a URL.

Parameters:

Name Type Description Default
url str

The URL pointing to the YAML data.

 required

Returns:

Name Type Description
T T

An instance of the dataclass.
Source code in mbusread/yamlable.py 
184
185
186
187
188
189
190
191
192
193
194
195
196
197
@classmethod
def load_from_yaml_url(cls: Type[T], url: str) -> T:
    """
    Loads a dataclass instance from a YAML string obtained from a URL.

    Args:
        url (str): The URL pointing to the YAML data.

    Returns:
        T: An instance of the dataclass.
    """
    yaml_str: str = cls.read_from_url(url)
    instance: T = cls.from_yaml(yaml_str)
    return instance

read_from_url(url) classmethod

Helper method to fetch content from a URL.

Source code in mbusread/yamlable.py 
253
254
255
256
257
258
259
260
261
262
@classmethod
def read_from_url(cls, url: str) -> str:
    """
    Helper method to fetch content from a URL.
    """
    with urllib.request.urlopen(url) as response:
        if response.status == 200:
            return response.read().decode()
        else:
            raise Exception(f"Unable to load data from URL: {url}")

remove_ignored_values(value, ignore_none=True, ignore_underscore=False, ignore_empty=True) classmethod

Recursively removes specified types of values from a dictionary or list. By default, it removes keys with None values. Optionally, it can also remove keys starting with an underscore.

Parameters:

Name Type Description Default
value Any

The value to process (dictionary, list, or other).

 required
ignore_none bool

Flag to indicate whether None values should be removed.

 True
ignore_underscore bool

Flag to indicate whether keys starting with an underscore should be removed.

 False
ignore_empty bool

Flag to indicate whether empty collections should be removed.

 True
Source code in mbusread/yamlable.py 
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
@classmethod
def remove_ignored_values(
    cls,
    value: Any,
    ignore_none: bool = True,
    ignore_underscore: bool = False,
    ignore_empty: bool = True,
) -> Any:
    """
    Recursively removes specified types of values from a dictionary or list.
    By default, it removes keys with None values. Optionally, it can also remove keys starting with an underscore.

    Args:
        value: The value to process (dictionary, list, or other).
        ignore_none: Flag to indicate whether None values should be removed.
        ignore_underscore: Flag to indicate whether keys starting with an underscore should be removed.
        ignore_empty: Flag to indicate whether empty collections should be removed.
    """

    def is_valid(v):
        """Check if the value is valid based on the specified flags."""
        if ignore_none and v is None:
            return False
        if ignore_empty:
            if isinstance(v, Mapping) and not v:
                return False  # Empty dictionary
            if (
                isinstance(v, Iterable)
                and not isinstance(v, (str, bytes))
                and not v
            ):
                return (
                    False  # Empty list, set, tuple, etc., but not string or bytes
                )
        return True

    if isinstance(value, Mapping):
        value = {
            k: YamlAble.remove_ignored_values(
                v, ignore_none, ignore_underscore, ignore_empty
            )
            for k, v in value.items()
            if is_valid(v) and (not ignore_underscore or not k.startswith("_"))
        }
    elif isinstance(value, Iterable) and not isinstance(value, (str, bytes)):
        value = [
            YamlAble.remove_ignored_values(
                v, ignore_none, ignore_underscore, ignore_empty
            )
            for v in value
            if is_valid(v)
        ]
    return value

represent_literal(dumper, data)

Custom representer for block scalar style for strings.

Source code in mbusread/yamlable.py 
111
112
113
114
115
116
117
def represent_literal(self, dumper: yaml.Dumper, data: str) -> yaml.Node:
    """
    Custom representer for block scalar style for strings.
    """
    if "\n" in data:
        return dumper.represent_scalar("tag:yaml.org,2002:str", data, style="|")
    return dumper.represent_scalar("tag:yaml.org,2002:str", data)

represent_none(_, __)

Custom representer for ignoring None values in the YAML output.

Source code in mbusread/yamlable.py 
105
106
107
108
109
def represent_none(self, _, __) -> yaml.Node:
    """
    Custom representer for ignoring None values in the YAML output.
    """
    return self._yaml_dumper.represent_scalar("tag:yaml.org,2002:null", "")

save_to_json_file(filename, **kwargs)

Saves the current dataclass instance to a JSON file.

Parameters:

Name Type Description Default
filename str

The path where the JSON file will be saved.

 required
**kwargs

Additional keyword arguments for the to_json method.

 {}
Source code in mbusread/yamlable.py 
241
242
243
244
245
246
247
248
249
250
251
def save_to_json_file(self, filename: str, **kwargs):
    """
    Saves the current dataclass instance to a JSON file.

    Args:
        filename (str): The path where the JSON file will be saved.
        **kwargs: Additional keyword arguments for the `to_json` method.
    """
    json_content: str = self.to_json(**kwargs)
    with open(filename, "w") as file:
        file.write(json_content)

save_to_yaml_file(filename)

Saves the current dataclass instance to a YAML file.

Parameters:

Name Type Description Default
filename str

The path where the YAML file will be saved.

 required
Source code in mbusread/yamlable.py 
199
200
201
202
203
204
205
206
207
208
def save_to_yaml_file(self, filename: str):
    """
    Saves the current dataclass instance to a YAML file.

    Args:
        filename (str): The path where the YAML file will be saved.
    """
    yaml_content: str = self.to_yaml()
    with open(filename, "w") as file:
        file.write(yaml_content)

to_yaml(ignore_none=True, ignore_underscore=True, allow_unicode=True, sort_keys=False)

Converts this dataclass object to a YAML string, with options to omit None values and/or underscore-prefixed variables, and using block scalar style for strings.

Parameters:

Name Type Description Default
ignore_none bool

Flag to indicate whether None values should be removed from the YAML output.

 True
ignore_underscore bool

Flag to indicate whether attributes starting with an underscore should be excluded from the YAML output.

 True
allow_unicode bool

Flag to indicate whether to allow unicode characters in the output.

 True
sort_keys bool

Flag to indicate whether to sort the dictionary keys in the output.

 False

Returns:

Type Description
str

A string representation of the dataclass object in YAML format.
Source code in mbusread/yamlable.py 
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
def to_yaml(
    self,
    ignore_none: bool = True,
    ignore_underscore: bool = True,
    allow_unicode: bool = True,
    sort_keys: bool = False,
) -> str:
    """
    Converts this dataclass object to a YAML string, with options to omit None values and/or underscore-prefixed variables,
    and using block scalar style for strings.

    Args:
        ignore_none: Flag to indicate whether None values should be removed from the YAML output.
        ignore_underscore: Flag to indicate whether attributes starting with an underscore should be excluded from the YAML output.
        allow_unicode: Flag to indicate whether to allow unicode characters in the output.
        sort_keys: Flag to indicate whether to sort the dictionary keys in the output.

    Returns:
        A string representation of the dataclass object in YAML format.
    """
    obj_dict = asdict(self)
    self._yaml_setup()
    clean_dict = self.remove_ignored_values(
        obj_dict, ignore_none, ignore_underscore
    )
    yaml_str = yaml.dump(
        clean_dict,
        Dumper=self._yaml_dumper,
        default_flow_style=False,
        allow_unicode=allow_unicode,
        sort_keys=sort_keys,
    )
    return yaml_str

lod_storable(cls)

Decorator to make a class LoDStorable by inheriting from YamlAble. This decorator also ensures the class is a dataclass and has JSON serialization/deserialization capabilities.

Source code in mbusread/yamlable.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def lod_storable(cls):
    """
    Decorator to make a class LoDStorable by
    inheriting from YamlAble.
    This decorator also ensures the class is a
    dataclass and has JSON serialization/deserialization
    capabilities.
    """
    cls = dataclass(cls)  # Apply the @dataclass decorator
    cls = dataclass_json(cls)  # Apply the @dataclass_json decorator

    class LoDStorable(YamlAble, cls):
        """
        decorator class
        """

        __qualname__ = cls.__qualname__
        pass

    LoDStorable.__name__ = cls.__name__
    LoDStorable.__doc__ = cls.__doc__

    return LoDStorable