Code Duplication - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer


title = "Shimadzu ICPE-9000 Multitype"


def Import(context, request):
    """ Read Shimadzu GICPE-9000 Multitype analysis results
    """
    form = request.form
    # TODO form['file'] sometimes returns a list
    infile = form['instrument_results_file'][0] if \
        isinstance(form['instrument_results_file'], list) else \
        form['instrument_results_file']
    artoapply = form['artoapply']
    override = form['results_override']

    instrument = form.get('instrument', None)
    errors = []
    logs = []

    # Load the most suitable parser according to file extension/options/etc...
    parser = None
    if not hasattr(infile, 'filename'):
        errors.append(_("No file selected"))
    parser = ICPEMultitypeCSVParser(infile)

    if parser:
        # Load the importer
        status = ['sample_received', 'attachment_due', 'to_be_verified']
        if artoapply == 'received':
            status = ['sample_received']
        elif artoapply == 'received_tobeverified':
            status = ['sample_received', 'attachment_due', 'to_be_verified']

        over = [False, False]
        if override == 'nooverride':
            over = [False, False]
        elif override == 'override':
            over = [True, False]
        elif override == 'overrideempty':
            over = [True, True]

        importer = ICPEMultitypeImporter(parser=parser,
                                         context=context,
                                         allowed_ar_states=status,
                                         allowed_analysis_states=None,
                                         override=over,
                                         instrument_uid=instrument)
        tbex = ''
        try:
            importer.process()
        except:
            tbex = traceback.format_exc()
        errors = importer.errors
        logs = importer.logs
        warns = importer.warns
        if tbex:
            errors.append(tbex)

    results = {'errors': errors, 'log': logs, 'warns': warns}

    return json.dumps(results)


class ICPEMultitypeCSVParser(InstrumentCSVResultsFileParser):


title = "Agilent - Masshunter"


def Import(context, request):
    """ Read Agilent Masshunter analysis results
    """
    form = request.form
    # TODO form['file'] sometimes returns a list
    infile = form['instrument_results_file'][0] if isinstance(
            form['instrument_results_file'], list) else \
        form['instrument_results_file']
    artoapply = form['artoapply']
    override = form['results_override']

    instrument = form.get('instrument', None)
    errors = []
    logs = []

    # Load the most suitable parser according to file extension/options/etc...
    parser = None
    if not hasattr(infile, 'filename'):
        errors.append(_("No file selected"))
    parser = AgilentMasshunterParser(infile)

    if parser:
        # Load the importer
        status = ['sample_received', 'attachment_due', 'to_be_verified']
        if artoapply == 'received':
            status = ['sample_received']
        elif artoapply == 'received_tobeverified':
            status = ['sample_received', 'attachment_due', 'to_be_verified']

        over = [False, False]
        if override == 'nooverride':
            over = [False, False]
        elif override == 'override':
            over = [True, False]
        elif override == 'overrideempty':
            over = [True, True]

        importer = AgilentMasshunterImporter(parser=parser,
                                             context=context,
                                             allowed_ar_states=status,
                                             allowed_analysis_states=None,
                                             override=over,
                                             instrument_uid=instrument)
        tbex = ''
        try:
            importer.process()
        except:
            tbex = traceback.format_exc()
        errors = importer.errors
        logs = importer.logs
        warns = importer.warns
        if tbex:
            errors.append(tbex)

    results = {'errors': errors, 'log': logs, 'warns': warns}

    return json.dumps(results)


class AgilentMasshunterParser(InstrumentCSVResultsFileParser):


title = "Shimadzu - GCMS-QP2010 SE"


