il2fb.parsers.mission.MissionParser._get_scouting() - Code Metrics - Inspection of "refactor MissionParser._get_scouting()" - IL2HorusTeam/il2fb-mission-parser - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 2095ec...4c8271 )

by Oleksandr

created 2016-01-06 12:01 UTC

_get_scouting() B

↳ Parent: il2fb.parsers.mission.MissionParser

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	9
CRAP Score	5

Metric	Value
cc	5
dl	0
loc	17
ccs	9
cts	9
cp	1
crap	5
rs	8.5455

# -*- coding: utf-8 -*-

import six
import sys

from .exceptions import MissionParsingError
from .sections.main import MainSectionParser
from .sections.season import SeasonSectionParser
from .sections.weather import WeatherSectionParser
from .sections.respawn_time import RespawnTimeSectionParser
from .sections.mds import MDSSectionParser, MDSScoutsSectionParser
from .sections.chiefs import ChiefsSectionParser, ChiefRoadSectionParser
from .sections.nstationary import NStationarySectionParser
from .sections.buildings import BuildingsSectionParser
from .sections.target import TargetSectionParser
from .sections.born_place import (
    BornPlaceSectionParser, BornPlaceAircraftsSectionParser,
    BornPlaceAirForcesSectionParser,
)
from .sections.static_camera import StaticCameraSectionParser
from .sections.front_marker import FrontMarkerSectionParser
from .sections.rocket import RocketSectionParser
from .sections.wing import (
    FlightSectionParser, FlightInfoSectionParser, FlightRouteSectionParser,
)
from .utils import move_if_present, set_if_present, strip_comments


