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

##
## This file is part of the libsigrokdecode project.
##
## Copyright (C) 2016 Fabian J. Stumpf <sigrok@fabianstumpf.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

class Decoder(srd.Decoder):
    api_version = 3
    id = 'dmx512'
    name = 'DMX512'
    longname = 'Digital MultipleX 512'
    desc = 'Digital MultipleX 512 (DMX512) lighting protocol.'
    license = 'gplv2+'
    inputs = ['logic']
    outputs = []
    tags = ['Embedded/industrial', 'Lighting']
    channels = (
        {'id': 'dmx', 'name': 'DMX data', 'desc': 'Any DMX data line'},
    )
    annotations = (
        ('bit', 'Bit'),
        ('break', 'Break'),
        ('mab', 'Mark after break'),
        ('startbit', 'Start bit'),
        ('stopbits', 'Stop bit'),
        ('startcode', 'Start code'),
        ('channel', 'Channel'),
        ('interframe', 'Interframe'),
        ('interpacket', 'Interpacket'),
        ('data', 'Data'),
        ('error', 'Error'),
    )
    annotation_rows = (
        ('name', 'Logical', (1, 2, 5, 6, 7, 8)),
        ('data', 'Data', (9,)),
        ('bits', 'Bits', (0, 3, 4)),
        ('errors', 'Errors', (10,)),
    )

    def __init__(self):
        self.reset()

    def reset(self):
        self.samplerate = None
        self.sample_usec = None
        self.run_start = -1
        self.run_bit = 0
        self.state = 'FIND BREAK'

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

    def metadata(self, key, value):
        if key == srd.SRD_CONF_SAMPLERATE:
            self.samplerate = value
            self.sample_usec = 1 / value * 1000000
            self.skip_per_bit = int(4 / self.sample_usec)

    def putr(self, data):
        self.put(self.run_start, self.samplenum, self.out_ann, data)

    def decode(self):
        if not self.samplerate:
            raise SamplerateError('Cannot decode without samplerate.')
        while True:
            # Seek for an interval with no state change with a length between
            # 88 and 1000000 us (BREAK).
            if self.state == 'FIND BREAK':
                (dmx,) = self.wait({0: 'h' if self.run_bit == 0 else 'l'})
                runlen = (self.samplenum - self.run_start) * self.sample_usec
                if runlen > 88 and runlen < 1000000:
                    self.putr([1, ['Break']])
                    self.bit_break = self.run_bit
                    self.state = 'MARK MAB'
                    self.channel = 0
                elif runlen >= 1000000:
                    # Error condition.
                    self.putr([10, ['Invalid break length']])
                self.run_bit = dmx
                self.run_start = self.samplenum
            # Directly following the BREAK is the MARK AFTER BREAK.
            elif self.state == 'MARK MAB':
                (dmx,) = self.wait({0: 'h' if self.run_bit == 0 else 'l'})
                self.putr([2, ['MAB']])
                self.state = 'READ BYTE'
                self.channel = 0
                self.bit = 0
                self.aggreg = dmx
                self.run_start = self.samplenum
            # Mark and read a single transmitted byte
            # (start bit, 8 data bits, 2 stop bits).
            elif self.state == 'READ BYTE':
                (dmx,) = self.wait()
                self.next_sample = self.run_start + (self.bit + 1) * self.skip_per_bit
                self.aggreg += dmx
                if self.samplenum != self.next_sample:
                    continue
                bit_value = 0 if round(self.aggreg/self.skip_per_bit) == self.bit_break else 1

                if self.bit == 0:
                    self.byte = 0
                    self.putr([3, ['Start bit']])
                    if bit_value != 0:
                        # (Possibly) invalid start bit, mark but don't fail.
                        self.put(self.samplenum, self.samplenum,
                                 self.out_ann, [10, ['Invalid start bit']])
                elif self.bit >= 9:
                    self.put(self.samplenum - self.skip_per_bit,
                        self.samplenum, self.out_ann, [4, ['Stop bit']])
                    if bit_value != 1:
                        # Invalid stop bit, mark.
                        self.put(self.samplenum, self.samplenum,
                            self.out_ann, [10, ['Invalid stop bit']])
                        if self.bit == 10:
                            # On invalid 2nd stop bit, search for new break.
                            self.run_bit = dmx
                            self.state = 'FIND BREAK'
                else:
                    # Label and process one bit.
                    self.put(self.samplenum - self.skip_per_bit,
                        self.samplenum, self.out_ann, [0, [str(bit_value)]])
                    self.byte |= bit_value << (self.bit - 1)

                # Label a complete byte.
                if self.bit == 10:
                    if self.channel == 0:
                        d = [5, ['Start code']]
                    else:
                        d = [6, ['Channel ' + str(self.channel)]]
                    self.put(self.run_start, self.next_sample, self.out_ann, d)
                    self.put(self.run_start + self.skip_per_bit,
                        self.next_sample - 2 * self.skip_per_bit,
                        self.out_ann, [9, [str(self.byte) + ' / ' + \
                        str(hex(self.byte))]])
                    # Continue by scanning the IFT.
                    self.channel += 1
                    self.run_start = self.samplenum
                    self.run_bit = dmx
                    self.state = 'MARK IFT'

                self.aggreg = dmx
                self.bit += 1
            # Mark the INTERFRAME-TIME between bytes / INTERPACKET-TIME between packets.
            elif self.state == 'MARK IFT':
                (dmx,) = self.wait({0: 'h' if self.run_bit == 0 else 'l'})
                if self.channel > 512:
                    self.putr([8, ['Interpacket']])
                    self.state = 'FIND BREAK'
                    self.run_bit = dmx
                    self.run_start = self.samplenum
                else:
                    self.putr([7, ['Interframe']])
                    self.state = 'READ BYTE'
                    self.bit = 0
                    self.run_start = self.samplenum