def Import(context, request):
    """ Read Shimadzu GCMS-TQ8030 GC/MS/MS analysis results
    """
    form = request.form
    # TODO form['file'] sometimes returns a list
    infile = form['instrument_results_file'][0] if \
        isinstance(form['instrument_results_file'], list) \
        else form['instrument_results_file']
    override = form['results_override']
    artoapply = form['artoapply']
    instrument = form.get('instrument', None)
    errors = []
    logs = []

    # Load the most suitable parser according to file extension/options/etc...
    parser = None
    if not hasattr(infile, 'filename'):
        errors.append(_("No file selected"))
    parser = GCMSQP2010SECSVParser(infile)

    if parser:
        # Load the importer
        status = ['sample_received', 'attachment_due', 'to_be_verified']
        if artoapply == 'received':
            status = ['sample_received']
        elif artoapply == 'received_tobeverified':
            status = ['sample_received', 'attachment_due', 'to_be_verified']

        over = [False, False]
        if override == 'nooverride':
            over = [False, False]
        elif override == 'override':
            over = [True, False]
        elif override == 'overrideempty':
            over = [True, True]

        importer = GCMSQP2010SEImporter(parser=parser,
                                        context=context,
                                        allowed_ar_states=status,
                                        allowed_analysis_states=None,
                                        override=over,
                                        instrument_uid=instrument)
        tbex = ''
        try:
            importer.process()
        except:
            tbex = traceback.format_exc()
        errors = importer.errors
        logs = importer.logs
        warns = importer.warns
        if tbex:
            errors.append(tbex)

    results = {'errors': errors, 'log': logs, 'warns': warns}

    return json.dumps(results)


class GCMSQP2010SECSVParser(InstrumentCSVResultsFileParser):


title = "Shimadzu GCMS-TQ8030 GC/MS/MS"


def Import(context, request):
    """ Read Shimadzu GCMS-TQ8030 GC/MS/MS analysis results
    """
    form = request.form
    # TODO form['file'] sometimes returns a list
    infile = form['instrument_results_file'][0] if \
        isinstance(form['instrument_results_file'], list) else \
        form['instrument_results_file']
    artoapply = form['artoapply']
    override = form['results_override']
    instrument = form.get('instrument', None)
    errors = []
    logs = []

    # Load the most suitable parser according to file extension/options/etc...
    parser = None
    if not hasattr(infile, 'filename'):
        errors.append(_("No file selected"))
    parser = GCMSTQ8030GCMSMSCSVParser(infile)

    if parser:
        # Load the importer
        status = ['sample_received', 'attachment_due', 'to_be_verified']
        if artoapply == 'received':
            status = ['sample_received']
        elif artoapply == 'received_tobeverified':
            status = ['sample_received', 'attachment_due', 'to_be_verified']

        over = [False, False]
        if override == 'nooverride':
            over = [False, False]
        elif override == 'override':
            over = [True, False]
        elif override == 'overrideempty':
            over = [True, True]

        importer = GCMSTQ8030GCMSMSImporter(parser=parser,
                                            context=context,
                                            allowed_ar_states=status,
                                            allowed_analysis_states=None,
                                            override=over,
                                            instrument_uid=instrument)
        tbex = ''
        try:
            importer.process()
        except:
            tbex = traceback.format_exc()
        errors = importer.errors
        logs = importer.logs
        warns = importer.warns
        if tbex:
            errors.append(tbex)

    results = {'errors': errors, 'log': logs, 'warns': warns}

    return json.dumps(results)


class GCMSTQ8030GCMSMSCSVParser(InstrumentCSVResultsFileParser):


title = "Shimadzu LC MS/MS Nexera X2 LCMS-8050"


