parse_shifts

parse_shifts0.py

import csv
from datetime import time, datetime as dt

DAY_NAMES = ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"]

def find_shift_agents(csv_filename, search_dt):
    shifts = [[] for day in DAY_NAMES]
    with open(csv_filename) as f:
        for name, shift_ranges in csv.reader(f):
            for shift_range in shift_ranges.split("  / "):
                # e.g. "Mon-Wed 5 pm - 12 am"
                day_range, time_range = shift_range.split(" ", 1)
                start_day, stop_day = map(DAY_NAMES.index, day_range.split("-"))
                if stop_day < start_day:
                    days = [*range(stop_day + 1), *range(start_day, 7)]
                else:
                    days = range(start_day, stop_day + 1)
                start_time, stop_time = (
                    dt.strptime(
                        t,
                        "%I{} %p".format(":%M" if ":" in t else "")
                    )
                    for t in time_range.split(" - ")
                )
                for day in days:
                    shifts[day].append((name, start_time, stop_time))
    search_time = search_dt.replace(year=1900, month=1, day=1)
    agents = [
        name
        for (name, start, stop) in shifts[search_dt.weekday()]
        if (start <= search_time <= stop
            or start <= search_time and stop == dt(1900, 1, 1))
    ]
    return agents

parse_shifts1.py

from typing import List, Tuple, Iterator
from datetime import datetime as dt, timedelta as delta
import csv
import pdb


DAY_NAMES = ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"] # type: list

TimeRange = Tuple[dt, dt]

def parse_shift_range(shift_range: str, epoch: dt) -> Iterator[TimeRange]:
    """
    epoch is the midnight of the monday of the week we're talking about
    >>> pdb.set_trace()
    >>> tr = parse_shift_range("Mon-Tue 11:30 am - 7 pm", dt(2019, 5, 6))
    >>> [(t1.isoformat(), t2.isoformat()) for (t1, t2) in tr]
    [('2019-05-06T11:30:00', '2019-05-06T19:00:00'), \
('2019-05-07T11:30:00', '2019-05-07T19:00:00')]

    """
    days, times = shift_range.split(" ", 1)

    start_day, stop_day = (DAY_NAMES.index(day) for day in days.split("-"))
    start, stop = (
        dt.strptime(
            t,
            "%I{} %p".format(":%M" if ":" in t else "")
        ) - dt(1900, 1, 1) # strptime sets Jan 1 1900 as the default date
        for t in times.split(" - ")
    )
    if start > stop and stop == delta(0):
        stop = delta(1)
        # C strptime uses the convention that 12 am refers to the midnight of
        # the day preceeding day, not the next day
        # in the examples provided, "5 pm - 12 am" is used
        # it is understood that 12 am means 24:00 PM in this context
    if start_day <= stop_day:
        offsets = [*range(start_day, stop_day + 1)]
    else:
        offsets = [*range(0, stop_day + 1), *range(start_day, 7)]
    for offset in offsets:
        date = epoch + delta(offset)
        yield (date + start, date + stop)


def find_shift_agents(csv_filename: str, search_dt: dt) -> List[str]:
    # e.g. search_dt is 2019-05-10T11:41:46
    # take date (2019-05-10), subtract weekday offset (Mon=0, Sun=6)
    # as number of days to go back in time
    epoch_date = search_dt.date() - delta(search_dt.weekday())
    # then revert to dt for 0:00
    epoch = dt(epoch_date.year, epoch_date.month, epoch_date.day)
    with open(csv_filename) as f:
        reader = csv.reader(f)
        shifts = [] # type: List[Tuple[dt, dt, str]]
        for (name, shift_ranges) in reader:
            for shift_range in shift_ranges.split("  / "):
                shifts.extend(
                    (*times, name)
                    for times in parse_shift_range(shift_range, epoch)
                )
    working_agents = [
        name
        for (start, stop, name) in shifts
        if start <= search_dt <= stop
    ]
    return working_agents

parse_shifts2.py

import csv
from datetime import time
from dateutil.parser import parse


parse_date = "Mon Tue Wed Thu Fri Sat Sun".split().index

def parse_time(t):
    return parse(t).time()

def find_shift_agents(csv_filename, search_datet):
    search_date, search_time = search_datet.weekday(), search_datet.time()
    for name, shift_ranges in csv.reader(open(csv_filename)):
        for shift_range in shift_ranges.split("/ "):
            date_range, time_range = (
                range.split("-") for range in shift_range.split(" ", 1)
            )
            start_date, stop_date = map(parse_date, date_range)
            start_time, stop_time = map(parse_time, time_range)
            if (
                    (
                        all if start_date < start_date else any
                    )(
                        [start_date <= search_date, search_date <= stop_date]
                    )
                    and start_time <= search_time
                    and (search_time <= stop_time or (stop_time == time(0)))
                ):
                    yield name

shift_schedule1.csv

shift_schedule2.csv

test_cases.py

import os
import csv
import pdb
from importlib import import_module
from datetime import datetime as dt
from typing import List, Tuple, Callable, Iterator
import pytest
import mypy.api


