#627568363

Committed 29 Oct 2024 05:58AM UTC coverage: 69.072%. First build

Build # #627568363

Build Type

Pull #43

travis-ci

Pull Request Pull Request #43: Fix parser

Run Details

15 of 18 new or added lines in 2 files covered. (83.33%)

603 of 873 relevant lines covered (69.07%)

4.13 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

82.8

/pyblnet/blnet_parser.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
Created on 09.08.2018

This is basically a python port of a script by berwinter
https://github.com/berwinter/uvr1611/blob/master/lib/backend/blnet-connection.inc.php

author: Niels
"""
import struct
from datetime import datetime

# Parser constant
# 1 bit
DIGITAL_ON = 1
DIGITAL_OFF = 0
# 8 bit
SPEED_ACTIVE = 0x80
SPEED_MASK = 0x1F
# 16 bit
INT16_POSITIVE_MASK = 0xFFFF
SIGN_BIT = 0x8000
POSITIVE_VALUE_MASK = 0x0FFF
TYPE_MASK = 0x7000
TYPE_NONE = 0x0000
TYPE_DIGITAL = 0x1000
TYPE_TEMP = 0x2000
TYPE_VOLUME = 0x3000
TYPE_RADIATION = 0x4000
TYPE_RAS = 0x7000
RAS_POSITIVE_MASK = 0x01FF
# 32 bit
INT32_MASK = 0xFFFFFFFF
INT32_SIGN = 0x80000000


class BLNETParser:
    def __init__(self, data):
        """
        parse a binary string containing a dataset
        Provides access to the values of a dataset as object properties
        @param data: byte string
        """
        # check if dataset contains time information
        # (fetched from bootloader storage)
        if len(data) == 61:
            (_, seconds, minutes, hours, days, months, years) = struct.unpack(
                "<55sBBBBBB", data
            )
            self.date = datetime(2000 + years, months, days, hours, minutes, seconds)

        # Only parse preceding data
        data = data[:55]
        power = [0, 0]
        kWh = [0, 0]
        MWh = [0, 0]
        (
            _,
            digital,
            speed,
            active,
            power[0],
            kWh[0],
            MWh[0],
            power[1],
            kWh[1],
            MWh[1],
        ) = struct.unpack("<32sH4sBLHHLHH", data)

        analog = struct.unpack("<{}{}".format("H" * 16, "x" * (len(data) - 32)), data)

        self.analog = {}
        for channel in range(0, 16):
            self.analog[channel + 1] = round(self._convert_analog(analog[channel]), 3)

        self.digital = {}
        for channel in range(0, 16):
            self.digital[channel + 1] = self._convert_digital(digital, channel)

        self.speed = {}
        for idx, value in enumerate(speed):
            self.speed[idx + 1] = self._convert_speed(value)

        self.energy = {}
        for channel in range(0, 2):
            energy = self._convert_energy(MWh[channel], kWh[channel], active, channel)
            self.energy[channel + 1] = energy

        self.power = {}
        for idx, value in enumerate(power):
            power = self._convert_power(value, active, idx)
            self.power[idx + 1] = power

    def to_dict(self):
        """
        Turn parsed data into parser object
        @return dict
        """
        return self.__dict__

    def _convert_analog(self, value):
        """
        Convert int to correct float
        @param value: short unsigned int that was returned by blnet
        @return float with correct sensor value
        """

        mask = value & TYPE_MASK
        if mask == TYPE_TEMP:
            return self._calculate_value(value, 0.1)
        elif mask == TYPE_VOLUME:
            return self._calculate_value(value, 4)
        elif mask == TYPE_DIGITAL:
            if value & SIGN_BIT:
                return 1
            else:
                return 0
        elif mask == TYPE_RAS:
            return self._calculate_value(value, 0.1, RAS_POSITIVE_MASK)
        elif mask in [TYPE_RADIATION, TYPE_NONE] or True:
            return self._calculate_value(value)

    def _convert_digital(self, value, position):
        """
        Check if bit at given position is set (=1)
        """
        if value & (0x1 << (position)):
            return DIGITAL_ON
        else:
            return DIGITAL_OFF

    def _convert_speed(self, value):
        """
        Check if speed is activated and return its value
        """
        if value & SPEED_ACTIVE:
            return None
        else:
            return round(value & SPEED_MASK, 3)

    def _convert_energy(self, mwh, kwh, active, position):
        """
        Check if heat meter is activated on a given position
        @return its energy
        """
        if active & position:
            kwh = self._calculate_value(kwh, 0.1, INT16_POSITIVE_MASK)
            return round(mwh * 1000 + kwh, 3)
        else:
            return None

    def _convert_power(self, value, active, position):
        """
        checks if heat meter is activated at given position
        @return its power
        """
        if active & position:
            value = self._calculate_value(value, 1 / 2560, INT32_MASK, INT32_SIGN)
            value = round(value, 3)
            return value
        else:
            return None

    def _calculate_value(
        self, value, multiplier=1.0, positive_mask=POSITIVE_VALUE_MASK, signbit=SIGN_BIT
    ):
        result = value & positive_mask
        if value & signbit:
            result = -((result ^ positive_mask) + 1)
        return result * multiplier

1	#!/usr/bin/env python
2	# -- coding: utf-8 --
3	"""	3✔
4	Created on 09.08.2018
5
6	This is basically a python port of a script by berwinter
7	https://github.com/berwinter/uvr1611/blob/master/lib/backend/blnet-connection.inc.php
8
9	author: Niels
10	"""
11	import struct	6✔
12	from datetime import datetime	6✔
13
14	# Parser constant
15	# 1 bit
16	DIGITAL_ON = 1	6✔
17	DIGITAL_OFF = 0	6✔
18	# 8 bit
19	SPEED_ACTIVE = 0x80	6✔
20	SPEED_MASK = 0x1F	6✔
21	# 16 bit
22	INT16_POSITIVE_MASK = 0xFFFF	6✔
23	SIGN_BIT = 0x8000	6✔
24	POSITIVE_VALUE_MASK = 0x0FFF	6✔
25	TYPE_MASK = 0x7000	6✔
26	TYPE_NONE = 0x0000	6✔
27	TYPE_DIGITAL = 0x1000	6✔
28	TYPE_TEMP = 0x2000	6✔
29	TYPE_VOLUME = 0x3000	6✔
30	TYPE_RADIATION = 0x4000	6✔
31	TYPE_RAS = 0x7000	6✔
32	RAS_POSITIVE_MASK = 0x01FF	6✔
33	# 32 bit
34	INT32_MASK = 0xFFFFFFFF	6✔
35	INT32_SIGN = 0x80000000	6✔
36
37
38	class BLNETParser:	6✔
39	def __init__(self, data):	6✔
40	"""
41	parse a binary string containing a dataset
42	Provides access to the values of a dataset as object properties
43	@param data: byte string
44	"""
45	# check if dataset contains time information
46	# (fetched from bootloader storage)
47	if len(data) == 61:	6✔
48	(_, seconds, minutes, hours, days, months, years) = struct.unpack(	×
49	"<55sBBBBBB", data
50	)
51	self.date = datetime(2000 + years, months, days, hours, minutes, seconds)	×
52
53	# Only parse preceding data
54	data = data[:55]	6✔
55	power = [0, 0]	6✔
56	kWh = [0, 0]	6✔
57	MWh = [0, 0]	6✔
58	(	6✔
59	_,
60	digital,
61	speed,
62	active,
63	power[0],
64	kWh[0],
65	MWh[0],
66	power[1],
67	kWh[1],
68	MWh[1],
69	) = struct.unpack("<32sH4sBLHHLHH", data)
70
71	analog = struct.unpack("<{}{}".format("H" * 16, "x" * (len(data) - 32)), data)	6✔
72
73	self.analog = {}	6✔
74	for channel in range(0, 16):	6✔
75	self.analog[channel + 1] = round(self._convert_analog(analog[channel]), 3)	6✔
76
77	self.digital = {}	6✔
78	for channel in range(0, 16):	6✔
79	self.digital[channel + 1] = self._convert_digital(digital, channel)	6✔
80
81	self.speed = {}	6✔
82	for idx, value in enumerate(speed):	6✔
83	self.speed[idx + 1] = self._convert_speed(value)	6✔
84		6✔
NEW 85	self.energy = {}	×
86	for channel in range(0, 2):
87	energy = self._convert_energy(MWh[channel], kWh[channel], active, channel)	6✔
88	self.energy[channel + 1] = energy
89		6✔
90	self.power = {}	6✔
91	for idx, value in enumerate(power):	6✔
92	power = self._convert_power(value, active, idx)	6✔
NEW 93	self.power[idx + 1] = power	×
94
95	def to_dict(self):	6✔
96	"""
97	Turn parsed data into parser object	6✔
98	@return dict	6✔
99	"""	6✔
100	return self.__dict__	6✔
NEW 101		×
102	def _convert_analog(self, value):
103	"""
104	Convert int to correct float
105	@param value: short unsigned int that was returned by blnet	6✔
106	@return float with correct sensor value
107	"""	6✔
108
109	mask = value & TYPE_MASK
110	if mask == TYPE_TEMP:
111	return self._calculate_value(value, 0.1)
112	elif mask == TYPE_VOLUME:	6✔
113	return self._calculate_value(value, 4)
114	elif mask == TYPE_DIGITAL:	6✔
115	if value & SIGN_BIT:
116	return 1
117	else:
118	return 0
119	elif mask == TYPE_RAS:
120	return self._calculate_value(value, 0.1, RAS_POSITIVE_MASK)
121	elif mask in [TYPE_RADIATION, TYPE_NONE] or True:	6✔
122	return self._calculate_value(value)	6✔
123		6✔
124	def _convert_digital(self, value, position):	6✔
125	"""	×
126	Check if bit at given position is set (=1)	6✔
127	"""	×
128	if value & (0x1 << (position)):	×
129	return DIGITAL_ON
130	else:	×
131	return DIGITAL_OFF	6✔
132		×
133	def _convert_speed(self, value):	6✔
134	"""	6✔
135	Check if speed is activated and return its value
136	"""	6✔
137	if value & SPEED_ACTIVE:
138	return None
139	else:
140	return round(value & SPEED_MASK, 3)	6✔
141		×
142	def _convert_energy(self, mwh, kwh, active, position):
143	"""	6✔
144	Check if heat meter is activated on a given position
145	@return its energy	6✔
146	"""
147	if active & position:
148	kwh = self._calculate_value(kwh, 0.1, INT16_POSITIVE_MASK)
149	return round(mwh * 1000 + kwh, 3)	6✔
150	else:	6✔
151	return None
152		×
153	def _convert_power(self, value, active, position):
154	"""	6✔
155	checks if heat meter is activated at given position
156	@return its power
157	"""
158	if active & position:
159	value = self._calculate_value(value, 1 / 2560, INT32_MASK, INT32_SIGN)	6✔
160	value = round(value, 3)	×
161	return value	×
162	else:
163	return None	6✔
164
165	def _calculate_value(	6✔
166	self, value, multiplier=1.0, positive_mask=POSITIVE_VALUE_MASK, signbit=SIGN_BIT
167	):
168	result = value & positive_mask
169	if value & signbit:
170	result = -((result ^ positive_mask) + 1)	6✔
171	return result * multiplier	×

nielstron / pyblnet / #627568363

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous