Fix some flake8 warnings

abuse_checker.py

@@ -5,7 +5,7 @@
 from data_types import AbuseRule
 
 
-def last_time(delta:datetime.timedelta):
+def last_time(delta: datetime.timedelta):
     return datetime.datetime.now() - delta
 
 

cds.py

@@ -22,6 +22,7 @@
 
 try:
     import settings
+
     LOAD_TEST_DATA = settings.LOAD_TEST_DATA
 except ImportError:
     LOAD_TEST_DATA = os.environ.get('LOAD_TEST_DATA', False)
@@ -158,7 +159,6 @@ def get_closest_bus_stop(self, bus_info: CdsRouteBus):
             bus_positions.append(bus_info.get_bus_position())
         closest_on_route = self.get_closest_bus_stop_checked(bus_info.route_name_, bus_positions)
 
-
         if closest_on_route and bus_info.distance(closest_on_route) < threshold:
             return closest_on_route
 
@@ -248,6 +248,7 @@ def load_all_cds_buses_from_db(self) -> List[CdsRouteBus]:
         def update_last_bus_data(buses):
             for bus in buses:
                 self.add_last_bus_data(bus.name_, bus.get_bus_position())
+
         while self.fetching_in_progress:
             self.logger.info("Waiting for previous DB query")
             time.sleep(1)
@@ -437,7 +438,8 @@ def count_buses(buses: Iterable[CdsRouteBus], time_interval):
             if full_info:
                 short_result = [d for d in cds_buses if time_check(d, now - timedelta(minutes=10))]
                 sort_routes = sorted(short_result, key=lambda x: natural_sort_key(x.route_name_))
-                grouped = [(k, len(list(g))) for k, g in groupby(sort_routes, lambda x: f'{x.route_name_:5s} ({x.proj_id_:3d})')]
+                grouped = [(k, len(list(g))) for k, g in
+                           groupby(sort_routes, lambda x: f'{x.route_name_:5s} ({x.proj_id_:3d})')]
                 buses_list += (('{:10s} => {}'.format(i[0], i[1])) for i in grouped)
             buses_list.append(bus_stats_text)
             text = '\n'.join(buses_list)

data_providers.py

@@ -22,7 +22,7 @@
 except ImportError:
     settings = None
     env = os.environ
-    if all((x in env for x in ("CDS_HOST", "CDS_DB_PATH", "CDS_USER", "CDS_PASS", ))):
+    if all((x in env for x in ("CDS_HOST", "CDS_DB_PATH", "CDS_USER", "CDS_PASS",))):
         CDS_HOST = env['CDS_HOST']
         CDS_DB_PATH = env['CDS_DB_PATH']
         CDS_USER = env['CDS_USER']
@@ -33,6 +33,7 @@
 
 class CdsDBDataProvider(CdsBaseDataProvider):
     CACHE_TIMEOUT = 30