def Import(context, request):
    """ Read Shimadzu LC MS/MS Nexera X2 LCMS-8050 analysis results
    """
    form = request.form
    infile = form['instrument_results_file'][0] if \
        isinstance(form['instrument_results_file'], list) else \
        form['instrument_results_file']
    artoapply = form['artoapply']
    override = form['results_override']
    instrument = form.get('instrument', None)
    errors = []
    logs = []

    # Load the most suitable parser according to file extension/options/etc...
    parser = None
    if not hasattr(infile, 'filename'):
        errors.append(_("No file selected"))
    parser = TSVParser(infile)

    if parser:
        # Load the importer
        status = ['sample_received', 'attachment_due', 'to_be_verified']
        if artoapply == 'received':
            status = ['sample_received']
        elif artoapply == 'received_tobeverified':
            status = ['sample_received', 'attachment_due', 'to_be_verified']

        over = [False, False]
        if override == 'nooverride':
            over = [False, False]
        elif override == 'override':
            over = [True, False]
        elif override == 'overrideempty':
            over = [True, True]

        importer = LCMS8050_Importer(parser=parser,
                                     context=context,
                                     allowed_ar_states=status,
                                     allowed_analysis_states=None,
                                     override=over,
                                     instrument_uid=instrument)
        tbex = ''
        try:
            importer.process()
        except:
            tbex = traceback.format_exc()
        errors = importer.errors
        logs = importer.logs
        warns = importer.warns
        if tbex:
            errors.append(tbex)

    results = {'errors': errors, 'log': logs, 'warns': warns}

    return json.dumps(results)


class TSVParser(InstrumentCSVResultsFileParser):


title = "Shimadzu HPLC-PDA Nexera-I LC2040C"


def Import(context, request):
    """ Read Shimadzu HPLC-PDA Nexera-I LC2040C analysis results
    """
    form = request.form
    infile = form['instrument_results_file'][0] if \
        isinstance(form['instrument_results_file'], list) else \
        form['instrument_results_file']
    artoapply = form['artoapply']
    override = form['results_override']
    instrument = form.get('instrument', None)
    errors = []
    logs = []

    # Load the most suitable parser according to file extension/options/etc...
    parser = None
    if not hasattr(infile, 'filename'):
        errors.append(_("No file selected"))
    parser = TSVParser(infile)

    if parser:
        # Load the importer
        status = ['sample_received', 'attachment_due', 'to_be_verified']
        if artoapply == 'received':
            status = ['sample_received']
        elif artoapply == 'received_tobeverified':
            status = ['sample_received', 'attachment_due', 'to_be_verified']

        over = [False, False]
        if override == 'nooverride':
            over = [False, False]
        elif override == 'override':
            over = [True, False]
        elif override == 'overrideempty':
            over = [True, True]

        importer = LC2040C_Importer(parser=parser,
                                    context=context,
                                    allowed_ar_states=status,
                                    allowed_analysis_states=None,
                                    override=over,
                                    instrument_uid=instrument)
        tbex = ''
        try:
            importer.process()
        except:
            tbex = traceback.format_exc()
        errors = importer.errors
        logs = importer.logs
        warns = importer.warns
        if tbex:
            errors.append(tbex)

    results = {'errors': errors, 'log': logs, 'warns': warns}

    return json.dumps(results)


class TSVParser(InstrumentCSVResultsFileParser):

		@@ 21-77 (lines=57) @@
18		title = "Shimadzu ICPE-9000 Multitype"
19
20
21		def Import(context, request):
22		""" Read Shimadzu GICPE-9000 Multitype analysis results
23		"""
24		form = request.form
25		# TODO form['file'] sometimes returns a list
26		infile = form['instrument_results_file'][0] if \
27		isinstance(form['instrument_results_file'], list) else \
28		form['instrument_results_file']
29		artoapply = form['artoapply']
30		override = form['results_override']
31
32		instrument = form.get('instrument', None)
33		errors = []
34		logs = []
35
36		# Load the most suitable parser according to file extension/options/etc...
37		parser = None
38		if not hasattr(infile, 'filename'):
39		errors.append(_("No file selected"))
40		parser = ICPEMultitypeCSVParser(infile)
41
42		if parser:
43		# Load the importer
44		status = ['sample_received', 'attachment_due', 'to_be_verified']
45		if artoapply == 'received':
46		status = ['sample_received']
47		elif artoapply == 'received_tobeverified':
48		status = ['sample_received', 'attachment_due', 'to_be_verified']
49
50		over = [False, False]
51		if override == 'nooverride':
52		over = [False, False]
53		elif override == 'override':
54		over = [True, False]
55		elif override == 'overrideempty':
56		over = [True, True]
57
58		importer = ICPEMultitypeImporter(parser=parser,
59		context=context,
60		allowed_ar_states=status,
61		allowed_analysis_states=None,
62		override=over,
63		instrument_uid=instrument)
64		tbex = ''
65		try:
66		importer.process()
67		except:
68		tbex = traceback.format_exc()
69		errors = importer.errors
70		logs = importer.logs
71		warns = importer.warns
72		if tbex:
73		errors.append(tbex)
74
75		results = {'errors': errors, 'log': logs, 'warns': warns}
76
77		return json.dumps(results)
78
79
80		class ICPEMultitypeCSVParser(InstrumentCSVResultsFileParser):

		@@ 21-77 (lines=57) @@
