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