refactor: type hints

main
Khiem Ton 2 years ago
parent 901a0e4eec
commit 239da45adf
Signed by: th4tkh13m
GPG Key ID: 4D9CF147DCADD05D

@ -15,7 +15,7 @@ class Database:
self.data.mkdir(parents=True, exist_ok=True)
self.engine = create_engine(f"sqlite:///{(self.data / 'data.db').absolute()}")
def initialize_database(self):
def initialize_database(self) -> bool:
logger.info("Creating databases.")
try:
Pokemon.metadata.create_all(self.engine)
@ -23,6 +23,8 @@ class Database:
PlayerHistory.metadata.create_all(self.engine)
except ArgumentError:
logger.info("Databases existed.")
finally:
return True
def get_engine(self) -> Engine:
return self.engine

@ -3,6 +3,7 @@ from sqlalchemy.orm import declarative_base
from sqlalchemy import Column, DateTime, PrimaryKeyConstraint
from sqlalchemy.orm import Session
from datetime import datetime
from typing import Tuple, Optional
Base = declarative_base()
@ -17,7 +18,7 @@ class PlayerHistory(Base):
def __init__(self, engine=None):
self.engine = engine
def get_last_activity(self):
def get_last_activity(self) -> Optional[Tuple[datetime, Location]]:
time, location = None, None
with Session(self.engine) as session:
result = session.query(PlayerHistory).order_by(PlayerHistory.time.desc()).first()
@ -28,11 +29,10 @@ class PlayerHistory(Base):
return time, location
def add_activity(self, location: Location):
def add_activity(self, location: Location) -> None:
with Session(self.engine) as session:
self.time = datetime.now()
self.location = location
session.add(self)
session.commit()
session.close()
return True

