[libsigrokdecode.git] / decoders / am230x / pd.py

##
## This file is part of the libsigrokdecode project.
##
## Copyright (C) 2014 Johannes Roemer <jroemer@physik.uni-wuerzburg.de>
##
## This program is free software; you can redistribute it and/or modify
## it under the terms of the GNU General Public License as published by
## the Free Software Foundation; either version 2 of the License, or
## (at your option) any later version.
##
## This program is distributed in the hope that it will be useful,
## but WITHOUT ANY WARRANTY; without even the implied warranty of
## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
## GNU General Public License for more details.
##
## You should have received a copy of the GNU General Public License
## along with this program; if not, see <http://www.gnu.org/licenses/>.
##

import sigrokdecode as srd

# Define valid timing values (in microseconds).
timing = {
    'START LOW'     : {'min': 750, 'max': 25000},
    'START HIGH'    : {'min': 10, 'max': 10000},
    'RESPONSE LOW'  : {'min': 50, 'max': 90},
    'RESPONSE HIGH' : {'min': 50, 'max': 90},
    'BIT LOW'       : {'min': 45, 'max': 90},
    'BIT 0 HIGH'    : {'min': 20, 'max': 35},
    'BIT 1 HIGH'    : {'min': 65, 'max': 80},
}

class SamplerateError(Exception):
    pass

