kerapu.command.TestsetShredderCommand.TestShredderCommand.__maak_xpath() - Code Metrics - SetBased/py-kerapu - Measure and Improve Code Quality continuously with Scrutinizer

TestShredderCommand.__maak_xpath() A
last analyzed 2020-10-26 17:19 UTC

↳ Parent: kerapu.command.TestsetShredderCommand

Complexity

Conditions

Size

Total Lines	16
Code Lines	8

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	12

Importance

Changes

Metric	Value
cc	3
eloc	8
nop	1
dl	0
loc	16
ccs	0
cts	7
cp	0
crap	12
rs	10
c	0
b	0
f	0

import csv
import datetime
import os
import random
import shutil
import string
import zipfile
from typing import Iterable, List, Dict

from cleo import Command
from lxml import etree

from kerapu.style.KerapuStyle import KerapuStyle


class TestShredderCommand(Command):
    """
    Converteert XML-bestand met de testset naar een CSV-bestand

    kerapu:test-shredder
        {testset-zip : ZIP-bestand met de testset}
        {testset-csv : Path waar het CSV-bestand met de tests moeten worden opgeslagen}
    """

    # ------------------------------------------------------------------------------------------------------------------
    def __extract_zip_file(self, zip_filename: str, tmp_dir: str):
        """
        Extracts het ZIP-bestand met de testset in een folder.

        :param str zip_filename: Het path naar het ZIP-bestand met de testset.
        :param str tmp_dir: Path naar de folder.
        """
        self.output.writeln('Uitpakken van <fso>{}</fso> in <fso>{}</fso>'.format(zip_filename, tmp_dir))

        with zipfile.ZipFile(zip_filename, 'r') as zip_ref:
            zip_ref.extractall(tmp_dir)

    # ------------------------------------------------------------------------------------------------------------------
    @staticmethod
    def ordinal(path: str) -> int:
        """
        Geeft het volgnummer van een test.

        :param str path: Het path naar het XML-bestand met de test case.
        """
        parts = os.path.basename(path).split('_')

        return int(parts[6])

    # ------------------------------------------------------------------------------------------------------------------
    def __lees_test_cases_lijst(self, folder: str) -> List:
        """
        Geeft een lijst met alle bestanden in een folder.

        :param str folder: Het path naar de folder.
        """
        entries = os.listdir(folder)
        filenames = list()
        for entry in entries:
            path = os.path.join(folder, entry)
            if os.path.isfile(path):
                filenames.append(path)

        self.output.writeln('Aantal gevonden test cases: {}'.format(len(filenames)))

        return sorted(filenames, key=TestShredderCommand.ordinal)

    # ------------------------------------------------------------------------------------------------------------------
    @staticmethod
    def __maak_xpath(parts: Iterable) -> str:
        """
        Maakt een string met een xpath.

        :param tuple parts: The onderdelen van het xpath.

        :rtype: str
        """
        xpath = ''
        for part in parts:
            if xpath:
                xpath += '/'
            xpath += 'xmlns:' + part

        return xpath

    # ------------------------------------------------------------------------------------------------------------------
    @staticmethod
    def __convert_date(date: str) -> str:
        """
        Converteert een datum in YYYYMMDD formaat naar YYYY-MM-DD format.

        :param str date: De datum in YYYYMMDD format.

        :rtype: str
        """
        return date[:4] + '-' + date[4:6] + '-' + date[6:8]

    # ------------------------------------------------------------------------------------------------------------------
    @staticmethod
    def __leeftijd_geboorte_datum(date: str, leeftijd: int) -> str:
        """
        Geeft de geboortedatum gegeven een datum en een leeftijd (en de persoon is niet jarig).

        :param str date: De gegeven datum in YYYY-MM-DD format.
        :param int leeftijd: De leeftijd in jaren.

        :rtype: int
        """
        date = datetime.date(int(date[:4]) - leeftijd, int(date[5:7]), int(date[8:10]))
        date -= datetime.timedelta(days=1)

        return date.isoformat()

    # ------------------------------------------------------------------------------------------------------------------
    def __shred_xml_bestand(self, filename: str) -> Dict:
        """
        Leest de relevante data in een XML-bestand met een test case.

        :param str filename: De filenaam van het XML bestand.

        :rtype: dict
        """
        doc = etree.parse(filename)

        xpath = '/soapenv:Envelope/soapenv:Body/xmlns:FICR_IN900101NL04'
        namespaces = {'soapenv': 'http://schemas.xmlsoap.org/soap/envelope/',
                      'xmlns':   'urn:hl7-org:v3'}

        # Lees declaratiecode.
        parts = ('ControlActProcess', 'subject', 'Declaratiedataset', 'component', 'subtraject', 'id')
        elements = doc.xpath(xpath + '/' + self.__maak_xpath(parts), namespaces=namespaces)
        declaratie_code = elements[0].get('extension')

        # Lees specialismecode.
        parts = ('ControlActProcess', 'subject', 'Declaratiedataset', 'component', 'subtraject', 'derivedFrom',
                 'zorgtraject', 'responsibleParty', 'assignedPerson', 'code')
        elements = doc.xpath(xpath + '/' + self.__maak_xpath(parts), namespaces=namespaces)
        specialisme_code = elements[0].get('code')

        # Lees diagnosecode.
        parts = (
            'ControlActProcess', 'subject', 'Declaratiedataset', 'component', 'subtraject', 'pertinentInformation1',
            'typerendeDiagnose', 'value')
        elements = doc.xpath(xpath + '/' + self.__maak_xpath(parts), namespaces=namespaces)
        diagnose_code = elements[0].get('code')

        # Lees zorgtypecode.
        parts = ('ControlActProcess', 'subject', 'Declaratiedataset', 'component', 'subtraject', 'code')
        elements = doc.xpath(xpath + '/' + self.__maak_xpath(parts), namespaces=namespaces)
        zorg_type_code = elements[0].get('code') if elements else None

        # Lees zorgvraagcode.
        parts = ('ControlActProcess', 'subject', 'Declaratiedataset', 'component', 'subtraject', 'derivedFrom',
                 'zorgtraject', 'reason', 'zorgvraag', 'value')
        elements = doc.xpath(xpath + '/' + self.__maak_xpath(parts), namespaces=namespaces)
        zorg_vraag_code = elements[0].get('code') if elements else None

        # Lees begindatum.
        parts = ('ControlActProcess', 'subject', 'Declaratiedataset', 'component', 'subtraject', 'effectiveTime', 'low')
        elements = doc.xpath(xpath + '/' + self.__maak_xpath(parts), namespaces=namespaces)
        begin_datum = self.__convert_date(elements[0].get('value')) if elements else None

        # Lees de geboortedatum van de patient.
        parts = ('ControlActProcess', 'subject', 'Declaratiedataset', 'subject', 'patient', 'subjectOf', 'leeftijd',
                 'value')
        elements = doc.xpath(xpath + '/' + self.__maak_xpath(parts), namespaces=namespaces)
        leeftijd = int(elements[0].get('value')) if elements else None
        geboorte_datum = self.__leeftijd_geboorte_datum(begin_datum, leeftijd)

        # Lees het geslacht van de patient.
        parts = ('ControlActProcess', 'subject', 'Declaratiedataset', 'subject', 'patient', 'patientPerson',
                 'administrativeGenderCode')
        elements = doc.xpath(xpath + '/' + self.__maak_xpath(parts), namespaces=namespaces)
        geslacht_code = elements[0].get('code') if elements else None

        # Lees de AGB-code van de zorginstelling.
        parts = ('ControlActProcess', 'subject', 'Declaratiedataset', 'author', 'assignedOrganization', 'id')
        elements = doc.xpath(xpath + '/' + self.__maak_xpath(parts), namespaces=namespaces)
        zorg_instelling_code = elements[0].get('extension') if elements else None

        # Lees alle zorgactiviteiten.
        zorg_activiteiten = list()
        parts = ('ControlActProcess', 'subject', 'Declaratiedataset', 'component', 'subtraject', 'debit',
                 'zorgactiviteit')
        elements = doc.xpath(xpath + '/' + self.__maak_xpath(parts), namespaces=namespaces)
        for element in elements:
            path = 'xmlns:code'
            sub_elements = element.xpath(path, namespaces=namespaces)
            zorg_activiteit_code = sub_elements[0].get('code') if sub_elements else None

            path = 'xmlns:repeatNumber'
            sub_elements = element.xpath(path, namespaces=namespaces)
            aantal = int(sub_elements[0].get('value')) if sub_elements else None

            zorg_activiteiten.append((zorg_activiteit_code, aantal))

        return {'subtraject_nummer':    os.path.basename(filename),
                'declaratie_code':      declaratie_code,
                'specialisme_code':     specialisme_code,
                'diagnose_code':        diagnose_code,
                'zorg_type_code':       zorg_type_code,
                'zorg_vraag_code':      zorg_vraag_code,
                'begin_datum':          begin_datum,
                'geboorte_datum':       geboorte_datum,
                'geslacht_code':        geslacht_code,
                'zorg_instelling_code': zorg_instelling_code,
                'zorg_activiteiten':    zorg_activiteiten}

    # ----------------------------------------------------------------------------------------------------------------------
    @staticmethod
    def __write_subtraject(writer, subtraject: Dict) -> None:
        """
        Schrijft het subtraject met alle zorgactiviteiten naar een CSV-bestand.

        :param writer: De handle naar de CSV writer.
        :param dict subtraject: De details van het subtract.
        """
        writer.writerow((subtraject['subtraject_nummer'],
                         subtraject['specialisme_code'],
                         subtraject['diagnose_code'],
                         subtraject['zorg_type_code'],
                         subtraject['zorg_vraag_code'],
                         subtraject['begin_datum'],
                         subtraject['geboorte_datum'],
                         subtraject['geslacht_code'],
                         subtraject['zorg_instelling_code'],
                         subtraject['declaratie_code']))

        for zorgactiviteit in subtraject['zorg_activiteiten']:
            writer.writerow((zorgactiviteit[0], zorgactiviteit[1]))

    # ----------------------------------------------------------------------------------------------------------------------
    def __extract_files(self, writer, filenames: List) -> None:
        """
        Extract de data van een lijst met XML-bestanden met test cases en schrijft deze data naar een CSV-bestand.

        :param writer:  De handle naar de CSV writer.
        :param list filenames: De lijst met bestandsnamen van XML-bestanden met test cases.
        """
        for filename in filenames:
            subtraject = self.__shred_xml_bestand(filename)
            self.__write_subtraject(writer, subtraject)

    # ------------------------------------------------------------------------------------------------------------------
    def handle(self) -> int:
        """
        Executes the command.
        """
        self.output = KerapuStyle(self.input, self.output)

        zip_filename = self.argument('testset-zip')
        csv_filename = self.argument('testset-csv')
        tmp_dir = '.kerapu-' + ''.join(random.choices(string.ascii_lowercase, k=12))

        os.mkdir(tmp_dir)

        self.__extract_zip_file(zip_filename, tmp_dir)
        files = self.__lees_test_cases_lijst(tmp_dir)

        with open(csv_filename, 'w', encoding='utf-8') as handle:
            csv_writer = csv.writer(handle, dialect=csv.unix_dialect)
            self.__extract_files(csv_writer, files)

        shutil.rmtree(tmp_dir)

        return 0