18		title = "Agilent - Masshunter"
19
20
21		def Import(context, request):
22		""" Read Agilent Masshunter analysis results
23		"""
24		form = request.form
25		# TODO form['file'] sometimes returns a list
26		infile = form['instrument_results_file'][0] if isinstance(
27		form['instrument_results_file'], list) else \
28		form['instrument_results_file']
29		artoapply = form['artoapply']
30		override = form['results_override']
31
32		instrument = form.get('instrument', None)
33		errors = []
34		logs = []
35
36		# Load the most suitable parser according to file extension/options/etc...
37		parser = None
38		if not hasattr(infile, 'filename'):
39		errors.append(_("No file selected"))
40		parser = AgilentMasshunterParser(infile)
41
42		if parser:
43		# Load the importer
44		status = ['sample_received', 'attachment_due', 'to_be_verified']
45		if artoapply == 'received':
46		status = ['sample_received']
47		elif artoapply == 'received_tobeverified':
48		status = ['sample_received', 'attachment_due', 'to_be_verified']
49
50		over = [False, False]
51		if override == 'nooverride':
52		over = [False, False]
53		elif override == 'override':
54		over = [True, False]
55		elif override == 'overrideempty':
56		over = [True, True]
57
58		importer = AgilentMasshunterImporter(parser=parser,
59		context=context,
60		allowed_ar_states=status,
61		allowed_analysis_states=None,
62		override=over,
63		instrument_uid=instrument)
64		tbex = ''
65		try:
66		importer.process()
67		except:
68		tbex = traceback.format_exc()
69		errors = importer.errors
70		logs = importer.logs
71		warns = importer.warns
72		if tbex:
73		errors.append(tbex)
74
75		results = {'errors': errors, 'log': logs, 'warns': warns}
76
77		return json.dumps(results)
78
79
80		class AgilentMasshunterParser(InstrumentCSVResultsFileParser):

		@@ 21-76 (lines=56) @@
18		title = "Shimadzu - GCMS-QP2010 SE"
19
20
21		def Import(context, request):
22		""" Read Shimadzu GCMS-TQ8030 GC/MS/MS analysis results
23		"""
24		form = request.form
25		# TODO form['file'] sometimes returns a list
26		infile = form['instrument_results_file'][0] if \
27		isinstance(form['instrument_results_file'], list) \
28		else form['instrument_results_file']
29		override = form['results_override']
30		artoapply = form['artoapply']
31		instrument = form.get('instrument', None)
32		errors = []
33		logs = []
34
35		# Load the most suitable parser according to file extension/options/etc...
36		parser = None
37		if not hasattr(infile, 'filename'):
38		errors.append(_("No file selected"))
39		parser = GCMSQP2010SECSVParser(infile)
40
41		if parser:
42		# Load the importer
43		status = ['sample_received', 'attachment_due', 'to_be_verified']
44		if artoapply == 'received':
45		status = ['sample_received']
46		elif artoapply == 'received_tobeverified':
47		status = ['sample_received', 'attachment_due', 'to_be_verified']
48
49		over = [False, False]
50		if override == 'nooverride':
51		over = [False, False]
52		elif override == 'override':
53		over = [True, False]
54		elif override == 'overrideempty':
55		over = [True, True]
56
57		importer = GCMSQP2010SEImporter(parser=parser,
58		context=context,
59		allowed_ar_states=status,
60		allowed_analysis_states=None,
61		override=over,
62		instrument_uid=instrument)
63		tbex = ''
64		try:
65		importer.process()
66		except:
67		tbex = traceback.format_exc()
68		errors = importer.errors
69		logs = importer.logs
70		warns = importer.warns
71		if tbex:
72		errors.append(tbex)
73
74		results = {'errors': errors, 'log': logs, 'warns': warns}
75
76		return json.dumps(results)
77
78
79		class GCMSQP2010SECSVParser(InstrumentCSVResultsFileParser):

		@@ 20-75 (lines=56) @@
