[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.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({'skip': 1})
                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 = '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):
	55	self.samplerate = None
	56	self.sample_usec = None
	57	self.run_start = -1
	58	self.run_bit = 0
	59	self.state = 'FIND BREAK'
	60
	61	def start(self):
	62	self.out_ann = self.register(srd.OUTPUT_ANN)
	63
	64	def metadata(self, key, value):
	65	if key == srd.SRD_CONF_SAMPLERATE:
	66	self.samplerate = value
	67	self.sample_usec = 1 / value * 1000000
	68	self.skip_per_bit = int(4 / self.sample_usec)
	69
	70	def putr(self, data):
	71	self.put(self.run_start, self.samplenum, self.out_ann, data)
	72
	73	def decode(self):
	74	if not self.samplerate:
	75	raise SamplerateError('Cannot decode without samplerate.')
	76	while True:
	77	# Seek for an interval with no state change with a length between
	78	# 88 and 1000000 us (BREAK).
	79	if self.state == 'FIND BREAK':
	80	(dmx,) = self.wait({0: 'h' if self.run_bit == 0 else 'l'})
	81	runlen = (self.samplenum - self.run_start) * self.sample_usec
	82	if runlen > 88 and runlen < 1000000:
	83	self.putr([1, ['Break']])
	84	self.bit_break = self.run_bit
	85	self.state = 'MARK MAB'
	86	self.channel = 0
	87	elif runlen >= 1000000:
	88	# Error condition.
	89	self.putr([10, ['Invalid break length']])
	90	self.run_bit = dmx
	91	self.run_start = self.samplenum
	92	# Directly following the BREAK is the MARK AFTER BREAK.
	93	elif self.state == 'MARK MAB':
	94	(dmx,) = self.wait({0: 'h' if self.run_bit == 0 else 'l'})
	95	self.putr([2, ['MAB']])
	96	self.state = 'READ BYTE'
	97	self.channel = 0
	98	self.bit = 0
	99	self.aggreg = dmx
	100	self.run_start = self.samplenum
	101	# Mark and read a single transmitted byte
	102	# (start bit, 8 data bits, 2 stop bits).
	103	elif self.state == 'READ BYTE':
	104	(dmx,) = self.wait({'skip': 1})
	105	self.next_sample = self.run_start + (self.bit + 1) * self.skip_per_bit
	106	self.aggreg += dmx
	107	if self.samplenum != self.next_sample:
	108	continue
	109	bit_value = 0 if round(self.aggreg/self.skip_per_bit) == self.bit_break else 1
	110
	111	if self.bit == 0:
	112	self.byte = 0
	113	self.putr([3, ['Start bit']])
	114	if bit_value != 0:
	115	# (Possibly) invalid start bit, mark but don't fail.
	116	self.put(self.samplenum, self.samplenum,
	117	self.out_ann, [10, ['Invalid start bit']])
	118	elif self.bit >= 9:
	119	self.put(self.samplenum - self.skip_per_bit,
	120	self.samplenum, self.out_ann, [4, ['Stop bit']])
	121	if bit_value != 1:
	122	# Invalid stop bit, mark.
	123	self.put(self.samplenum, self.samplenum,
	124	self.out_ann, [10, ['Invalid stop bit']])
	125	if self.bit == 10:
	126	# On invalid 2nd stop bit, search for new break.
	127	self.run_bit = dmx
	128	self.state = 'FIND BREAK'
	129	else:
	130	# Label and process one bit.
	131	self.put(self.samplenum - self.skip_per_bit,
	132	self.samplenum, self.out_ann, [0, [str(bit_value)]])
	133	self.byte \|= bit_value << (self.bit - 1)
	134
	135	# Label a complete byte.
	136	if self.bit == 10:
	137	if self.channel == 0:
	138	d = [5, ['Start code']]
	139	else:
	140	d = [6, ['Channel ' + str(self.channel)]]
	141	self.put(self.run_start, self.next_sample, self.out_ann, d)
	142	self.put(self.run_start + self.skip_per_bit,
	143	self.next_sample - 2 * self.skip_per_bit,
	144	self.out_ann, [9, [str(self.byte) + ' / ' + \
	145	str(hex(self.byte))]])
	146	# Continue by scanning the IFT.
	147	self.channel += 1
	148	self.run_start = self.samplenum
	149	self.run_bit = dmx
	150	self.state = 'MARK IFT'
	151
	152	self.aggreg = dmx
	153	self.bit += 1
	154	# Mark the INTERFRAME-TIME between bytes / INTERPACKET-TIME between packets.
	155	elif self.state == 'MARK IFT':
	156	(dmx,) = self.wait({0: 'h' if self.run_bit == 0 else 'l'})
	157	if self.channel > 512:
	158	self.putr([8, ['Interpacket']])
	159	self.state = 'FIND BREAK'
	160	self.run_bit = dmx
	161	self.run_start = self.samplenum
	162	else:
	163	self.putr([7, ['Interframe']])
	164	self.state = 'READ BYTE'
	165	self.bit = 0
	166	self.run_start = self.samplenum