# ----------------------------------------------------------------------------------------------------------------------


1			import csv
2			import datetime
3			import os
4			import random
5			import shutil
6			import string
7			import zipfile
8			from typing import Iterable, List, Dict
9
10			from cleo import Command
11			from lxml import etree
12
13			from kerapu.style.KerapuStyle import KerapuStyle
14
15
16			class TestShredderCommand(Command):
17			"""
18			Converteert XML-bestand met de testset naar een CSV-bestand
19
20			kerapu:test-shredder
21			{testset-zip : ZIP-bestand met de testset}
22			{testset-csv : Path waar het CSV-bestand met de tests moeten worden opgeslagen}
23			"""
24
25			# ------------------------------------------------------------------------------------------------------------------
26			def __extract_zip_file(self, zip_filename: str, tmp_dir: str):
27			"""
28			Extracts het ZIP-bestand met de testset in een folder.
29
30			:param str zip_filename: Het path naar het ZIP-bestand met de testset.
31			:param str tmp_dir: Path naar de folder.
32			"""
33			self.output.writeln('Uitpakken van <fso>{}</fso> in <fso>{}</fso>'.format(zip_filename, tmp_dir))
34
35			with zipfile.ZipFile(zip_filename, 'r') as zip_ref:
36			zip_ref.extractall(tmp_dir)
37
38			# ------------------------------------------------------------------------------------------------------------------
39			@staticmethod
40			def ordinal(path: str) -> int:
41			"""
42			Geeft het volgnummer van een test.
43
44			:param str path: Het path naar het XML-bestand met de test case.
45			"""
46			parts = os.path.basename(path).split('_')
47
48			return int(parts[6])
49
50			# ------------------------------------------------------------------------------------------------------------------
51			def __lees_test_cases_lijst(self, folder: str) -> List:
52			"""
53			Geeft een lijst met alle bestanden in een folder.
54
55			:param str folder: Het path naar de folder.
56			"""
57			entries = os.listdir(folder)
58			filenames = list()
59			for entry in entries:
60			path = os.path.join(folder, entry)
61			if os.path.isfile(path):
62			filenames.append(path)
63
64			self.output.writeln('Aantal gevonden test cases: {}'.format(len(filenames)))
65
66			return sorted(filenames, key=TestShredderCommand.ordinal)
67
68			# ------------------------------------------------------------------------------------------------------------------
69			@staticmethod
70			def __maak_xpath(parts: Iterable) -> str:
71			"""
72			Maakt een string met een xpath.
73
74			:param tuple parts: The onderdelen van het xpath.
75
76			:rtype: str
77			"""
78			xpath = ''
79			for part in parts:
80			if xpath:
81			xpath += '/'
82			xpath += 'xmlns:' + part
83
84			return xpath
85
86			# ------------------------------------------------------------------------------------------------------------------
87			@staticmethod
88			def __convert_date(date: str) -> str:
89			"""
90			Converteert een datum in YYYYMMDD formaat naar YYYY-MM-DD format.
91
92			:param str date: De datum in YYYYMMDD format.
93
94			:rtype: str
95			"""
96			return date[:4] + '-' + date[4:6] + '-' + date[6:8]
97
98			# ------------------------------------------------------------------------------------------------------------------
99			@staticmethod
100			def __leeftijd_geboorte_datum(date: str, leeftijd: int) -> str:
101			"""
102			Geeft de geboortedatum gegeven een datum en een leeftijd (en de persoon is niet jarig).
103
104			:param str date: De gegeven datum in YYYY-MM-DD format.
105			:param int leeftijd: De leeftijd in jaren.
106
107			:rtype: int
108			"""
109			date = datetime.date(int(date[:4]) - leeftijd, int(date[5:7]), int(date[8:10]))
110			date -= datetime.timedelta(days=1)
111
112			return date.isoformat()
113
114			# ------------------------------------------------------------------------------------------------------------------
115			def __shred_xml_bestand(self, filename: str) -> Dict:
116			"""
117			Leest de relevante data in een XML-bestand met een test case.
118
119			:param str filename: De filenaam van het XML bestand.
120
121			:rtype: dict
122			"""
123			doc = etree.parse(filename)
124
125			xpath = '/soapenv:Envelope/soapenv:Body/xmlns:FICR_IN900101NL04'
126			namespaces = {'soapenv': 'http://schemas.xmlsoap.org/soap/envelope/',
127			'xmlns': 'urn:hl7-org:v3'}
128
129			# Lees declaratiecode.
130			parts = ('ControlActProcess', 'subject', 'Declaratiedataset', 'component', 'subtraject', 'id')
131			elements = doc.xpath(xpath + '/' + self.__maak_xpath(parts), namespaces=namespaces)
132			declaratie_code = elements[0].get('extension')
133
134			# Lees specialismecode.
135			parts = ('ControlActProcess', 'subject', 'Declaratiedataset', 'component', 'subtraject', 'derivedFrom',
136			'zorgtraject', 'responsibleParty', 'assignedPerson', 'code')
137			elements = doc.xpath(xpath + '/' + self.__maak_xpath(parts), namespaces=namespaces)
138			specialisme_code = elements[0].get('code')
139
140			# Lees diagnosecode.
141			parts = (
142			'ControlActProcess', 'subject', 'Declaratiedataset', 'component', 'subtraject', 'pertinentInformation1',
143			'typerendeDiagnose', 'value')
144			elements = doc.xpath(xpath + '/' + self.__maak_xpath(parts), namespaces=namespaces)
145			diagnose_code = elements[0].get('code')
146
147			# Lees zorgtypecode.
148			parts = ('ControlActProcess', 'subject', 'Declaratiedataset', 'component', 'subtraject', 'code')
149			elements = doc.xpath(xpath + '/' + self.__maak_xpath(parts), namespaces=namespaces)
150			zorg_type_code = elements[0].get('code') if elements else None
151
152			# Lees zorgvraagcode.
153			parts = ('ControlActProcess', 'subject', 'Declaratiedataset', 'component', 'subtraject', 'derivedFrom',
154			'zorgtraject', 'reason', 'zorgvraag', 'value')
155			elements = doc.xpath(xpath + '/' + self.__maak_xpath(parts), namespaces=namespaces)
156			zorg_vraag_code = elements[0].get('code') if elements else None
157
158			# Lees begindatum.
159			parts = ('ControlActProcess', 'subject', 'Declaratiedataset', 'component', 'subtraject', 'effectiveTime', 'low')
160			elements = doc.xpath(xpath + '/' + self.__maak_xpath(parts), namespaces=namespaces)
161			begin_datum = self.__convert_date(elements[0].get('value')) if elements else None
162
163			# Lees de geboortedatum van de patient.
164			parts = ('ControlActProcess', 'subject', 'Declaratiedataset', 'subject', 'patient', 'subjectOf', 'leeftijd',
165			'value')
166			elements = doc.xpath(xpath + '/' + self.__maak_xpath(parts), namespaces=namespaces)
167			leeftijd = int(elements[0].get('value')) if elements else None
168			geboorte_datum = self.__leeftijd_geboorte_datum(begin_datum, leeftijd)
169
170			# Lees het geslacht van de patient.
171			parts = ('ControlActProcess', 'subject', 'Declaratiedataset', 'subject', 'patient', 'patientPerson',
172			'administrativeGenderCode')
173			elements = doc.xpath(xpath + '/' + self.__maak_xpath(parts), namespaces=namespaces)
174			geslacht_code = elements[0].get('code') if elements else None
175
176			# Lees de AGB-code van de zorginstelling.
177			parts = ('ControlActProcess', 'subject', 'Declaratiedataset', 'author', 'assignedOrganization', 'id')
178			elements = doc.xpath(xpath + '/' + self.__maak_xpath(parts), namespaces=namespaces)
179			zorg_instelling_code = elements[0].get('extension') if elements else None
180
181			# Lees alle zorgactiviteiten.
182			zorg_activiteiten = list()
183			parts = ('ControlActProcess', 'subject', 'Declaratiedataset', 'component', 'subtraject', 'debit',
184			'zorgactiviteit')
185			elements = doc.xpath(xpath + '/' + self.__maak_xpath(parts), namespaces=namespaces)
186			for element in elements:
187			path = 'xmlns:code'
188			sub_elements = element.xpath(path, namespaces=namespaces)
189			zorg_activiteit_code = sub_elements[0].get('code') if sub_elements else None
190
191			path = 'xmlns:repeatNumber'
192			sub_elements = element.xpath(path, namespaces=namespaces)
193			aantal = int(sub_elements[0].get('value')) if sub_elements else None
194
195			zorg_activiteiten.append((zorg_activiteit_code, aantal))
196
197			return {'subtraject_nummer': os.path.basename(filename),
198			'declaratie_code': declaratie_code,
199			'specialisme_code': specialisme_code,
200			'diagnose_code': diagnose_code,
201			'zorg_type_code': zorg_type_code,
202			'zorg_vraag_code': zorg_vraag_code,
203			'begin_datum': begin_datum,
204			'geboorte_datum': geboorte_datum,
205			'geslacht_code': geslacht_code,
206			'zorg_instelling_code': zorg_instelling_code,
207			'zorg_activiteiten': zorg_activiteiten}
208
209			# ----------------------------------------------------------------------------------------------------------------------
210			@staticmethod
211			def __write_subtraject(writer, subtraject: Dict) -> None:
212			"""
213			Schrijft het subtraject met alle zorgactiviteiten naar een CSV-bestand.
214
215			:param writer: De handle naar de CSV writer.
216			:param dict subtraject: De details van het subtract.
217			"""
218			writer.writerow((subtraject['subtraject_nummer'],
219			subtraject['specialisme_code'],
220			subtraject['diagnose_code'],
221			subtraject['zorg_type_code'],
222			subtraject['zorg_vraag_code'],
223			subtraject['begin_datum'],
224			subtraject['geboorte_datum'],
225			subtraject['geslacht_code'],
226			subtraject['zorg_instelling_code'],
227			subtraject['declaratie_code']))
228
229			for zorgactiviteit in subtraject['zorg_activiteiten']:
230			writer.writerow((zorgactiviteit[0], zorgactiviteit[1]))
231
232			# ----------------------------------------------------------------------------------------------------------------------
233			def __extract_files(self, writer, filenames: List) -> None:
234			"""
235			Extract de data van een lijst met XML-bestanden met test cases en schrijft deze data naar een CSV-bestand.
236
237			:param writer: De handle naar de CSV writer.
238			:param list filenames: De lijst met bestandsnamen van XML-bestanden met test cases.
239			"""
240			for filename in filenames:
241			subtraject = self.__shred_xml_bestand(filename)
242			self.__write_subtraject(writer, subtraject)
243
244			# ------------------------------------------------------------------------------------------------------------------
245			def handle(self) -> int:
246			"""
247			Executes the command.
248			"""
249			self.output = KerapuStyle(self.input, self.output)
250
251			zip_filename = self.argument('testset-zip')
252			csv_filename = self.argument('testset-csv')
253			tmp_dir = '.kerapu-' + ''.join(random.choices(string.ascii_lowercase, k=12))
254
255			os.mkdir(tmp_dir)
256
257			self.__extract_zip_file(zip_filename, tmp_dir)
258			files = self.__lees_test_cases_lijst(tmp_dir)
259
260			with open(csv_filename, 'w', encoding='utf-8') as handle:
261			csv_writer = csv.writer(handle, dialect=csv.unix_dialect)
262			self.__extract_files(csv_writer, files)
263
264			shutil.rmtree(tmp_dir)
265
266			return 0
267
268			# ----------------------------------------------------------------------------------------------------------------------
269

SetBased / py-kerapu

GitHub Access Token became invalid

TestShredderCommand.__maak_xpath() A last analyzed 2020-10-26 17:19 UTC

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

TestShredderCommand.__maak_xpath() A
last analyzed 2020-10-26 17:19 UTC