17		title = "Shimadzu GCMS-TQ8030 GC/MS/MS"
18
19
20		def Import(context, request):
21		""" Read Shimadzu GCMS-TQ8030 GC/MS/MS analysis results
22		"""
23		form = request.form
24		# TODO form['file'] sometimes returns a list
25		infile = form['instrument_results_file'][0] if \
26		isinstance(form['instrument_results_file'], list) else \
27		form['instrument_results_file']
28		artoapply = form['artoapply']
29		override = form['results_override']
30		instrument = form.get('instrument', None)
31		errors = []
32		logs = []
33
34		# Load the most suitable parser according to file extension/options/etc...
35		parser = None
36		if not hasattr(infile, 'filename'):
37		errors.append(_("No file selected"))
38		parser = GCMSTQ8030GCMSMSCSVParser(infile)
39
40		if parser:
41		# Load the importer
42		status = ['sample_received', 'attachment_due', 'to_be_verified']
43		if artoapply == 'received':
44		status = ['sample_received']
45		elif artoapply == 'received_tobeverified':
46		status = ['sample_received', 'attachment_due', 'to_be_verified']
47
48		over = [False, False]
49		if override == 'nooverride':
50		over = [False, False]
51		elif override == 'override':
52		over = [True, False]
53		elif override == 'overrideempty':
54		over = [True, True]
55
56		importer = GCMSTQ8030GCMSMSImporter(parser=parser,
57		context=context,
58		allowed_ar_states=status,
59		allowed_analysis_states=None,
60		override=over,
61		instrument_uid=instrument)
62		tbex = ''
63		try:
64		importer.process()
65		except:
66		tbex = traceback.format_exc()
67		errors = importer.errors
68		logs = importer.logs
69		warns = importer.warns
70		if tbex:
71		errors.append(tbex)
72
73		results = {'errors': errors, 'log': logs, 'warns': warns}
74
75		return json.dumps(results)
76
77
78		class GCMSTQ8030GCMSMSCSVParser(InstrumentCSVResultsFileParser):

		@@ 20-74 (lines=55) @@
17		title = "Shimadzu LC MS/MS Nexera X2 LCMS-8050"
18
19
20		def Import(context, request):
21		""" Read Shimadzu LC MS/MS Nexera X2 LCMS-8050 analysis results
22		"""
23		form = request.form
24		infile = form['instrument_results_file'][0] if \
25		isinstance(form['instrument_results_file'], list) else \
26		form['instrument_results_file']
27		artoapply = form['artoapply']
28		override = form['results_override']
29		instrument = form.get('instrument', None)
30		errors = []
31		logs = []
32
33		# Load the most suitable parser according to file extension/options/etc...
34		parser = None
35		if not hasattr(infile, 'filename'):
36		errors.append(_("No file selected"))
37		parser = TSVParser(infile)
38
39		if parser:
40		# Load the importer
41		status = ['sample_received', 'attachment_due', 'to_be_verified']
42		if artoapply == 'received':
43		status = ['sample_received']
44		elif artoapply == 'received_tobeverified':
45		status = ['sample_received', 'attachment_due', 'to_be_verified']
46
47		over = [False, False]
48		if override == 'nooverride':
49		over = [False, False]
50		elif override == 'override':
51		over = [True, False]
52		elif override == 'overrideempty':
53		over = [True, True]
54
55		importer = LCMS8050_Importer(parser=parser,
56		context=context,
57		allowed_ar_states=status,
58		allowed_analysis_states=None,
59		override=over,
60		instrument_uid=instrument)
61		tbex = ''
62		try:
63		importer.process()
64		except:
65		tbex = traceback.format_exc()
66		errors = importer.errors
67		logs = importer.logs
68		warns = importer.warns
69		if tbex:
70		errors.append(tbex)
71
72		results = {'errors': errors, 'log': logs, 'warns': warns}
73
74		return json.dumps(results)
75
76
77		class TSVParser(InstrumentCSVResultsFileParser):

