1

.gitignore

@@ -0,0 +1,161 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+.pytest_cache/
+cover/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+.pybuilder/
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+#   For a library or package, you might want to ignore these files since the code is
+#   intended to run in multiple environments; otherwise, check them in:
+# .python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+#Pipfile.lock
+
+# poetry
+#   Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
+#   This is especially recommended for binary packages to ensure reproducibility, and is more
+#   commonly ignored for libraries.
+#   https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
+#poetry.lock
+
+# pdm
+#   Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
+#pdm.lock
+#   pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
+#   in version control.
+#   https://pdm.fming.dev/#use-with-ide
+.pdm.toml
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# pytype static type analyzer
+.pytype/
+
+# Cython debug symbols
+cython_debug/
+
+# PyCharm
+#  JetBrains specific template is maintained in a separate JetBrains.gitignore that can
+#  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
+#  and can be added to the global gitignore or merged into this file.  For a more nuclear
+#  option (not recommended) you can uncomment the following to ignore the entire idea folder.
+#.idea/
+.history

src/data_classes.py

@@ -0,0 +1,59 @@
+from ast import Dict, List
+from dataclasses import dataclass
+import datetime
+
+future_exit_time = datetime.datetime(9999, 12, 31, 23, 59, 59)
+@dataclass
+class Event:
+    id: int
+    trailer_id: int
+    facility_id: int
+    entrance_time: datetime
+    exit_time: datetime
+    @classmethod
+    def from_dict(cls, data: Dict):
+        return cls(
+            id=data["id"],
+            trailer_id=data["trailer_id"],
+            facility_id =data["facility_id"],
+            entrance_time = datetime.fromisoformat(data["entrance_time"]),
+            exit_time = (datetime.fromisoformat(data["exit_time"]) if data["exit_time"] != None else future_exit_time),
+        )
+
+
+from dataclasses import dataclass, field
+from typing import List
+from datetime import datetime
+
+@dataclass
+class TimeWindow:
+    start_time: datetime
+    exit_time: datetime
+    events: List[Event] = field(default_factory=list)
+
+    @classmethod
+    def from_event(cls, event: Event):
+        return cls(
+            start_time=event.entrance_time,
+            exit_time=event.exit_time,
+            events=[event]
+        )
+
+    def get_events_count(self):
+        return len(self.events)
+    
+    def has_overlap_with_event(self, event: Event):
+        # when the event overlaps with the window from the left
+        if event.exit_time >=self.start_time and event.entrance_time <=self.start_time:
+            return True
+        # when the event overlaps with the window from the right
+        elif event.entrance_time <=self.exit_time and event.exit_time >=self.exit_time:
+            return True
+        # when the event is completely inside the window
+        elif event.entrance_time >=self.start_time and event.exit_time <=self.exit_time:
+            return True
+        else:
+            return False
+        
+    def is_exact_match(self, event: Event):
+        return event.entrance_time == self.start_time and event.exit_time == self.exit_time

src/main.py

@@ -0,0 +1,23 @@
+from datetime import datetime
+import json
+from data_classes import Event, TimeWindow
+from utils import filter_event_by_date_facility_id, find_max_continuous_sequence, loadDataFromJson, twenty_four_hours_bucket_time_series_by_minute
+
+def find_most_occupied_window(target_date:datetime, target_facility_id:int):
+    data = loadDataFromJson()
+    valid_events = [Event.from_dict(event) for event in data if filter_event_by_date_facility_id(Event.from_dict(event), target_date, target_facility_id)]
+    occupied_spots_per_minute,max_occupied_spots_by_events_by_minutes = twenty_four_hours_bucket_time_series_by_minute(valid_events,target_date)
+    _max,start_time, end_time = find_max_continuous_sequence(occupied_spots_per_minute)
+    #convert start and end time to datetime
+    start_time = datetime(target_date.year, target_date.month, target_date.day, start_time//60, start_time%60)
+    end_time = datetime(target_date.year, target_date.month, target_date.day, end_time//60, end_time%60)
+    max_sequence = (start_time.__str__(), end_time.__str__())
+    return ([start_time, end_time],max(occupied_spots_per_minute), occupied_spots_per_minute)
+
+target_date = datetime.fromisoformat("2022-01-18 00:00:00")
+target_facility_id = 1
+result = find_most_occupied_window(target_date, target_facility_id) 
+#problem 1
+print("Most number of slots at the same time is {}".format(result[1]))
+#problem 2
+print("Most occupied window is between {} and {} with {} occupied spots".format(result[0][0],result[0][1],result[1]))

src/utils.py

@@ -0,0 +1,68 @@
+import time
+from data_classes import Event, TimeWindow
+import datetime
+import json
+
+def twenty_four_hours_bucket_time_series_by_minute(valid_events:list[Event],target_date:datetime)->(list[int],list[list[Event]]): # type: ignore
+    #create a pre defined time window for 24 hours by minute precision
+    twenty_four_hour_by_minutes_array = [0]*(24*60)
+    twenty_four_hour_by_minutes_events_array = [[]]*(24*60)
+    default_start_time = 0
+    default_end_time = 24*60
+    for event in valid_events:
+        if(event.entrance_time.date() != target_date.date()):
+            start_time:time = default_start_time
+        else:
+            start_time:time = event.entrance_time.time().hour*60 + event.entrance_time.time().minute
+        if(event.exit_time.date() != target_date.date()):
+            end_time:time = default_end_time
+        else:
+            end_time:time = event.exit_time.time().hour*60 + event.exit_time.time().minute
+            
+        for i in range(start_time,end_time):
+            twenty_four_hour_by_minutes_array[i] += 1
+            twenty_four_hour_by_minutes_events_array[i].append(event)
+
+    return (twenty_four_hour_by_minutes_array,twenty_four_hour_by_minutes_events_array)
+
+def find_max_continuous_sequence(x):
+    # Find the maximum continuous sequence of the maximum element
+    max_element = max(x)
+    # Find the maximum continuous sequence of the maximum element
+    start = -1
+    end = -1
+    current_start = -1
+    current_length = 0
+    max_length = 0
+
+    for i in range(len(x)):
+        if x[i] == max_element:
+            if current_start == -1:
+                current_start = i
+            current_length += 1
+        else:
+            if current_length > max_length:
+                max_length = current_length
+                start = current_start
+                end = i - 1
+            current_start = -1
+            current_length = 0
+
+    # Check if the last sequence is the maximum continuous sequence
+    if current_length > max_length:
+        start = current_start
+        end = len(x) - 1
+
+    return (max_element, start, end)
+
+def loadDataFromJson():
+    with open('data/events.json') as f:
+        data = json.load(f)
+    return data
+
+def filter_event_by_date_facility_id(event:Event,target_date:datetime, facility_id:int):
+    if event.facility_id == facility_id: 
+        if event.entrance_time.date() <= target_date.date() and event.exit_time.date() >= target_date.date():
+            return True
+    else:
+        return False