class Decoder(srd.Decoder):
    api_version = 3
    id = 'am230x'
    name = 'AM230x/DHTxx/RHTxx'
    longname = 'Aosong AM230x/DHTxx/RHTxx'
    desc = 'Aosong AM230x/DHTxx/RHTxx humidity/temperature sensor protocol.'
    license = 'gplv2+'
    inputs = ['logic']
    outputs = ['am230x']
    channels = (
        {'id': 'sda', 'name': 'SDA', 'desc': 'Single wire serial data line'},
    )
    options = (
        {'id': 'device', 'desc': 'Device type',
            'default': 'am230x', 'values': ('am230x/rht', 'dht11')},
    )
    annotations = (
        ('start', 'Start'),
        ('response', 'Response'),
        ('bit', 'Bit'),
        ('end', 'End'),
        ('byte', 'Byte'),
        ('humidity', 'Relative humidity in percent'),
        ('temperature', 'Temperature in degrees Celsius'),
        ('checksum', 'Checksum'),
    )
    annotation_rows = (
        ('bits', 'Bits', (0, 1, 2, 3)),
        ('bytes', 'Bytes', (4,)),
        ('results', 'Results', (5, 6, 7)),
    )

    def putfs(self, data):
        self.put(self.fall, self.samplenum, self.out_ann, data)

    def putb(self, data):
        self.put(self.bytepos[-1], self.samplenum, self.out_ann, data)

    def putv(self, data):
        self.put(self.bytepos[-2], self.samplenum, self.out_ann, data)

    def reset_variables(self):
        self.state = 'WAIT FOR START LOW'
        self.fall = 0
        self.rise = 0
        self.bits = []
        self.bytepos = []

    def is_valid(self, name):
        dt = 0
        if name.endswith('LOW'):
            dt = self.samplenum - self.fall
        elif name.endswith('HIGH'):
            dt = self.samplenum - self.rise
        if dt >= self.cnt[name]['min'] and dt <= self.cnt[name]['max']:
            return True
        return False

    def bits2num(self, bitlist):
        number = 0
        for i in range(len(bitlist)):
            number += bitlist[-1 - i] * 2**i
        return number

    def calculate_humidity(self, bitlist):
        h = 0
        if self.options['device'] == 'dht11':
            h = self.bits2num(bitlist[0:8])
        else:
            h = self.bits2num(bitlist) / 10
        return h

    def calculate_temperature(self, bitlist):
        t = 0
        if self.options['device'] == 'dht11':
            t = self.bits2num(bitlist[0:8])
        else:
            t = self.bits2num(bitlist[1:]) / 10
            if bitlist[0] == 1:
                t = -t
        return t

    def calculate_checksum(self, bitlist):
        checksum = 0
        for i in range(8, len(bitlist) + 1, 8):
            checksum += self.bits2num(bitlist[i-8:i])
        return checksum % 256

    def __init__(self):
        self.reset()

    def reset(self):
        self.samplerate = None
        self.reset_variables()

    def start(self):
        self.out_ann = self.register(srd.OUTPUT_ANN)

    def metadata(self, key, value):
        if key != srd.SRD_CONF_SAMPLERATE:
            return
        self.samplerate = value
        # Convert microseconds to sample counts.
        self.cnt = {}
        for e in timing:
            self.cnt[e] = {}
            for t in timing[e]:
                self.cnt[e][t] = timing[e][t] * self.samplerate / 1000000

    def handle_byte(self, bit):
        self.bits.append(bit)
        self.putfs([2, ['Bit: %d' % bit, '%d' % bit]])
        self.fall = self.samplenum
        self.state = 'WAIT FOR BIT HIGH'
        if len(self.bits) % 8 == 0:
            byte = self.bits2num(self.bits[-8:])
            self.putb([4, ['Byte: %#04x' % byte, '%#04x' % byte]])
            if len(self.bits) == 16:
                h = self.calculate_humidity(self.bits[-16:])
                self.putv([5, ['Humidity: %.1f %%' % h, 'RH = %.1f %%' % h]])
            elif len(self.bits) == 32:
                t = self.calculate_temperature(self.bits[-16:])
                self.putv([6, ['Temperature: %.1f °C' % t, 'T = %.1f °C' % t]])
            elif len(self.bits) == 40:
                parity = self.bits2num(self.bits[-8:])
                if parity == self.calculate_checksum(self.bits[0:32]):
                    self.putb([7, ['Checksum: OK', 'OK']])
                else:
                    self.putb([7, ['Checksum: not OK', 'NOK']])
                self.state = 'WAIT FOR END'
            self.bytepos.append(self.samplenum)

    def decode(self):
        if not self.samplerate:
            raise SamplerateError('Cannot decode without samplerate.')
        while True:
            # State machine.
            if self.state == 'WAIT FOR START LOW':
                self.wait({0: 'f'})
                self.fall = self.samplenum
                self.state = 'WAIT FOR START HIGH'
            elif self.state == 'WAIT FOR START HIGH':
                self.wait({0: 'r'})
                if self.is_valid('START LOW'):
                    self.rise = self.samplenum
                    self.state = 'WAIT FOR RESPONSE LOW'
                else:
                    self.reset_variables()
            elif self.state == 'WAIT FOR RESPONSE LOW':
                self.wait({0: 'f'})
                if self.is_valid('START HIGH'):
                    self.putfs([0, ['Start', 'S']])
                    self.fall = self.samplenum
                    self.state = 'WAIT FOR RESPONSE HIGH'
                else:
                    self.reset_variables()
            elif self.state == 'WAIT FOR RESPONSE HIGH':
                self.wait({0: 'r'})
                if self.is_valid('RESPONSE LOW'):
                    self.rise = self.samplenum
                    self.state = 'WAIT FOR FIRST BIT'
                else:
                    self.reset_variables()
            elif self.state == 'WAIT FOR FIRST BIT':
                self.wait({0: 'f'})
                if self.is_valid('RESPONSE HIGH'):
                    self.putfs([1, ['Response', 'R']])
                    self.fall = self.samplenum
                    self.bytepos.append(self.samplenum)
                    self.state = 'WAIT FOR BIT HIGH'
                else:
                    self.reset_variables()
            elif self.state == 'WAIT FOR BIT HIGH':
                self.wait({0: 'r'})
                if self.is_valid('BIT LOW'):
                    self.rise = self.samplenum
                    self.state = 'WAIT FOR BIT LOW'
                else:
                    self.reset_variables()
            elif self.state == 'WAIT FOR BIT LOW':
                self.wait({0: 'f'})
                if self.is_valid('BIT 0 HIGH'):
                    bit = 0
                elif self.is_valid('BIT 1 HIGH'):
                    bit = 1
                else:
                    self.reset_variables()
                    continue
                self.handle_byte(bit)
            elif self.state == 'WAIT FOR END':
                self.wait({0: 'r'})
                self.putfs([3, ['End', 'E']])
                self.reset_variables()
Commit	Line	Data
b1f2b85b JR	1	##
	2	## This file is part of the libsigrokdecode project.
	3	##
	4	## Copyright (C) 2014 Johannes Roemer <jroemer@physik.uni-wuerzburg.de>
	5	##
	6	## This program is free software; you can redistribute it and/or modify
	7	## it under the terms of the GNU General Public License as published by
	8	## the Free Software Foundation; either version 2 of the License, or
	9	## (at your option) any later version.
	10	##
	11	## This program is distributed in the hope that it will be useful,
	12	## but WITHOUT ANY WARRANTY; without even the implied warranty of
	13	## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
	14	## GNU General Public License for more details.
	15	##
	16	## You should have received a copy of the GNU General Public License