class MissionParser(object):
    """
    Parses a whole mission file.
    View :ref:`detailed description <mission-parser>`.
    """

    def __init__(self):
        self.parsers = [
            MainSectionParser(),
            SeasonSectionParser(),
            WeatherSectionParser(),
            RespawnTimeSectionParser(),
            MDSSectionParser(),
            MDSScoutsSectionParser(),
            ChiefsSectionParser(),
            ChiefRoadSectionParser(),
            NStationarySectionParser(),
            BuildingsSectionParser(),
            TargetSectionParser(),
            BornPlaceSectionParser(),
            BornPlaceAircraftsSectionParser(),
            BornPlaceAirForcesSectionParser(),
            StaticCameraSectionParser(),
            FrontMarkerSectionParser(),
            RocketSectionParser(),
            FlightSectionParser(),
            FlightRouteSectionParser(),
        ]
        self.flight_info_parser = FlightInfoSectionParser()

    def parse(self, mission):
        if isinstance(mission, six.string_types):
            with open(mission, 'r') as f:
                return self.parse_stream(f)
        else:
            return self.parse_stream(mission)

    def parse_stream(self, sequence):
        self._current_parser = None
        self.data = {}

        for i, line in enumerate(sequence):
            line = strip_comments(line)
            if self.is_section_name(line):
                self._finalize_current_parser()
                section_name = self.get_section_name(line)
                self._current_parser = self._get_parser(section_name)
            elif self._current_parser:
                self._try_to_parse_line(i, line)

        self._finalize_current_parser()
        return self._clean()

    @staticmethod
    def is_section_name(line):
        return line.startswith('[') and line.endswith(']')

    @staticmethod
    def get_section_name(line):
        return line.strip('[]')

    def _get_parser(self, section_name):
        parser = self.flight_info_parser
        flights = self.data.get('flights')

        if flights is not None and parser.start(section_name):
            return parser

        for parser in self.parsers:
            if parser.start(section_name):
                return parser

        return None

    def _finalize_current_parser(self):
        if not self._current_parser:
            return
        try:
            data = self._current_parser.stop()
        except Exception:
            error_type, original_msg, traceback = sys.exc_info()
            msg = (
                "{0} during finalization of \"{1}\": {2}"
                .format(error_type.__name__,
                        self._current_parser.__class__.__name__,
                        original_msg))
            self._raise_error(msg, traceback)
        else:
            self.data.update(data)
        finally:
            self._current_parser = None

    def _try_to_parse_line(self, line_number, line):
        try:
            self._current_parser.parse_line(line)
        except Exception:
            error_type, original_msg, traceback = sys.exc_info()
            msg = (
                "{0} in line #{1} (\"{2}\"): {3}"
                .format(error_type.__name__, line_number, line, original_msg))
            self._raise_error(msg, traceback)

    @staticmethod
    def _raise_error(message, traceback):
        error = MissionParsingError(message)
        six.reraise(MissionParsingError, error, traceback)

    def _clean(self):
        result = {}

        move_if_present(result, self.data, 'location_loader')
        move_if_present(result, self.data, 'player')
        move_if_present(result, self.data, 'targets')

        set_if_present(result, 'conditions', self._get_conditions())
        set_if_present(result, 'objects', self._get_objects())

        return result

    def _get_conditions(self):
        result = {}

        set_if_present(result, 'time_info', self._get_time_info())
        set_if_present(result, 'meteorology', self._get_meteorology())
        set_if_present(result, 'scouting', self._get_scouting())

        move_if_present(result, self.data, 'respawn_time')

        if 'conditions' in self.data:
            conditions = self.data['conditions']

            move_if_present(result, conditions, 'radar')
            move_if_present(result, conditions, 'communication')
            move_if_present(result, conditions, 'home_bases')
            move_if_present(result, conditions, 'crater_visibility_muptipliers')

        return result

    def _get_time_info(self):
        result = {}

        move_if_present(result, self.data, 'date')
        if 'time' in self.data:
            result.update({
                'time': self.data['time']['value'],
                'is_fixed': self.data['time']['is_fixed'],
            })

        return result

    def _get_meteorology(self):
        result = {}

        move_if_present(result, self.data, 'weather', 'weather_conditions')
        move_if_present(result, self.data, 'cloud_base')

        if 'weather' in self.data:
            result.update(self.data.pop('weather'))

        return result

    def _get_scouting(self):
        try:
            result = self.data['conditions'].pop('scouting')
        except KeyError:
            result = {}

        keys = [
            key for key in self.data.keys()
            if key.startswith(MDSScoutsSectionParser.output_prefix)
        ]
        scouts = {
            self.data[key]['belligerent']: self.data[key]['aircrafts']
            for key in keys
        }
        set_if_present(result, 'scouts', scouts)

        return result

    def _get_objects(self):
        result = {}

        set_if_present(result, 'moving_units', self._get_moving_units())
        set_if_present(result, 'flights', self._get_flights())
        set_if_present(result, 'home_bases', self._get_home_bases())

        move_if_present(result, self.data, 'stationary')
        move_if_present(result, self.data, 'buildings')
        move_if_present(result, self.data, 'cameras')
        move_if_present(result, self.data, 'markers')
        move_if_present(result, self.data, 'rockets')

        return result

    def _get_moving_units(self):
        units = self.data.pop('moving_units', [])
        for unit in units:
            key = "{}{}".format(ChiefRoadSectionParser.output_prefix, unit['id'])
            unit['route'] = self.data.pop(key, [])
        return units

    def _get_flights(self):
        keys = self.data.pop('flights', [])
        flights = [self.data.pop(key) for key in keys if key in self.data]
        for flight in flights:
            key = "{}{}".format(FlightRouteSectionParser.output_prefix, flight['id'])
            flight['route'] = self.data.pop(key, [])
        return flights

    def _get_home_bases(self):
        home_bases = self.data.pop('home_bases', [])
        for i, home_base in enumerate(home_bases):
            key = "{}{}".format(BornPlaceAircraftsSectionParser.output_prefix, i)
            home_base['spawning']['aircraft_limitations']['allowed_aircrafts'] = self.data.pop(key, [])

            key = "{}{}".format(BornPlaceAirForcesSectionParser.output_prefix, i)
            home_base['spawning']['allowed_air_forces'] = self.data.pop(key, [])
        return home_bases