+
     def __init__(self, logger):
         self.logger = logger
         self.cds_db = fdb.connect(host=CDS_HOST, database=CDS_DB_PATH, user=CDS_USER,
@@ -82,6 +83,7 @@ def make_names_lower(x):
 
 class CdsTestDataProvider(CdsBaseDataProvider):
     CACHE_TIMEOUT = 0.0001
+
     def __init__(self, logger):
         self.logger = logger
         self.test_data_files = []
@@ -122,4 +124,3 @@ def load_all_cds_buses(self) -> List[CdsRouteBus]:
 
 def get_data_provider(logger):
     return CdsTestDataProvider(logger) if LOAD_TEST_DATA else CdsDBDataProvider(logger)
-

data_types.py

@@ -163,4 +163,4 @@ class StatsData(NamedTuple):
     min30: int
     min60: int
     total: int
-    text: str
+    text: str

helpers.py

@@ -96,9 +96,10 @@ def azimuth(glon1, glat1, glon2, glat2):
 def parse_int(s, default=0):
     try:
         return int(s), True
-    except:
+    except Exception:
         return default, False
 
+
 def distance_km(glat1, glon1, glat2, glon2):
     r = 6373.0
 

test_cds.py

@@ -15,8 +15,6 @@
 logger = logging.getLogger("vrnbus")
 
 
-
-
 class CdsRouteTestCase(unittest.TestCase):
     def __init__(self, *args, **kwargs):
         super(CdsRouteTestCase, self).__init__(*args, **kwargs)
@@ -86,7 +84,6 @@ def test_closest_bus_stop(self):
         logger.info(f"{station}; {route_bus.distance_km(station):.4f}  {route_bus.distance(station):.4f}")
 
 
-
 class CdsDataGatheringTestCase(unittest.TestCase):
     def __init__(self, *args, **kwargs):
         super(CdsDataGatheringTestCase, self).__init__(*args, **kwargs)
@@ -106,6 +103,7 @@ def call_common_methods(self, cds):
         all_data = cds.load_all_cds_buses_from_db()
         cds.calc_avg_speed()
 
+
 class CdsSpeedTestCase(unittest.TestCase):
     def __init__(self, *args, **kwargs):
         super(CdsSpeedTestCase, self).__init__(*args, **kwargs)
@@ -126,6 +124,7 @@ def test_speed_businfo(self):
         finish = datetime.datetime.now()
         logger.info(f"{finish - start}")
 
+
 class CdsBusStopIndexTestCase(unittest.TestCase):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
@@ -160,6 +159,7 @@ def test_routes_on_near_stations(self):
         routes_2 = self.cds.get_routes_on_bus_stop("Рабочий проспект (из центра)")
         self.assertListEqual(routes_1, routes_2)
 
+
 class CdsBusArrivalTestCases(unittest.TestCase):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
@@ -176,7 +176,7 @@ def test_arrival(self):
         result = self.processor.get_arrival("про", 51.692727, 39.18297)
         stops = result['bus_stops']
         counts = 0
-        for k,v in stops.items():
+        for k, v in stops.items():
             self.logger.info(k)
             self.logger.info(v)
             counts += len(v.split('\n'))
@@ -187,5 +187,6 @@ def test_arrival_distance(self):
         result = self.cds.get_dist("27", 'Площадь Застава (в центр)', 'Центральный автовокзал (в центр)')
         self.logger.info(result)
 
+
 if __name__ == '__main__':
     unittest.main()

test_helpers.py

@@ -37,6 +37,5 @@ def test_azimuth(self):
                 self.assertEqual(f(*params), result)
 
 
-
 if __name__ == '__main__':
     unittest.main()

test_tracking.py

@@ -13,10 +13,13 @@
                     handlers=[logging.StreamHandler()])
 
 logger = logging.getLogger("vrnbus")
+
+
 class FakeUser():
     def __init__(self, id=None):
         self.id = id if id else 42
 
+
 class TrackingEventTest(unittest.TestCase):
     def test_event_parsing(self):
         f = get_event_by_name
@@ -39,6 +42,7 @@ def test_event_parsing(self):
             with self.subTest(f'{event_name}, {result}'):
                 self.assertEqual(f(event_name), result)
 
+
 class TrackingTest(unittest.TestCase):
     def test_something(self):
         tracker = EventTracker(logger)
@@ -63,8 +67,8 @@ def test_detailed_stats(self):
 
         tracker.tg(TgEvent.START, user)
         for i in range(500):
-            tracker.tg(TgEvent(i%8 + 1), FakeUser(100500 + i%7))
-            tracker.web(WebEvent(i%3 + 1), f'127.0.0.{i%3}')
+            tracker.tg(TgEvent(i % 8 + 1), FakeUser(100500 + i % 7))
+            tracker.web(WebEvent(i % 3 + 1), f'127.0.0.{i%3}')
         stats = tracker.stats()
         detailed_stats = tracker.stats(True)
 
@@ -103,4 +107,4 @@ def run_rules_check(self, check_result):
             checker.add_user_event(WebEvent.BUSINFO, user_id)
 
         self.assertEqual(checker.check_user(WebEvent.BUSMAP, user_id), check_result)
-        self.assertEqual(checker.check_user(WebEvent.BUSINFO, user_id), check_result)
+        self.assertEqual(checker.check_user(WebEvent.BUSINFO, user_id), check_result)