4539e9ca	17	## along with this program; if not, see <http://www.gnu.org/licenses/>.
b1f2b85b JR	18	##
	19
	20	import sigrokdecode as srd
	21
	22	# Define valid timing values (in microseconds).
	23	timing = {
	24	'START LOW' : {'min': 750, 'max': 25000},
	25	'START HIGH' : {'min': 10, 'max': 10000},
ed758f09 UH	26	'RESPONSE LOW' : {'min': 50, 'max': 90},
ed758f09 UH	27	'RESPONSE HIGH' : {'min': 50, 'max': 90},
b1f2b85b JR	28	'BIT LOW' : {'min': 45, 'max': 90},
	29	'BIT 0 HIGH' : {'min': 20, 'max': 35},
	30	'BIT 1 HIGH' : {'min': 65, 'max': 80},
	31	}
	32
	33	class SamplerateError(Exception):
	34	pass
	35
	36	class Decoder(srd.Decoder):
b6bac00c	37	api_version = 3
b1f2b85b	38	id = 'am230x'
e2ba53bb KP	39	name = 'AM230x/DHTxx/RHTxx'
	40	longname = 'Aosong AM230x/DHTxx/RHTxx'
	41	desc = 'Aosong AM230x/DHTxx/RHTxx humidity/temperature sensor protocol.'
b1f2b85b JR	42	license = 'gplv2+'
	43	inputs = ['logic']
	44	outputs = ['am230x']
	45	channels = (
	46	{'id': 'sda', 'name': 'SDA', 'desc': 'Single wire serial data line'},
	47	)
	48	options = (
cda1352a	49	{'id': 'device', 'desc': 'Device type',
e2ba53bb	50	'default': 'am230x', 'values': ('am230x/rht', 'dht11')},
b1f2b85b JR	51	)
	52	annotations = (
	53	('start', 'Start'),
	54	('response', 'Response'),
	55	('bit', 'Bit'),
	56	('end', 'End'),
	57	('byte', 'Byte'),
	58	('humidity', 'Relative humidity in percent'),
	59	('temperature', 'Temperature in degrees Celsius'),
	60	('checksum', 'Checksum'),
	61	)
	62	annotation_rows = (
	63	('bits', 'Bits', (0, 1, 2, 3)),
	64	('bytes', 'Bytes', (4,)),
	65	('results', 'Results', (5, 6, 7)),
	66	)
	67
	68	def putfs(self, data):
	69	self.put(self.fall, self.samplenum, self.out_ann, data)
	70
	71	def putb(self, data):
	72	self.put(self.bytepos[-1], self.samplenum, self.out_ann, data)
	73
	74	def putv(self, data):
	75	self.put(self.bytepos[-2], self.samplenum, self.out_ann, data)
	76
779c540d	77	def reset_variables(self):
b1f2b85b	78	self.state = 'WAIT FOR START LOW'
b1f2b85b JR	79	self.fall = 0
	80	self.rise = 0
	81	self.bits = []
	82	self.bytepos = []
	83
	84	def is_valid(self, name):
	85	dt = 0
	86	if name.endswith('LOW'):
	87	dt = self.samplenum - self.fall
	88	elif name.endswith('HIGH'):
	89	dt = self.samplenum - self.rise
	90	if dt >= self.cnt[name]['min'] and dt <= self.cnt[name]['max']:
	91	return True
	92	return False
	93
	94	def bits2num(self, bitlist):
	95	number = 0
47405f47 UH	96	for i in range(len(bitlist)):
47405f47 UH	97	number += bitlist[-1 - i] * 2**i
b1f2b85b JR	98	return number
	99
	100	def calculate_humidity(self, bitlist):
	101	h = 0
cda1352a	102	if self.options['device'] == 'dht11':
b1f2b85b JR	103	h = self.bits2num(bitlist[0:8])
	104	else:
	105	h = self.bits2num(bitlist) / 10
	106	return h
	107
	108	def calculate_temperature(self, bitlist):
	109	t = 0
cda1352a	110	if self.options['device'] == 'dht11':
b1f2b85b JR	111	t = self.bits2num(bitlist[0:8])
	112	else:
	113	t = self.bits2num(bitlist[1:]) / 10
	114	if bitlist[0] == 1:
	115	t = -t
	116	return t
	117
	118	def calculate_checksum(self, bitlist):
	119	checksum = 0
47405f47	120	for i in range(8, len(bitlist) + 1, 8):
b1f2b85b JR	121	checksum += self.bits2num(bitlist[i-8:i])
	122	return checksum % 256
	123
92b7b49f	124	def __init__(self):
10aeb8ea GS	125	self.reset()
	126
	127	def reset(self):
b1f2b85b	128	self.samplerate = None
779c540d	129	self.reset_variables()
b1f2b85b JR	130
	131	def start(self):
	132	self.out_ann = self.register(srd.OUTPUT_ANN)
	133
	134	def metadata(self, key, value):
	135	if key != srd.SRD_CONF_SAMPLERATE:
	136	return
	137	self.samplerate = value
	138	# Convert microseconds to sample counts.
	139	self.cnt = {}
	140	for e in timing:
	141	self.cnt[e] = {}
	142	for t in timing[e]:
	143	self.cnt[e][t] = timing[e][t] * self.samplerate / 1000000
	144