1		# -- coding: utf-8 --
2
3	1	import six
4	1	import sys
5
6	1	from .exceptions import MissionParsingError
7	1	from .sections.main import MainSectionParser
8	1	from .sections.season import SeasonSectionParser
9	1	from .sections.weather import WeatherSectionParser
10	1	from .sections.respawn_time import RespawnTimeSectionParser
11	1	from .sections.mds import MDSSectionParser, MDSScoutsSectionParser
12	1	from .sections.chiefs import ChiefsSectionParser, ChiefRoadSectionParser
13	1	from .sections.nstationary import NStationarySectionParser
14	1	from .sections.buildings import BuildingsSectionParser
15	1	from .sections.target import TargetSectionParser
16	1	from .sections.born_place import (
17		BornPlaceSectionParser, BornPlaceAircraftsSectionParser,
18		BornPlaceAirForcesSectionParser,
19		)
20	1	from .sections.static_camera import StaticCameraSectionParser
21	1	from .sections.front_marker import FrontMarkerSectionParser
22	1	from .sections.rocket import RocketSectionParser
23	1	from .sections.wing import (
24		FlightSectionParser, FlightInfoSectionParser, FlightRouteSectionParser,
25		)
26	1	from .utils import move_if_present, set_if_present, strip_comments
27
28
29	1	class MissionParser(object):
30		"""
31		Parses a whole mission file.
32		View :ref:`detailed description <mission-parser>`.
33		"""
34
35	1	def __init__(self):
36	1	self.parsers = [
37		MainSectionParser(),
38		SeasonSectionParser(),
39		WeatherSectionParser(),
40		RespawnTimeSectionParser(),
41		MDSSectionParser(),
42		MDSScoutsSectionParser(),
43		ChiefsSectionParser(),
44		ChiefRoadSectionParser(),
45		NStationarySectionParser(),
46		BuildingsSectionParser(),
47		TargetSectionParser(),
48		BornPlaceSectionParser(),
49		BornPlaceAircraftsSectionParser(),
50		BornPlaceAirForcesSectionParser(),
51		StaticCameraSectionParser(),
52		FrontMarkerSectionParser(),
53		RocketSectionParser(),
54		FlightSectionParser(),
55		FlightRouteSectionParser(),
56		]
57	1	self.flight_info_parser = FlightInfoSectionParser()
58
59	1	def parse(self, mission):
60	1	if isinstance(mission, six.string_types):
61	1	with open(mission, 'r') as f:
62	1	return self.parse_stream(f)
63		else:
64	1	return self.parse_stream(mission)
65
66	1	def parse_stream(self, sequence):
67	1	self._current_parser = None
68	1	self.data = {}
69
70	1	for i, line in enumerate(sequence):
71	1	line = strip_comments(line)
72	1	if self.is_section_name(line):
73	1	self._finalize_current_parser()
74	1	section_name = self.get_section_name(line)
75	1	self._current_parser = self._get_parser(section_name)
76	1	elif self._current_parser:
77	1	self._try_to_parse_line(i, line)
78
79	1	self._finalize_current_parser()
80	1	return self._clean()
81
82	1	@staticmethod
83		def is_section_name(line):
84	1	return line.startswith('[') and line.endswith(']')
85
86	1	@staticmethod
87		def get_section_name(line):
88	1	return line.strip('[]')
89
90	1	def _get_parser(self, section_name):
91	1	parser = self.flight_info_parser
92	1	flights = self.data.get('flights')
93
94	1	if flights is not None and parser.start(section_name):
95	1	return parser
96
97	1	for parser in self.parsers:
98	1	if parser.start(section_name):
99	1	return parser
100
101	1	return None
102
103	1	def _finalize_current_parser(self):
104	1	if not self._current_parser:
105	1	return
106	1	try:
107	1	data = self._current_parser.stop()
108	1	except Exception:
109	1	error_type, original_msg, traceback = sys.exc_info()
110	1	msg = (
111		"{0} during finalization of \"{1}\": {2}"
112		.format(error_type.__name__,
113		self._current_parser.__class__.__name__,
114		original_msg))
115	1	self._raise_error(msg, traceback)
116		else:
117	1	self.data.update(data)
118		finally:
119	1	self._current_parser = None
120
121	1	def _try_to_parse_line(self, line_number, line):
122	1	try:
123	1	self._current_parser.parse_line(line)
124	1	except Exception:
125	1	error_type, original_msg, traceback = sys.exc_info()
126	1	msg = (
127		"{0} in line #{1} (\"{2}\"): {3}"
128		.format(error_type.__name__, line_number, line, original_msg))
129	1	self._raise_error(msg, traceback)
130
131	1	@staticmethod
132		def _raise_error(message, traceback):
133	1	error = MissionParsingError(message)
134	1	six.reraise(MissionParsingError, error, traceback)
135
136	1	def _clean(self):
137	1	result = {}
138
139	1	move_if_present(result, self.data, 'location_loader')
140	1	move_if_present(result, self.data, 'player')
141	1	move_if_present(result, self.data, 'targets')
142
143	1	set_if_present(result, 'conditions', self._get_conditions())
144	1	set_if_present(result, 'objects', self._get_objects())
145
146	1	return result
147
148	1	def _get_conditions(self):
149	1	result = {}
150
151	1	set_if_present(result, 'time_info', self._get_time_info())
152	1	set_if_present(result, 'meteorology', self._get_meteorology())
153	1	set_if_present(result, 'scouting', self._get_scouting())
154
155	1	move_if_present(result, self.data, 'respawn_time')
156
157	1	if 'conditions' in self.data:
158	1	conditions = self.data['conditions']
159
160	1	move_if_present(result, conditions, 'radar')
161	1	move_if_present(result, conditions, 'communication')
162	1	move_if_present(result, conditions, 'home_bases')
163	1	move_if_present(result, conditions, 'crater_visibility_muptipliers')
164
165	1	return result
166
167	1	def _get_time_info(self):
168	1	result = {}
169
170	1	move_if_present(result, self.data, 'date')
171	1	if 'time' in self.data:
172	1	result.update({
173		'time': self.data['time']['value'],
174		'is_fixed': self.data['time']['is_fixed'],
175		})
176
177	1	return result
178
179	1	def _get_meteorology(self):
180	1	result = {}
181
182	1	move_if_present(result, self.data, 'weather', 'weather_conditions')
183	1	move_if_present(result, self.data, 'cloud_base')
184
185	1	if 'weather' in self.data:
186	1	result.update(self.data.pop('weather'))
187
188	1	return result
189
190	1	def _get_scouting(self):
191	1	try:
192	1	result = self.data['conditions'].pop('scouting')
193	1	except KeyError:
194	1	result = {}
195
196	1	keys = [
197		key for key in self.data.keys()
198		if key.startswith(MDSScoutsSectionParser.output_prefix)
199		]
200	1	scouts = {
201		self.data[key]['belligerent']: self.data[key]['aircrafts']
202		for key in keys
203		}
204	1	set_if_present(result, 'scouts', scouts)
205
206	1	return result
207
208	1	def _get_objects(self):
209	1	result = {}
210
211	1	set_if_present(result, 'moving_units', self._get_moving_units())
212	1	set_if_present(result, 'flights', self._get_flights())
213	1	set_if_present(result, 'home_bases', self._get_home_bases())
214
215	1	move_if_present(result, self.data, 'stationary')
216	1	move_if_present(result, self.data, 'buildings')
217	1	move_if_present(result, self.data, 'cameras')
218	1	move_if_present(result, self.data, 'markers')
219	1	move_if_present(result, self.data, 'rockets')
220
221	1	return result
222
223	1	def _get_moving_units(self):
224	1	units = self.data.pop('moving_units', [])
225	1	for unit in units:
226	1	key = "{}{}".format(ChiefRoadSectionParser.output_prefix, unit['id'])
227	1	unit['route'] = self.data.pop(key, [])
228	1	return units
229
230	1	def _get_flights(self):
231	1	keys = self.data.pop('flights', [])
232	1	flights = [self.data.pop(key) for key in keys if key in self.data]
233	1	for flight in flights:
234	1	key = "{}{}".format(FlightRouteSectionParser.output_prefix, flight['id'])
235	1	flight['route'] = self.data.pop(key, [])
236	1	return flights
237
238	1	def _get_home_bases(self):
239	1	home_bases = self.data.pop('home_bases', [])
240	1	for i, home_base in enumerate(home_bases):
241	1	key = "{}{}".format(BornPlaceAircraftsSectionParser.output_prefix, i)
242	1	home_base['spawning']['aircraft_limitations']['allowed_aircrafts'] = self.data.pop(key, [])
243
244	1	key = "{}{}".format(BornPlaceAirForcesSectionParser.output_prefix, i)
245	1	home_base['spawning']['allowed_air_forces'] = self.data.pop(key, [])
246		return home_bases
247

IL2HorusTeam / il2fb-mission-parser

Push — master ( 2095ec...4c8271 )

_get_scouting() B

Complexity

Size

Duplication

Code Coverage

Duplication Side-by-Side

Filter issues like