@ -16,7 +16,7 @@ class PokemonList:
self.engine = engine
self.timedelta = timedelta(minutes=2)
def sort_by_name(self, reverse=False):
def sort_by_name(self, reverse: bool = False) -> List[Pokemon]:
with Session(self.engine) as session:
if reverse:
result = session.query(Pokemon).where(Pokemon.visited.is_(False), Pokemon.end_time > datetime.now() + self.timedelta)\
@ -28,7 +28,7 @@ class PokemonList:
session.close()
return result.all()
def sort_by_cp(self, reverse=False):
def sort_by_cp(self, reverse: bool = False) -> List[Pokemon]:
with Session(self.engine) as session:
if reverse:
result = session.query(Pokemon).where(Pokemon.visited.is_(False), Pokemon.end_time > datetime.now() + self.timedelta) \
@ -40,7 +40,7 @@ class PokemonList:
session.close()
return result.all()
def sort_by_level(self, reverse=False):
def sort_by_level(self, reverse: bool = False) -> List[Pokemon]:
with Session(self.engine) as session:
if reverse:
result = session.query(Pokemon).where(Pokemon.visited.is_(False), Pokemon.end_time > datetime.now() + self.timedelta) \
@ -52,28 +52,28 @@ class PokemonList:
session.close()
return result.all()
def sort_by_distance(self, location: Location, reverse=False):
def sort_by_distance(self, location: Location, reverse: bool = False) -> List[Pokemon]:
with Session(self.engine) as session:
result = session.query(Pokemon).where(Pokemon.visited.is_(False), Pokemon.end_time > datetime.now() + self.timedelta)
session.commit()
session.close()
return sorted(result.all(), key=lambda pokemon: location.distance(pokemon.location), reverse=reverse)
def search_by_name(self, query: str):
def search_by_name(self, query: str) -> List[Pokemon]:
with Session(self.engine) as session:
result = session.query(Pokemon).where(Pokemon.visited.is_(False), Pokemon.end_time > datetime.now() + self.timedelta)
session.commit()
session.close()
return list(filter(lambda pokemon: partial_ratio(query.lower(), pokemon.name) >= 80, result.all()))
def insert_to_database(self, pokemons: List[Pokemon]):
def insert_to_database(self, pokemons: List[Pokemon]) -> None:
logger.info("Add pokemons to database.")
with Session(self.engine) as session:
session.add_all(pokemons)
session.commit()
session.close()
def visit_pokemon(self, pokemon: Pokemon):
def visit_pokemon(self, pokemon: Pokemon) -> None:
with Session(self.engine) as session:
new_pokemon = session.query(Pokemon).where(Pokemon.name == pokemon.name,
Pokemon.location == pokemon.location,
@ -83,3 +83,4 @@ class PokemonList:
session.commit()
session.close()

@ -16,7 +16,7 @@ class RaidList:
self.engine = engine
self.timedelta = timedelta(minutes=2)
def sort_by_name(self, reverse=False):
def sort_by_name(self, reverse: bool = False) -> List[Raid]:
with Session(self.engine) as session:
if reverse:
result = session.query(Raid).where(Raid.end_time > datetime.now() + self.timedelta) \
@ -28,7 +28,7 @@ class RaidList:
session.close()
return result.all()
def sort_by_level(self, reverse=False):
def sort_by_level(self, reverse: bool = False) -> List[Raid]:
with Session(self.engine) as session:
if reverse:
result = session.query(Raid).where(Raid.end_time > datetime.now() + self.timedelta) \
@ -40,21 +40,21 @@ class RaidList:
session.close()
return result.all()
def sort_by_distance(self, location: Location, reverse=False):
def sort_by_distance(self, location: Location, reverse: bool = False) -> List[Raid]:
with Session(self.engine) as session:
result = session.query(Raid).where(Raid.end_time > datetime.now() + self.timedelta)
session.commit()
session.close()
return sorted(result.all(), key=lambda raid: location.distance(raid.location), reverse=reverse)
def search_by_name(self, query: str):
def search_by_name(self, query: str) -> List[Raid]:
with Session(self.engine) as session:
result = session.query(Raid).where(Raid.end_time > datetime.now() + self.timedelta)
session.commit()
session.close()
return list(filter(lambda raid: partial_ratio(query.lower(), raid.name) >= 80, result.all()))
def insert_to_database(self, raids: List[Raid]):
def insert_to_database(self, raids: List[Raid]) -> None:
logger.info("Insert RAIDS to Database.")
with Session(self.engine) as session:
session.add_all(raids)

@ -3,6 +3,8 @@ from ispoof.spoofer.location import Location
from ispoof.spoofer.device import Device
from ispoof.lists.playerhistory import PlayerHistory
from datetime import datetime, timedelta
from typing import Tuple, Optional
from pathlib import Path
from math import ceil
import gpxpy
import time
@ -19,65 +21,65 @@ class Player:
self.history = PlayerHistory(engine=engine)
self.device = Device()
def prepare_device(self):
def prepare_device(self) -> None:
self.device.mount_image()
def set_location_with_query(self, query):
def set_location_with_query(self, query: str) -> None:
loc = Nominatim(user_agent="GetLoc")
get_loc = loc.geocode(query)
self.location = Location(get_loc.latitude, get_loc.longitude)
self.device.spoof_gps(self.location)
def get_last_activity(self):
def get_last_activity(self) -> Optional[Tuple[datetime, Location]]:
return self.history.get_last_activity()
def set_location(self, location: Location):
def set_location(self, location: Location) -> None:
self.location = location
self.device.spoof_gps(self.location)
def do_activity(self, location: Location):
return self.history.add_activity(location)
def do_activity(self, location: Location) -> None:
self.history.add_activity(location)
def get_location(self):
def get_location(self) -> Location:
return self.location
def distance_to(self, other):
def distance_to(self, other: Location) -> float:
return self.location.distance(other)
def get_cooldown(self, location):
time, history_location = self.get_last_activity()
def get_cooldown(self, location: Location) -> int:
_time, history_location = self.get_last_activity()
if time and history_location:
cooldown = history_location.get_cooldown(location)
current_cooldown = ceil((time + timedelta(minutes=cooldown) - datetime.now()).total_seconds() / 60)
current_cooldown = ceil((_time + timedelta(minutes=cooldown) - datetime.now()).total_seconds() / 60)
if current_cooldown < 0:
current_cooldown = 0
else:
current_cooldown = 0
return current_cooldown
def get_current_cooldown(self):
def get_current_cooldown(self) -> int:
self.cooldown = self.get_cooldown(self.location)
return self.cooldown
def change_gps_by_query(self, query: str, do_activity=False):
def change_gps_by_query(self, query: str, do_activity: bool = False) -> None:
self.set_location_with_query(query)
if do_activity:
self.do_activity(self.location)
def change_gps_by_location(self, location: Location, do_activity=False):
def change_gps_by_location(self, location: Location, do_activity: bool = False) -> None:
self.set_location(location)
if do_activity:
self.do_activity(self.location)
def set_last_location(self):
def set_last_location(self) -> None:
_, location = self.get_last_activity()
self.set_location(location)
def set_speed(self, speed):
def set_speed(self, speed: float) -> None:
self.speed = speed
def gpx_walking(self, gpx_file):
def gpx_walking(self, gpx_file: Path) -> None:
with open(gpx_file) as f:
gpx = gpxpy.parse(f)
gpx_points = []
@ -103,3 +105,4 @@ class Player:

@ -42,12 +42,12 @@ class Pokemon(Base):
self.end_time = end_time
self.country = country
def is_despawned(self):
def is_despawned(self) -> bool:
if datetime().now() > self.end_time:
return True
return False
def __repr__(self):
def __repr__(self) -> str:
return f"{self.name}: {self.cp}, {self.level}, {self.attack}-{self.defense}-{self.hp}, shinable: {self.shiny}, end: {self.end_time}"

@ -26,11 +26,11 @@ class Raid(Base):
self.end_time = end_time
self.country = country
def is_dispawned(self):
def is_dispawned(self) -> bool:
if datetime().now() > self.end_time:
return True
return False
def __repr__(self):
def __repr__(self) -> str:
return f"(Raid: {self.name} - {self.level}-star - {self.location} - Start: {self.start_time} - End: {self.end_time})"
Loading…
Cancel
Save