Commit	Line	Data
	1	##
	2	## This file is part of the libsigrokdecode project.
	3	##
	4	## Copyright (C) 2016 Fabian J. Stumpf <sigrok@fabianstumpf.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
	17	## along with this program; if not, see <http://www.gnu.org/licenses/>.
	18	##
	19
	20	import sigrokdecode as srd
	21
	22	class Decoder(srd.Decoder):
	23	api_version = 3
	24	id = 'dmx512'
	25	name = 'DMX512'
	26	longname = 'Digital MultipleX 512'
	27	desc = 'Digital MultipleX 512 (DMX512) lighting protocol.'
	28	license = 'gplv2+'
	29	inputs = ['logic']
	30	outputs = []
	31	tags = ['Embedded/industrial', 'Lighting']
	32	channels = (
	33	{'id': 'dmx', 'name': 'DMX data', 'desc': 'Any DMX data line'},
	34	)
	35	annotations = (
	36	('bit', 'Bit'),
	37	('break', 'Break'),
	38	('mab', 'Mark after break'),
	39	('startbit', 'Start bit'),
	40	('stopbits', 'Stop bit'),
	41	('startcode', 'Start code'),
	42	('channel', 'Channel'),
	43	('interframe', 'Interframe'),
	44	('interpacket', 'Interpacket'),
	45	('data', 'Data'),
	46	('error', 'Error'),
	47	)
	48	annotation_rows = (
	49	('name', 'Logical', (1, 2, 5, 6, 7, 8)),
	50	('data', 'Data', (9,)),
	51	('bits', 'Bits', (0, 3, 4)),
	52	('errors', 'Errors', (10,)),
	53	)
	54
	55	def __init__(self):
	56	self.reset()
	57
	58	def reset(self):
	59	self.samplerate = None
	60	self.sample_usec = None
	61	self.run_start = -1
	62	self.run_bit = 0
	63	self.state = 'FIND BREAK'
	64
	65	def start(self):
	66	self.out_ann = self.register(srd.OUTPUT_ANN)
	67
	68	def metadata(self, key, value):
	69	if key == srd.SRD_CONF_SAMPLERATE:
	70	self.samplerate = value
	71	self.sample_usec = 1 / value * 1000000
	72	self.skip_per_bit = int(4 / self.sample_usec)
	73
	74	def putr(self, data):
	75	self.put(self.run_start, self.samplenum, self.out_ann, data)
	76
	77	def decode(self):
	78	if not self.samplerate:
	79	raise SamplerateError('Cannot decode without samplerate.')
	80	while True:
	81	# Seek for an interval with no state change with a length between
	82	# 88 and 1000000 us (BREAK).
	83	if self.state == 'FIND BREAK':
	84	(dmx,) = self.wait({0: 'h' if self.run_bit == 0 else 'l'})
	85	runlen = (self.samplenum - self.run_start) * self.sample_usec
	86	if runlen > 88 and runlen < 1000000:
	87	self.putr([1, ['Break']])
	88	self.bit_break = self.run_bit
	89	self.state = 'MARK MAB'
	90	self.channel = 0
	91	elif runlen >= 1000000:
	92	# Error condition.
	93	self.putr([10, ['Invalid break length']])
	94	self.run_bit = dmx
	95	self.run_start = self.samplenum
	96	# Directly following the BREAK is the MARK AFTER BREAK.
	97	elif self.state == 'MARK MAB':
	98	(dmx,) = self.wait({0: 'h' if self.run_bit == 0 else 'l'})
	99	self.putr([2, ['MAB']])
	100	self.state = 'READ BYTE'
	101	self.channel = 0
	102	self.bit = 0
	103	self.aggreg = dmx
	104	self.run_start = self.samplenum
	105	# Mark and read a single transmitted byte
	106	# (start bit, 8 data bits, 2 stop bits).
	107	elif self.state == 'READ BYTE':
	108	(dmx,) = self.wait()
	109	self.next_sample = self.run_start + (self.bit + 1) * self.skip_per_bit
	110	self.aggreg += dmx
	111	if self.samplenum != self.next_sample:
	112	continue
	113	bit_value = 0 if round(self.aggreg/self.skip_per_bit) == self.bit_break else 1
	114
	115	if self.bit == 0:
	116	self.byte = 0
	117	self.putr([3, ['Start bit']])
	118	if bit_value != 0:
	119	# (Possibly) invalid start bit, mark but don't fail.
	120	self.put(self.samplenum, self.samplenum,
	121	self.out_ann, [10, ['Invalid start bit']])
	122	elif self.bit >= 9:
	123	self.put(self.samplenum - self.skip_per_bit,
	124	self.samplenum, self.out_ann, [4, ['Stop bit']])
	125	if bit_value != 1:
	126	# Invalid stop bit, mark.
	127	self.put(self.samplenum, self.samplenum,
	128	self.out_ann, [10, ['Invalid stop bit']])
	129	if self.bit == 10:
	130	# On invalid 2nd stop bit, search for new break.
	131	self.run_bit = dmx
	132	self.state = 'FIND BREAK'
	133	else:
	134	# Label and process one bit.
	135	self.put(self.samplenum - self.skip_per_bit,
	136	self.samplenum, self.out_ann, [0, [str(bit_value)]])
	137	self.byte \|= bit_value << (self.bit - 1)
	138
	139	# Label a complete byte.
	140	if self.bit == 10:
	141	if self.channel == 0:
	142	d = [5, ['Start code']]
	143	else:
	144	d = [6, ['Channel ' + str(self.channel)]]
	145	self.put(self.run_start, self.next_sample, self.out_ann, d)
	146	self.put(self.run_start + self.skip_per_bit,
	147	self.next_sample - 2 * self.skip_per_bit,
	148	self.out_ann, [9, [str(self.byte) + ' / ' + \
	149	str(hex(self.byte))]])
	150	# Continue by scanning the IFT.
	151	self.channel += 1
	152	self.run_start = self.samplenum
	153	self.run_bit = dmx
	154	self.state = 'MARK IFT'
	155
	156	self.aggreg = dmx
	157	self.bit += 1
	158	# Mark the INTERFRAME-TIME between bytes / INTERPACKET-TIME between packets.
	159	elif self.state == 'MARK IFT':
	160	(dmx,) = self.wait({0: 'h' if self.run_bit == 0 else 'l'})
	161	if self.channel > 512:
	162	self.putr([8, ['Interpacket']])
	163	self.state = 'FIND BREAK'
	164	self.run_bit = dmx
	165	self.run_start = self.samplenum
	166	else:
	167	self.putr([7, ['Interframe']])
	168	self.state = 'READ BYTE'
	169	self.bit = 0
	170	self.run_start = self.samplenum