WIP

anyway/infographics_utils.py

@@ -11,7 +11,6 @@
 # noinspection PyProtectedMember
 from flask_babel import _
 
-from anyway.app_and_db import db
 from anyway.request_params import (
     RequestParams,
     get_news_flash_location_text,

anyway/widgets/segment_junctions.py

@@ -0,0 +1,50 @@
+import logging
+from typing import Dict, List
+
+from app_and_db import db
+from models import RoadJunctionKM, RoadSegments
+
+
+class SegmentJunctions:
+    # {<road segment id>: [<non urban junction id, ...]}
+    __segment_junctions: Dict[int, List[int]] = {}
+    __singelton = None
+
+    def __init__(self):
+        if not self.__segment_junctions:
+            self.__segment_junctions = self.__calc_fill_segment_junctions()
+
+    @staticmethod
+    def get_instance():
+        if not SegmentJunctions.__singelton:
+            SegmentJunctions.__singelton = SegmentJunctions()
+        return SegmentJunctions.__singelton
+
+    def get_segment_junctions(self, segment: int) -> List[int]:
+        res = self.__segment_junctions.get(segment)
+        if res is None:
+            logging.warning(f"{segment}: no such segment in segment junctions data.")
+        return res or []
+
+    @staticmethod
+    def __calc_fill_segment_junctions():
+        tmp: List[RoadJunctionKM] = db.session.query(RoadJunctionKM).all()
+        res = {}
+        rkj = {}
+        for t in tmp:
+            if t.road not in rkj:
+                rkj[t.road] = {}
+            rkj[t.road][t.km] = t.non_urban_intersection
+        tmp: List[RoadSegments] = db.session.query(RoadSegments).all()
+        segments = {t.segment_id: t for t in tmp}
+        for k, v in segments.items():
+            if v.road not in rkj:
+                logging.warning(f"No junctions in road {v.road}.")
+                continue
+            # logger.debug(f"seg:{k},road:{v.road}:juncs:{rkj[v.road]}.")
+            junctions = [
+                rkj[v.road][km] for km in rkj[v.road].keys() if v.from_km - 1 <= km <= v.to_km + 1
+            ]
+            # logger.debug(f"segment junctions:{junctions}")
+            res[k] = junctions
+        return res

anyway/widgets/widget_utils.py

@@ -14,8 +14,8 @@
 from anyway.request_params import LocationInfo
 from anyway.vehicle_type import VehicleType
 from anyway.parsers import resolution_dict
-from anyway.models import RoadJunctionKM, RoadSegments
 from anyway.models import NewsFlash
+from widgets.segment_junctions import SegmentJunctions
 
 
 def get_query(table_obj, filters, start_time, end_time):
@@ -43,6 +43,7 @@ def get_expression_for_fields(filters, table_obj, op):
         ex = op(ex, get_filter_expression(table_obj, field_name, value))
     return ex
 
+
 # todo: remove road_segment_name if road_segment_id exists.
 def get_expression_for_location_fields(filters, table_obj):
     ex = get_expression_for_fields(filters, table_obj, and_)
@@ -286,47 +287,3 @@ def newsflash_has_location(newsflash: NewsFlash):
         resolution == BE_CONST.ResolutionCategories.SUBURBAN_ROAD.value
         and newsflash.road_segment_name
     ) or (resolution == BE_CONST.ResolutionCategories.STREET.value and newsflash.street1_hebrew)
-
-
-class SegmentJunctions:
-    # {<road segment id>: [<non urban junction id, ...]}
-    __segment_junctions: Dict[int, List[int]] = {}
-    __singelton = None
-
-    def __init__(self):
-        if not self.__segment_junctions:
-            self.__segment_junctions = self.__calc_fill_segment_junctions()
-
-    @staticmethod
-    def get_instance():
-        if not SegmentJunctions.__singelton:
-            SegmentJunctions.__singelton = SegmentJunctions()
-        return SegmentJunctions.__singelton
-
-    def get_segment_junctions(self, segment: int) -> List[int]:
-        res = self.__segment_junctions.get(segment)
-        if res is None:
-            logging.warning(f"{segment}: no such segment in segment junctions data.")
-        return res or []
-
-    @staticmethod
-    def __calc_fill_segment_junctions():
-        tmp: List[RoadJunctionKM] = db.session.query(RoadJunctionKM).all()
-        res = {}
-        rkj = {}
-        for t in tmp:
-            if t.road not in rkj:
-                rkj[t.road] = {}
-            rkj[t.road][t.km] = t.non_urban_intersection
-        tmp: List[RoadSegments] = db.session.query(RoadSegments).all()
-        segments = {t.segment_id: t for t in tmp}
-        for k, v in segments.items():
-            if v.road not in rkj:
-                logging.warning(f"No junctions in road {v.road}.")
-                continue
-            # logger.debug(f"seg:{k},road:{v.road}:juncs:{rkj[v.road]}.")
-            junctions = [rkj[v.road][km] for
-                         km in rkj[v.road].keys() if v.from_km-1 <= km <= v.to_km+1]
-            # logger.debug(f"segment junctions:{junctions}")
-            res[k] = junctions
-        return res

tests/test_infographics_utils.py

@@ -43,7 +43,7 @@ def test_format_two_level_items(self):
         self.assertEqual(self.items1_res, actual, "leve 2 formatting")
 
     def test_segment_junctions(self):
-        from anyway.infographics_utils import SegmentJunctions
+        from anyway.widgets.segment_junctions import SegmentJunctions
         i = SegmentJunctions.get_instance()
         segment_junctions = i.get_segment_junctions(1)
         self.assertTrue(isinstance(segment_junctions, list))