Compare revisions

- Accéder à d'autre infos telle que l'API des cours sur demande à la DISI pour intégrer ça aux prédictions (ex: cours en promo complète juste implique plus d'attente)

## Coté dev

- Ajouter une table pour les fermetures exceptionnelles

- Améliorer l'interface utilisateur avec le timer

- Protéger l'interface OpenAPI et mettre en place une interface admin pour les news et potentiellement modération (avec authentification)

- Permettre de définir les masques proprement et de manière à pouvoir généraliser à d'autre RU

def get_waiting_time(place: str, db: Session):

    """ Get the last estimated waiting time for the given place """

    current_date = datetime.now(tz=pytz.timezone("Europe/Paris"))

    weekday, current_time = current_date.weekday(), current_date.time()

    date, weekday, current_time = current_date.date(), current_date.weekday(), current_date.time()

    opening_hours = db.query(

        models.OpeningHours.open_time,

        models.OpeningHours.close_time).filter(

        models.OpeningHours.day == weekday).order_by(

        models.OpeningHours.open_time).all()

    for time_slot in opening_hours:

        closure = db.query(

            models.Closure).filter(

            models.Closure.place == place,

            models.Closure.beginning_date <= datetime.combine(date, time_slot.open_time),

            models.Closure.end_date >= datetime.combine(date, time_slot.open_time)).order_by(

            models.Closure.beginning_date).first()

        if not closure:

            if current_time < time_slot.open_time:

                return schemas.WaitingTime(next_timetable=time_slot.open_time.strftime('%Hh%M'))

            elif current_time <= time_slot.close_time:

            limit = datetime.combine(date.today(), time_slot.open_time)

            last_record = db.query(models.Records.waiting_time).filter(models.Records.place == place).filter(

                models.Records.date >= limit).order_by(models.Records.date.desc()).first()

                limit = datetime.combine(date, time_slot.open_time)

                last_record = db.query(

                    models.Records.waiting_time).filter(

                    models.Records.place == place).filter(

                    models.Records.date >= limit).order_by(

                    models.Records.date.desc()).first()

                waiting_time = None

                if last_record:

                    waiting_time = round(

def add_slot(slots_list, start_time, end_time, function):

    average_waiting_time = function(start_time, end_time)

    if average_waiting_time:

    waiting_time = function(start_time, end_time)

    if waiting_time:

        name = 60 * start_time.hour + start_time.minute

        slots_list.append(schemas.RecordRead(

            name=name, time=average_waiting_time))

        slots_list.append(schemas.RecordRead(name=name, time=waiting_time))

def get_avg_graph_points(place: str, weekday: int, min_time: time,

    for the current or next available timeslot"""

    current_date = datetime.now(tz=pytz.timezone("Europe/Paris"))

    weekday, current_time = current_date.weekday(), current_date.time()

    opening_hours = db.query(models.OpeningHours.open_time, models.OpeningHours.close_time).filter(

        models.OpeningHours.place == place, models.OpeningHours.day == weekday).order_by(models.OpeningHours.open_time).all()

    opening_hours = db.query(

        models.OpeningHours.open_time,

        models.OpeningHours.close_time).filter(

        models.OpeningHours.place == place,

        models.OpeningHours.day == weekday).order_by(

        models.OpeningHours.open_time).all()

    closure = db.query(

        models.Closure).filter(

        models.Closure.place == place,

        models.Closure.beginning_date <= current_date,

        models.Closure.end_date >= current_date).first()

    if not closure:

        for time_slot in opening_hours:

            if current_time <= time_slot.close_time:

                return get_avg_graph_points(place, weekday, time_slot.open_time, time_slot.close_time, timedelta(minutes=5), db)

def get_current_graph(place: str, db: Session):

    """ Get the waiting_time_graph for the current timeslot"""

    current_date = datetime.now(tz=pytz.timezone("Europe/Paris"))

    weekday, day, current_time = current_date.weekday(

    ), current_date.date(), current_date.time()

    opening_hours = db.query(models.OpeningHours.open_time, models.OpeningHours.close_time).filter(

        models.OpeningHours.place == place, models.OpeningHours.day == weekday).all()

    weekday, day, current_time = current_date.weekday(), current_date.date(), current_date.time()

    opening_hours = db.query(

        models.OpeningHours.open_time,

        models.OpeningHours.close_time).filter(

        models.OpeningHours.place == place,

        models.OpeningHours.day == weekday).all()

    closure = db.query(

        models.Closure).filter(

        models.Closure.place == place,

        models.Closure.beginning_date <= current_date,

        models.Closure.end_date >= current_date).first()

    if not closure:

        for time_slot in opening_hours:

            if time_slot.open_time <= current_time <= time_slot.close_time:

                points = get_current_graph_points(

            models.Comments.published_at.desc(),

            models.Comments.id.desc()).all()

    else:

        comments = db.query(models.Comments, models.Users.username).join(models.Users).filter(models.Comments.place == place).order_by(

            models.Comments.published_at.desc(), models.Comments.id.desc()).slice((page - 1) * 20, page * 20).all()

    comments_list = list(schemas.Comment(

        **comment.__dict__, username=username) for comment, username in comments)

        comments = db.query(

            models.Comments,

            models.Users.username).join(

            models.Users).filter(

            models.Comments.place == place).order_by(

            models.Comments.published_at.desc(),

            models.Comments.id.desc()).slice(

            (page - 1) * 20, page * 20).all()

    comments_list = [schemas.Comment(**comment.__dict__, username=username) for comment, username in comments]

    comments_list.reverse()

    return comments_list

def create_comment(user: schemas.User, place: str, new_comments: schemas.CommentBase, db: Session):

    """ Add a new comment to the database """

    date = datetime.now(tz=pytz.timezone("Europe/Paris"))

    db_comment = models.Comments(

        **new_comments.dict(), published_at=date, place=place, user_id=user.id)

    db_comment = models.Comments(**new_comments.dict(), published_at=date, place=place, user_id=user.id)

    db.add(db_comment)

    db.commit()

    db.refresh(db_comment)

def get_news(place: str, db: Session):

    """ Get the news for the given place """

    current_date = datetime.now(tz=pytz.timezone("Europe/Paris"))

    news = db.query(

        models.News).filter(

        models.News.place == place).order_by(

        models.News.place == place,

        models.News.end_date >= current_date).order_by(

        models.News.published_at.desc()).all()

    opening_hours = db.query(

        models.OpeningHours.open_time,

        models.OpeningHours.close_time).filter(

        models.OpeningHours.place == place,

        models.OpeningHours.day == current_date.weekday()).order_by(

        models.OpeningHours.open_time).all()

    next_timetable = None

    for time_slot in opening_hours:

        if current_date.time() < time_slot.open_time:

            next_timetable=time_slot.open_time

            break

    if not next_timetable:

        closure = db.query(

            models.Closure).filter(

            models.Closure.place == place,

            models.Closure.beginning_date <= current_date,

            models.Closure.end_date >= current_date).first()

    else:

        closure = db.query(

            models.Closure).filter(

            models.Closure.place == place,

            models.Closure.beginning_date <= datetime.combine(current_date.date(), next_timetable),

            models.Closure.end_date >= datetime.combine(current_date.date(), next_timetable)).first()

    if closure:

        closure_news = schemas.News(

            title="Fermeture exceptionnelle",

            content=f"{place} est exceptionnellement hors service jusqu'au {closure.end_date.strftime('%d/%m/%y à %Hh%M')}",

            end_date=closure.end_date,

            place=place,

            published_at=closure.beginning_date)

        news.append(closure_news)

    return news

            models.CollaborativeRecords.id == id).delete()

    db.commit()

    return

# Define CRUD operation for exceptional closure

def get_closure(place: str, db: Session):

    current_date = datetime.now(tz=pytz.timezone("Europe/Paris"))

    closures = db.query(

        models.Closure).filter(

        models.Closure.place == place,

        models.Closure.end_date >= current_date).order_by(

        models.Closure.beginning_date).all()

    return [schemas.Closure(**closure.__dict__) for closure in closures]

def create_closure(closure: schemas.Closure, db: Session):

    db_closure = models.Closure(**closure.dict())

    db.add(db_closure)

    db.commit()

    db.refresh(db_closure)

    return schemas.Closure(**closure.dict())

def delete_closure(id: int, db: Session):

    if id == 0:

        db.query(models.Closure).delete()

    else:

        db.query(models.Closure).filter(models.Closure.id == id).delete()

    db.commit()

    return

    close_time = Column(Time)

class Closure(Base):

    """ Register exceptional closure for a period sql table model """

    __tablename__ = "closure"

    id = Column(Integer, primary_key=True, index=True)

    place = Column(String(30))

    beginning_date = Column(DateTime)

    end_date = Column(DateTime)

class Users(Base):

    """ User sql table model """

    __tablename__ = "users"

        orm_mode = True

class ClosureBase(BaseModel):

    """ Closure schema base """

    place: str = Field(..., title="Name of the restaurant")

    beginning_date: datetime = Field(..., title="Beginning date of closure")

    end_date: datetime = Field(..., title="Ending date of closure")

class Closure(ClosureBase):

    """ Closure schema """

    id: int

    class Config:

        orm_mode = True

class Restaurant(BaseModel):

    """Restaurant schema for reading"""

    name: str

# Integration of routers

app.include_router(infos.router)

app.include_router(records.router)

app.include_router(stats.router)

app.include_router(comments.router)

app.include_router(news.router)

app.include_router(comments.router)

app.include_router(authentication.router)

app.include_router(websocket.router)

app.include_router(records.router)

from . import stats

from . import websocket

from . import records

from . import infos

from fastapi import APIRouter, Depends

from sqlalchemy.orm import Session

from typing import List

from db import schemas, crud

from db.database import get_db

router = APIRouter(prefix="/api", tags=["timetable"])

# Manage opening hours

@router.get('/{place}/opening_hours', response_model=List[schemas.OpeningHours])

async def get_opening_hours(place: str, db: Session = Depends(get_db)):

    return crud.get_opening_hours(place, db)

@router.post('/opening_hours', response_model=schemas.OpeningHours)

async def create_opening_hours(opening_hours: schemas.OpeningHoursBase, db: Session = Depends(get_db)):

    return crud.create_opening_hours(opening_hours, db)

@router.delete('/opening_hours/{id}', response_model=None)

async def delete_opening_hours(id: int, db: Session = Depends(get_db)):

    return crud.delete_opening_hours(id, db)

# Manage exceptional closure

@router.get('/{place}/closure', response_model=List[schemas.Closure])

async def get_closure(place: str, db: Session = Depends(get_db)):

    return crud.get_closure(place, db)

@router.post('/closure', response_model=schemas.Closure)

async def create_closure(closure: schemas.Closure, db: Session = Depends(get_db)):

    return crud.create_closure(closure, db)

@router.delete('/closure/{id}', response_model=None)

async def delete_closure(id: int, db: Session = Depends(get_db)):

    return crud.delete_closure(id, db)

# Render restaurants infos

@router.get('/restaurants', response_model=List[schemas.Restaurant])

async def get_restaurants(db: Session = Depends(get_db)):

    return crud.get_restaurants(db)

from fastapi import APIRouter, Depends

from sqlalchemy.orm import Session

from typing import List

from db import schemas, crud

from db.database import get_db

router = APIRouter(prefix="/api")

router = APIRouter(prefix="/api", tags=["stats"])

@router.get('/{place}/waiting_time', response_model=schemas.WaitingTime, tags=["stats"])

@router.get('/{place}/waiting_time', response_model=schemas.WaitingTime)

async def waiting_time(place: str, db: Session = Depends(get_db)):

    return crud.get_waiting_time(place, db)

@router.get('/{place}/stats/avg_graph', response_model=list, tags=["stats"])

@router.get('/{place}/stats/avg_graph', response_model=list)

async def stats(place: str, db: Session = Depends(get_db)):

    return crud.get_avg_graph(place, db)

@router.get('/{place}/stats/current_graph', response_model=schemas.Graph, tags=["stats"])

@router.get('/{place}/stats/current_graph', response_model=schemas.Graph)

async def stats(place: str, db: Session = Depends(get_db)):

    return crud.get_current_graph(place, db)

@router.get('/{place}/opening_hours',

            response_model=List[schemas.OpeningHours], tags=["timetable"])

async def get_opening_hours(place: str, db: Session = Depends(get_db)):

    return crud.get_opening_hours(place, db)

@router.post('/opening_hours', response_model=schemas.OpeningHours, tags=["timetable"])

async def create_opening_hours(opening_hours: schemas.OpeningHoursBase, db: Session = Depends(get_db)):

    return crud.create_opening_hours(opening_hours, db)

@router.delete('/opening_hours/{id}', response_model=None, tags=["timetable"])

async def delete_opening_hours(id: int, db: Session = Depends(get_db)):

    return crud.delete_opening_hours(id, db)

@router.get('/restaurants', response_model=List[schemas.Restaurant], tags=["timetable"])

async def get_restaurants(db: Session = Depends(get_db)):

    return crud.get_restaurants(db)