0b6bc7f9 UH	145	def handle_byte(self, bit):
	146	self.bits.append(bit)
	147	self.putfs([2, ['Bit: %d' % bit, '%d' % bit]])
	148	self.fall = self.samplenum
	149	self.state = 'WAIT FOR BIT HIGH'
	150	if len(self.bits) % 8 == 0:
	151	byte = self.bits2num(self.bits[-8:])
	152	self.putb([4, ['Byte: %#04x' % byte, '%#04x' % byte]])
	153	if len(self.bits) == 16:
	154	h = self.calculate_humidity(self.bits[-16:])
	155	self.putv([5, ['Humidity: %.1f %%' % h, 'RH = %.1f %%' % h]])
	156	elif len(self.bits) == 32:
	157	t = self.calculate_temperature(self.bits[-16:])
	158	self.putv([6, ['Temperature: %.1f °C' % t, 'T = %.1f °C' % t]])
	159	elif len(self.bits) == 40:
	160	parity = self.bits2num(self.bits[-8:])
	161	if parity == self.calculate_checksum(self.bits[0:32]):
	162	self.putb([7, ['Checksum: OK', 'OK']])
	163	else:
	164	self.putb([7, ['Checksum: not OK', 'NOK']])
	165	self.state = 'WAIT FOR END'
	166	self.bytepos.append(self.samplenum)
	167
b6bac00c	168	def decode(self):
b1f2b85b JR	169	if not self.samplerate:
b1f2b85b JR	170	raise SamplerateError('Cannot decode without samplerate.')
b6bac00c	171	while True:
b1f2b85b JR	172	# State machine.
b1f2b85b JR	173	if self.state == 'WAIT FOR START LOW':
b6bac00c	174	self.wait({0: 'f'})
b1f2b85b JR	175	self.fall = self.samplenum
	176	self.state = 'WAIT FOR START HIGH'
	177	elif self.state == 'WAIT FOR START HIGH':
b6bac00c	178	self.wait({0: 'r'})
b1f2b85b JR	179	if self.is_valid('START LOW'):
	180	self.rise = self.samplenum
	181	self.state = 'WAIT FOR RESPONSE LOW'
	182	else:
779c540d	183	self.reset_variables()
b1f2b85b	184	elif self.state == 'WAIT FOR RESPONSE LOW':
b6bac00c	185	self.wait({0: 'f'})
b1f2b85b JR	186	if self.is_valid('START HIGH'):
	187	self.putfs([0, ['Start', 'S']])
	188	self.fall = self.samplenum
	189	self.state = 'WAIT FOR RESPONSE HIGH'
	190	else:
779c540d	191	self.reset_variables()
b1f2b85b	192	elif self.state == 'WAIT FOR RESPONSE HIGH':
b6bac00c	193	self.wait({0: 'r'})
b1f2b85b JR	194	if self.is_valid('RESPONSE LOW'):
	195	self.rise = self.samplenum
	196	self.state = 'WAIT FOR FIRST BIT'
	197	else:
779c540d	198	self.reset_variables()
b1f2b85b	199	elif self.state == 'WAIT FOR FIRST BIT':
b6bac00c	200	self.wait({0: 'f'})
b1f2b85b JR	201	if self.is_valid('RESPONSE HIGH'):
	202	self.putfs([1, ['Response', 'R']])
	203	self.fall = self.samplenum
	204	self.bytepos.append(self.samplenum)
	205	self.state = 'WAIT FOR BIT HIGH'
	206	else:
779c540d	207	self.reset_variables()
b1f2b85b	208	elif self.state == 'WAIT FOR BIT HIGH':
b6bac00c	209	self.wait({0: 'r'})
b1f2b85b JR	210	if self.is_valid('BIT LOW'):
	211	self.rise = self.samplenum
	212	self.state = 'WAIT FOR BIT LOW'
	213	else:
779c540d	214	self.reset_variables()
b1f2b85b	215	elif self.state == 'WAIT FOR BIT LOW':
b6bac00c	216	self.wait({0: 'f'})
b1f2b85b JR	217	if self.is_valid('BIT 0 HIGH'):
	218	bit = 0
	219	elif self.is_valid('BIT 1 HIGH'):
	220	bit = 1
	221	else:
779c540d	222	self.reset_variables()
b1f2b85b	223	continue
0b6bc7f9	224	self.handle_byte(bit)
b1f2b85b	225	elif self.state == 'WAIT FOR END':
b6bac00c	226	self.wait({0: 'r'})
b1f2b85b	227	self.putfs([3, ['End', 'E']])
779c540d	228	self.reset_variables()