from parse_shifts1 import parse_shift_range

def test_parse_range():
    tr = parse_shift_range("Mon-Tue 11:30 am - 7 pm", dt(2019, 5, 6))
    assert [
        (t1.isoformat(), t2.isoformat()) for (t1, t2) in tr
    ] == [
        ("2019-05-06T11:30:00", "2019-05-06T19:00:00"),
        ("2019-05-07T11:30:00", "2019-05-07T19:00:00")
    ]


IMPL_FNAMES = [
    fname
    for fname in os.listdir(".")
    if fname.startswith("parse_shifts") and fname.endswith(".py")
]

IMPL_MODULES = [
    import_module(fname.split(".")[0])
    for fname in IMPL_FNAMES
]

def exhaust(iterator: Iterator) -> Callable:
    def exhausted_iterator(*args, **kwargs):
        return list(iterator(*args, **kwargs))
    return exhausted_iterator

IMPLS = [
    exhaust(module.find_shift_agents)
    for module in IMPL_MODULES
    if "find_shift_agents" in dir(module)
]

@pytest.fixture(params=IMPLS)
def impl(request) -> Callable:
    return request.param

FILES = [
    ("straightforward.csv", [["John", "Mon-Fri 9 am - 5 pm"]]),
    ("tricky_times.csv", [["John", "Mon-Fri 9 am - 4:59 pm"]]),
    ("tricky_days.csv", [["John", "Sun-Fri 9 am - 5 pm"]]),
    ("multiple.csv", [["John", "Mon-Wed 9 am - 4:59 pm  / Thu-Fri 9 am - 4 pm"]])
]

@pytest.fixture(params=FILES)
def fname(request) -> str:
    fname, rows = request.param
    with open(fname, "w") as f:
        writer = csv.writer(f)
        writer.writerows(rows)
    f.close()
    yield fname
    os.remove(fname)

CASES = [
    ("saturday", dt(2019, 5, 11, 12), []),
    ("before", dt(2019, 5, 14, 8, 59), []), # all the rest are Tuesdays
    ("after", dt(2019, 5, 14, 17, 1), []),
    ("simple", dt(2019, 5, 14, 12), ["John"]),
    ("past", dt(100, 1, 7, 12), ["John"]),
    ("future", dt(3666, 1, 5, 12), ["John"])
]

@pytest.fixture(params=CASES)
def case(request) -> Tuple[str, dt, List[str]]:
    return request.param

def test_generated_cases(fname: str, impl: Callable, case: Tuple[str, dt, List[str]]):
    name, search_dt, expected = case
    actual = impl(fname, search_dt)
    if actual != expected:
        pdb.set_trace()
        actual = impl(fname, search_dt)
    assert actual == expected


EXISTING_CASES = [
    (
        "shift_schedule1.csv",
        dt(2019, 5, 12, 11, 39),
        [
            "Patrice Perkinson", "Keshia Brutus", "Alesha Goggans",
            "Christiane Mcferron", "Yasmin Najera", "Glennis Gaunce",
            "Nam Gallego", "Cindi Dugas", "Anya Towe", "Mignon Ritzer",
            "Angie Mannion", "Lupita Bolden", "Carla Hicks", "Mauro Freeze",
            "Broderick Mandel", "Rex Kirsh", "Pierre Pedrosa",
            "Carmon Hazen", "Brigida O'Leary", "Sylvia Delozier"
        ]
    ),
    (
        "shift_schedule2.csv",
        dt(2019, 5, 12, 11, 39),
        [
            "Patrice Perkinson", "Keshia Brutus", "Denny Twedt",
            "Alesha Goggans", "Jarrod Zacarias", "Elba Hawks",
            "Christiane Mcferron", "Yasmin Najera", "Glennis Gaunce",
            "Penni Pflum", "Nam Gallego", "Cindi Dugas", "Anya Towe",
            "Mignon Ritzer", "Angie Mannion", "Lupita Bolden",
            "Carla Hicks", "Mauro Freeze", "Broderick Mandel", "Rex Kirsh",
            "Pierre Pedrosa", "Carmon Hazen", "Natisha Cuesta",
            "Brigida O'Leary", "Sylvia Delozier", "Cynthia Smith"
        ]
    )
]

@pytest.fixture(params=EXISTING_CASES)
def existing_case(request) -> Tuple[str, dt, List[str]]:
    return request.param

def test_existing_cases(impl, existing_case):
    fname, search_dt, expected = existing_case
    actual = impl(fname, search_dt)
    if actual != expected:
        pdb.set_trace()
        actual = impl(fname, search_dt)
    assert actual == expected


if __name__ == "__main__":
    print("Checking types")
    result = mypy.api.run([fname for fname in IMPL_FNAMES])
    if result[0]:
        print("\nType checking report:\n")
        print(result[0])  # stdout

    if result[1]:
        print("\nType error report:\n")
        print(result[1])  # stderr
    print("Checking unittests")
    pytest.main()