126 lines
3.2 KiB
Python
126 lines
3.2 KiB
Python
import hashlib
|
|
import json
|
|
from collections import Counter
|
|
from dataclasses import fields
|
|
from datetime import timedelta, datetime, date
|
|
from enum import Enum
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
import pytz
|
|
from pypolyline.cutil import decode_polyline
|
|
from pytz import tzinfo
|
|
|
|
from core.domain.map.GeoLocation import GeoLocation
|
|
from core.types.Id import Id
|
|
from core.types.IntId import IntId
|
|
|
|
|
|
def hash(o: str, size: Optional[int] = None) -> str:
|
|
# TODO return this!: hash = hashlib.sha1(o.encode()).hexdigest()
|
|
# if size is not None:
|
|
# hash = hash[-size:]
|
|
return o
|
|
|
|
|
|
def fileHash(path: Path, size: Optional[int] = None) -> str:
|
|
with path.open('r', encoding='utf-8') as f:
|
|
text = f.read()
|
|
hash = hashlib.sha1(text.encode()).hexdigest()
|
|
if size is not None:
|
|
hash = hash[-size:]
|
|
return hash
|
|
|
|
|
|
def chunks(data, size):
|
|
for i in range(0, len(data), size):
|
|
yield data[i:i + size]
|
|
|
|
|
|
def hash8(o: str) -> str:
|
|
return hash(o, size=8)
|
|
|
|
|
|
def encode(o: any):
|
|
if isinstance(o, Id):
|
|
return str(o.value)
|
|
if isinstance(o, IntId):
|
|
return int(o.value)
|
|
if isinstance(o, timedelta):
|
|
return int(o.total_seconds())
|
|
if isinstance(o, datetime):
|
|
return int(o.timestamp())
|
|
if isinstance(o, date):
|
|
return int(datetime.combine(o, datetime.min.time()).timestamp())
|
|
if isinstance(o, Enum):
|
|
return o.value
|
|
if isinstance(o, set):
|
|
return list(o)
|
|
if isinstance(o, list):
|
|
return o
|
|
return o.__dict__
|
|
|
|
|
|
def json_dump(o: any, f):
|
|
json.dump(obj=o, fp=f, default=encode)
|
|
|
|
|
|
def json_dumps(o: any) -> str:
|
|
return json.dumps(obj=o, default=encode)
|
|
|
|
|
|
def json_loads(o: any) -> str:
|
|
return json.loads(o)
|
|
|
|
|
|
def current_datetime_str() -> str:
|
|
return datetime.now().isoformat().replace(':', '_')
|
|
|
|
|
|
def polyline_decode(data: str) -> list[GeoLocation]:
|
|
return [GeoLocation(lon=arr[0], lat=arr[1]) for arr in decode_polyline(data.encode(), 6)]
|
|
|
|
|
|
def datetimeRange(start: datetime, end: datetime, step: timedelta) -> list[datetime]:
|
|
dates = []
|
|
currentDay = start
|
|
while currentDay <= end:
|
|
dates.append(currentDay)
|
|
currentDay += step
|
|
|
|
return dates
|
|
|
|
|
|
def saveDivision(a: float, b: float, default: float = None) -> Optional[float]:
|
|
return a / b if b else default
|
|
|
|
|
|
def percentage(a: list | int, b: list | int) -> Optional[float]:
|
|
aNum = len(a) if isinstance(a, list) else a
|
|
bNum = len(b) if isinstance(b, list) else b
|
|
if bNum == 0:
|
|
return None
|
|
return round(aNum / bNum * 100, 2)
|
|
|
|
|
|
def fromLocalToUtc(dt: datetime, localTimezone: tzinfo) -> datetime:
|
|
utc = localTimezone.localize(dt).astimezone(pytz.utc)
|
|
return datetime(year=utc.year, month=utc.month, day=utc.day, hour=utc.hour, minute=utc.minute, second=utc.second, microsecond=utc.microsecond)
|
|
|
|
|
|
def dateRange(start: date, end: date) -> set[date]:
|
|
return set([start + timedelta(days=x) for x in range((end - start).days)] + [end])
|
|
|
|
|
|
def initDataclass(cls: any, **kwargs):
|
|
valid_field_names = {field.name for field in fields(cls)}
|
|
return cls(**{k: v for k, v in kwargs.items() if k in valid_field_names})
|
|
|
|
# Find legs of optimization result
|
|
def countDuplicates(arr):
|
|
# Count occurrences of each number
|
|
counts = Counter(arr)
|
|
# Filter numbers that occur more than once
|
|
duplicates = [(num, count) for num, count in counts.items() if count > 1]
|
|
return duplicates
|