Code Duplication Length = 55-57 lines in 6 locations

bika/lims/exportimport/instruments/shimadzu/icpe/multitype.py 1 location

bika/lims/exportimport/instruments/agilent/masshunter/masshunter.py 1 location

bika/lims/exportimport/instruments/shimadzu/gcms/qp2010se.py 1 location

bika/lims/exportimport/instruments/shimadzu/gcms/tq8030.py 1 location

bika/lims/exportimport/instruments/shimadzu/nexera/LCMS8050.py 1 location

bika/lims/exportimport/instruments/shimadzu/nexera/LC2040C.py 1 location

		@@ 20-74 (lines=55) @@
17		title = "Shimadzu HPLC-PDA Nexera-I LC2040C"
18
19
20		def Import(context, request):
21		""" Read Shimadzu HPLC-PDA Nexera-I LC2040C analysis results
22		"""
23		form = request.form
24		infile = form['instrument_results_file'][0] if \
25		isinstance(form['instrument_results_file'], list) else \
26		form['instrument_results_file']
27		artoapply = form['artoapply']
28		override = form['results_override']
29		instrument = form.get('instrument', None)
30		errors = []
31		logs = []
32
33		# Load the most suitable parser according to file extension/options/etc...
34		parser = None
35		if not hasattr(infile, 'filename'):
36		errors.append(_("No file selected"))
37		parser = TSVParser(infile)
38
39		if parser:
40		# Load the importer
41		status = ['sample_received', 'attachment_due', 'to_be_verified']
42		if artoapply == 'received':
43		status = ['sample_received']
44		elif artoapply == 'received_tobeverified':
45		status = ['sample_received', 'attachment_due', 'to_be_verified']
46
47		over = [False, False]
48		if override == 'nooverride':
49		over = [False, False]
50		elif override == 'override':
51		over = [True, False]
52		elif override == 'overrideempty':
53		over = [True, True]
54
55		importer = LC2040C_Importer(parser=parser,
56		context=context,
57		allowed_ar_states=status,
58		allowed_analysis_states=None,
59		override=over,
60		instrument_uid=instrument)
61		tbex = ''
62		try:
63		importer.process()
64		except:
65		tbex = traceback.format_exc()
66		errors = importer.errors
67		logs = importer.logs
68		warns = importer.warns
69		if tbex:
70		errors.append(tbex)
71
72		results = {'errors': errors, 'log': logs, 'warns': warns}
73
74		return json.dumps(results)
75
76
77		class TSVParser(InstrumentCSVResultsFileParser):

senaite / senaite.core

Code Duplication Length = 55-57 lines in 6 locations

bika/lims/exportimport/instruments/shimadzu/icpe/multitype.py 1 location

bika/lims/exportimport/instruments/agilent/masshunter/masshunter.py 1 location

bika/lims/exportimport/instruments/shimadzu/gcms/qp2010se.py 1 location

bika/lims/exportimport/instruments/shimadzu/gcms/tq8030.py 1 location

bika/lims/exportimport/instruments/shimadzu/nexera/LCMS8050.py 1 location

bika/lims/exportimport/instruments/shimadzu/nexera/LC2040C.py 1 location