Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| de3a30ef34 | |||
| 7d419b375b |
+4
-189
@@ -1,197 +1,12 @@
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
__version__ = "1.0.11"
|
__version__ = "1.0.13"
|
||||||
|
|
||||||
import dataclasses
|
|
||||||
import hashlib
|
|
||||||
import json
|
|
||||||
import time
|
import time
|
||||||
from datetime import datetime, date
|
from typing import Callable
|
||||||
from pathlib import Path
|
|
||||||
from typing import Union, Callable
|
|
||||||
from types import SimpleNamespace
|
|
||||||
|
|
||||||
|
from .color_utils import *
|
||||||
def ansi_rgb(r: int, g: int, b: int, foreground: bool = True) -> str:
|
from .serializer import *
|
||||||
"""
|
|
||||||
Convert rgb color into ANSI escape code format
|
|
||||||
|
|
||||||
:param r:
|
|
||||||
:param g:
|
|
||||||
:param b:
|
|
||||||
:param foreground: Whether the color applies to forground
|
|
||||||
:return: Escape code
|
|
||||||
"""
|
|
||||||
c = '38' if foreground else '48'
|
|
||||||
return f'\033[{c};2;{r};{g};{b}m'
|
|
||||||
|
|
||||||
|
|
||||||
replacements = ["&0/\033[0;30m", "&1/\033[0;34m", "&2/\033[0;32m", "&3/\033[0;36m", "&4/\033[0;31m",
|
|
||||||
"&5/\033[0;35m", "&6/\033[0;33m", "&7/\033[0;37m", "&8/\033[1;30m", "&9/\033[1;34m",
|
|
||||||
"&a/\033[1;32m", "&b/\033[1;36m", "&c/\033[1;31m", "&d/\033[1;35m", "&e/\033[1;33m",
|
|
||||||
"&f/\033[1;37m",
|
|
||||||
"&r/\033[0m", "&l/\033[1m", "&o/\033[3m", "&n/\033[4m", "&-/\n"]
|
|
||||||
replacements = [(r[:2], r[3:]) for r in replacements]
|
|
||||||
|
|
||||||
|
|
||||||
def color(msg: str) -> str:
|
|
||||||
"""
|
|
||||||
Replace extended minecraft color codes in string
|
|
||||||
|
|
||||||
:param msg: Message with minecraft color codes
|
|
||||||
:return: Message with escape codes
|
|
||||||
"""
|
|
||||||
for code, esc in replacements:
|
|
||||||
msg = msg.replace(code, esc)
|
|
||||||
|
|
||||||
while '&gf(' in msg or '&gb(' in msg:
|
|
||||||
i = msg.index('&gf(') if '&gf(' in msg else msg.index('&gb(')
|
|
||||||
end = msg.index(')', i)
|
|
||||||
code = msg[i + 4:end]
|
|
||||||
fore = msg[i + 2] == 'f'
|
|
||||||
|
|
||||||
if code.startswith('#'):
|
|
||||||
rgb = tuple(int(code.lstrip('#')[i:i+2], 16) for i in (0, 2, 4))
|
|
||||||
else:
|
|
||||||
code = code.replace(',', ' ').replace(';', ' ').replace(' ', ' ')
|
|
||||||
rgb = tuple(int(c) for c in code.split(' '))
|
|
||||||
|
|
||||||
msg = msg[:i] + ansi_rgb(*rgb, foreground=fore) + msg[end + 1:]
|
|
||||||
|
|
||||||
return msg
|
|
||||||
|
|
||||||
|
|
||||||
def printc(msg: str):
|
|
||||||
"""
|
|
||||||
Print with color
|
|
||||||
|
|
||||||
:param msg: Message with minecraft color codes
|
|
||||||
"""
|
|
||||||
print(color(msg + '&r'))
|
|
||||||
|
|
||||||
|
|
||||||
def parse_date_time(iso: str) -> datetime:
|
|
||||||
"""
|
|
||||||
Parse date faster. Running 1,000,000 trials, this parse_date function is 4.03 times faster than
|
|
||||||
python's built-in dateutil.parser.isoparse() function.
|
|
||||||
|
|
||||||
Preconditions:
|
|
||||||
- iso is the output of datetime.isoformat() (In a format like "2021-10-20T23:50:14")
|
|
||||||
- iso is a valid date (this function does not check for the validity of the input)
|
|
||||||
|
|
||||||
:param iso: Input date
|
|
||||||
:return: Datetime object
|
|
||||||
"""
|
|
||||||
return datetime(int(iso[:4]), int(iso[5:7]), int(iso[8:10]),
|
|
||||||
int(iso[11:13]), int(iso[14:16]), int(iso[17:19]))
|
|
||||||
|
|
||||||
|
|
||||||
def parse_date_only(iso: str) -> datetime:
|
|
||||||
"""
|
|
||||||
Parse date faster.
|
|
||||||
|
|
||||||
Preconditions:
|
|
||||||
- iso starts with the format of "YYYY-MM-DD" (e.g. "2021-10-20" or "2021-10-20T10:04:14")
|
|
||||||
- iso is a valid date (this function does not check for the validity of the input)
|
|
||||||
|
|
||||||
:param iso: Input date
|
|
||||||
:return: Datetime object
|
|
||||||
"""
|
|
||||||
return datetime(int(iso[:4]), int(iso[5:7]), int(iso[8:10]))
|
|
||||||
|
|
||||||
|
|
||||||
class EnhancedJSONEncoder(json.JSONEncoder):
|
|
||||||
"""
|
|
||||||
An improvement to the json.JSONEncoder class, which supports:
|
|
||||||
encoding for dataclasses, encoding for datetime, and sets
|
|
||||||
"""
|
|
||||||
|
|
||||||
def default(self, o: object) -> object:
|
|
||||||
|
|
||||||
# Support encoding dataclasses
|
|
||||||
# https://stackoverflow.com/a/51286749/7346633
|
|
||||||
if dataclasses.is_dataclass(o):
|
|
||||||
return dataclasses.asdict(o)
|
|
||||||
|
|
||||||
# Simple namespace
|
|
||||||
if isinstance(o, SimpleNamespace):
|
|
||||||
return o.__dict__
|
|
||||||
|
|
||||||
# Support encoding datetime
|
|
||||||
if isinstance(o, (datetime, date)):
|
|
||||||
return o.isoformat()
|
|
||||||
|
|
||||||
# Support for sets
|
|
||||||
# https://stackoverflow.com/a/8230505/7346633
|
|
||||||
if isinstance(o, set):
|
|
||||||
return list(o)
|
|
||||||
|
|
||||||
return super().default(o)
|
|
||||||
|
|
||||||
|
|
||||||
def json_stringify(obj: object, indent: Union[int, None] = None) -> str:
|
|
||||||
"""
|
|
||||||
Serialize json string with support for dataclasses and datetime and sets and with custom
|
|
||||||
configuration.
|
|
||||||
|
|
||||||
Preconditions:
|
|
||||||
- obj != None
|
|
||||||
|
|
||||||
:param obj: Objects
|
|
||||||
:param indent: Indent size or none
|
|
||||||
:return: Json strings
|
|
||||||
"""
|
|
||||||
return json.dumps(obj, indent=indent, cls=EnhancedJSONEncoder, ensure_ascii=False)
|
|
||||||
|
|
||||||
|
|
||||||
def jsn(s: str):
|
|
||||||
return json.loads(s, object_hook=lambda d: SimpleNamespace(**d))
|
|
||||||
|
|
||||||
|
|
||||||
def write(file: Union[str, Path], text: str) -> None:
|
|
||||||
"""
|
|
||||||
Write text to a file
|
|
||||||
|
|
||||||
Preconditions:
|
|
||||||
- file != ''
|
|
||||||
|
|
||||||
:param file: File path (will be converted to lowercase)
|
|
||||||
:param text: Text
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
file = Path(file)
|
|
||||||
file.parent.mkdir(parents=True, exist_ok=True)
|
|
||||||
|
|
||||||
with file.open('w', encoding='utf-8') as f:
|
|
||||||
f.write(text)
|
|
||||||
|
|
||||||
|
|
||||||
def read(file: Union[str, Path]) -> str:
|
|
||||||
"""
|
|
||||||
Read file content
|
|
||||||
|
|
||||||
Preconditions:
|
|
||||||
- file != ''
|
|
||||||
|
|
||||||
:param file: File path (will be converted to lowercase)
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
return file.read_text('utf-8')
|
|
||||||
|
|
||||||
|
|
||||||
def md5(file: Union[str, Path]) -> str:
|
|
||||||
"""
|
|
||||||
Compute md5 of a file
|
|
||||||
|
|
||||||
:param file: File path
|
|
||||||
:return: md5 string
|
|
||||||
"""
|
|
||||||
file = Path(file)
|
|
||||||
hash_md5 = hashlib.md5()
|
|
||||||
with open(file, "rb") as f:
|
|
||||||
for chunk in iter(lambda: f.read(4096), b""):
|
|
||||||
hash_md5.update(chunk)
|
|
||||||
return hash_md5.hexdigest()
|
|
||||||
|
|
||||||
|
|
||||||
class Timer:
|
class Timer:
|
||||||
|
|||||||
@@ -0,0 +1,58 @@
|
|||||||
|
|
||||||
|
|
||||||
|
def ansi_rgb(r: int, g: int, b: int, foreground: bool = True) -> str:
|
||||||
|
"""
|
||||||
|
Convert rgb color into ANSI escape code format
|
||||||
|
|
||||||
|
:param r:
|
||||||
|
:param g:
|
||||||
|
:param b:
|
||||||
|
:param foreground: Whether the color applies to forground
|
||||||
|
:return: Escape code
|
||||||
|
"""
|
||||||
|
c = '38' if foreground else '48'
|
||||||
|
return f'\033[{c};2;{r};{g};{b}m'
|
||||||
|
|
||||||
|
|
||||||
|
replacements = ["&0/\033[0;30m", "&1/\033[0;34m", "&2/\033[0;32m", "&3/\033[0;36m", "&4/\033[0;31m",
|
||||||
|
"&5/\033[0;35m", "&6/\033[0;33m", "&7/\033[0;37m", "&8/\033[1;30m", "&9/\033[1;34m",
|
||||||
|
"&a/\033[1;32m", "&b/\033[1;36m", "&c/\033[1;31m", "&d/\033[1;35m", "&e/\033[1;33m",
|
||||||
|
"&f/\033[1;37m",
|
||||||
|
"&r/\033[0m", "&l/\033[1m", "&o/\033[3m", "&n/\033[4m", "&-/\n"]
|
||||||
|
replacements = [(r[:2], r[3:]) for r in replacements]
|
||||||
|
|
||||||
|
|
||||||
|
def color(msg: str) -> str:
|
||||||
|
"""
|
||||||
|
Replace extended minecraft color codes in string
|
||||||
|
|
||||||
|
:param msg: Message with minecraft color codes
|
||||||
|
:return: Message with escape codes
|
||||||
|
"""
|
||||||
|
for code, esc in replacements:
|
||||||
|
msg = msg.replace(code, esc)
|
||||||
|
|
||||||
|
while '&gf(' in msg or '&gb(' in msg:
|
||||||
|
i = msg.index('&gf(') if '&gf(' in msg else msg.index('&gb(')
|
||||||
|
end = msg.index(')', i)
|
||||||
|
code = msg[i + 4:end]
|
||||||
|
fore = msg[i + 2] == 'f'
|
||||||
|
|
||||||
|
if code.startswith('#'):
|
||||||
|
rgb = tuple(int(code.lstrip('#')[i:i+2], 16) for i in (0, 2, 4))
|
||||||
|
else:
|
||||||
|
code = code.replace(',', ' ').replace(';', ' ').replace(' ', ' ')
|
||||||
|
rgb = tuple(int(c) for c in code.split(' '))
|
||||||
|
|
||||||
|
msg = msg[:i] + ansi_rgb(*rgb, foreground=fore) + msg[end + 1:]
|
||||||
|
|
||||||
|
return msg
|
||||||
|
|
||||||
|
|
||||||
|
def printc(msg: str):
|
||||||
|
"""
|
||||||
|
Print with color
|
||||||
|
|
||||||
|
:param msg: Message with minecraft color codes
|
||||||
|
"""
|
||||||
|
print(color(msg + '&r'))
|
||||||
+153
-2
@@ -1,10 +1,17 @@
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import dataclasses
|
||||||
|
import datetime
|
||||||
|
import hashlib
|
||||||
import io
|
import io
|
||||||
|
import json
|
||||||
import pickle
|
import pickle
|
||||||
|
from pathlib import Path
|
||||||
|
from types import SimpleNamespace
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
|
||||||
def pickle_encode(obj: any, protocol=None, fix_imports=True) -> bytes:
|
def pickle_encode(obj: Any, protocol=None, fix_imports=True) -> bytes:
|
||||||
"""
|
"""
|
||||||
Encode object to pickle bytes
|
Encode object to pickle bytes
|
||||||
|
|
||||||
@@ -21,9 +28,153 @@ def pickle_encode(obj: any, protocol=None, fix_imports=True) -> bytes:
|
|||||||
return bio.getvalue()
|
return bio.getvalue()
|
||||||
|
|
||||||
|
|
||||||
def pickle_decode(by: bytes) -> any:
|
def pickle_decode(by: bytes) -> Any:
|
||||||
"""
|
"""
|
||||||
Decode pickle bytes to object
|
Decode pickle bytes to object
|
||||||
"""
|
"""
|
||||||
with io.BytesIO(by) as bio:
|
with io.BytesIO(by) as bio:
|
||||||
return pickle.load(bio)
|
return pickle.load(bio)
|
||||||
|
|
||||||
|
|
||||||
|
class EnhancedJSONEncoder(json.JSONEncoder):
|
||||||
|
"""
|
||||||
|
An improvement to the json.JSONEncoder class, which supports:
|
||||||
|
encoding for dataclasses, encoding for datetime, and sets
|
||||||
|
"""
|
||||||
|
|
||||||
|
def default(self, o: object) -> object:
|
||||||
|
|
||||||
|
# Support encoding dataclasses
|
||||||
|
# https://stackoverflow.com/a/51286749/7346633
|
||||||
|
if dataclasses.is_dataclass(o):
|
||||||
|
return dataclasses.asdict(o)
|
||||||
|
|
||||||
|
# Simple namespace
|
||||||
|
if isinstance(o, SimpleNamespace):
|
||||||
|
return o.__dict__
|
||||||
|
|
||||||
|
# Support encoding datetime
|
||||||
|
if isinstance(o, (datetime.datetime, datetime.date)):
|
||||||
|
return o.isoformat()
|
||||||
|
|
||||||
|
# Support for sets
|
||||||
|
# https://stackoverflow.com/a/8230505/7346633
|
||||||
|
if isinstance(o, set):
|
||||||
|
return list(o)
|
||||||
|
|
||||||
|
return super().default(o)
|
||||||
|
|
||||||
|
|
||||||
|
def json_stringify(obj: object, **kwargs) -> str:
|
||||||
|
"""
|
||||||
|
Serialize json string with support for dataclasses and datetime and sets and with custom
|
||||||
|
configuration.
|
||||||
|
|
||||||
|
Preconditions:
|
||||||
|
- obj != None
|
||||||
|
|
||||||
|
:param obj: Objects
|
||||||
|
:return: Json strings
|
||||||
|
"""
|
||||||
|
args = dict(ensure_ascii=False, cls=EnhancedJSONEncoder)
|
||||||
|
args.update(kwargs)
|
||||||
|
return json.dumps(obj, **args)
|
||||||
|
|
||||||
|
|
||||||
|
def jsn(s: str) -> SimpleNamespace:
|
||||||
|
return json.loads(s, object_hook=lambda d: SimpleNamespace(**d))
|
||||||
|
|
||||||
|
|
||||||
|
def ensure_dir(path: Path | str) -> Path:
|
||||||
|
"""
|
||||||
|
Ensure that the directory exists (and create if not)
|
||||||
|
|
||||||
|
:returns The directory
|
||||||
|
"""
|
||||||
|
path = Path(path)
|
||||||
|
path.mkdir(parents=True, exist_ok=True)
|
||||||
|
return path
|
||||||
|
|
||||||
|
|
||||||
|
def ensure_parent(path: Path | str) -> Path:
|
||||||
|
"""
|
||||||
|
Ensure that the parent directory of a path exists (and create if not)
|
||||||
|
|
||||||
|
:return: The directory
|
||||||
|
"""
|
||||||
|
path = Path(path)
|
||||||
|
ensure_dir(path.parent)
|
||||||
|
return path
|
||||||
|
|
||||||
|
|
||||||
|
def write(fp: Path | str, data: bytes | str):
|
||||||
|
"""
|
||||||
|
Make sure the directory exists, and then write data, either in bytes or string.
|
||||||
|
|
||||||
|
Also forces utf-8 encoding for strings.
|
||||||
|
"""
|
||||||
|
fp = ensure_parent(fp)
|
||||||
|
|
||||||
|
if isinstance(data, str):
|
||||||
|
return fp.write_text(data, 'utf-8')
|
||||||
|
if isinstance(data, bytes):
|
||||||
|
return fp.write_bytes(data)
|
||||||
|
|
||||||
|
|
||||||
|
def read(file: Path | str) -> str:
|
||||||
|
"""
|
||||||
|
Read file content, force utf-8
|
||||||
|
|
||||||
|
:param file: File path
|
||||||
|
:return: File content
|
||||||
|
"""
|
||||||
|
return Path(file).read_text('utf-8')
|
||||||
|
|
||||||
|
|
||||||
|
def write_json(fp: Path | str, data: Any):
|
||||||
|
write(fp, json_stringify(data))
|
||||||
|
|
||||||
|
|
||||||
|
def parse_date_time(iso: str) -> datetime.datetime:
|
||||||
|
"""
|
||||||
|
Parse date faster. Running 1,000,000 trials, this parse_date function is 4.03 times faster than
|
||||||
|
python's built-in dateutil.parser.isoparse() function.
|
||||||
|
|
||||||
|
Preconditions:
|
||||||
|
- iso is the output of datetime.isoformat() (In a format like "2021-10-20T23:50:14")
|
||||||
|
- iso is a valid date (this function does not check for the validity of the input)
|
||||||
|
|
||||||
|
:param iso: Input date
|
||||||
|
:return: Datetime object
|
||||||
|
"""
|
||||||
|
return datetime.datetime(int(iso[:4]), int(iso[5:7]), int(iso[8:10]),
|
||||||
|
int(iso[11:13]), int(iso[14:16]), int(iso[17:19]))
|
||||||
|
|
||||||
|
|
||||||
|
def parse_date_only(iso: str) -> datetime.datetime:
|
||||||
|
"""
|
||||||
|
Parse date faster.
|
||||||
|
|
||||||
|
Preconditions:
|
||||||
|
- iso starts with the format of "YYYY-MM-DD" (e.g. "2021-10-20" or "2021-10-20T10:04:14")
|
||||||
|
- iso is a valid date (this function does not check for the validity of the input)
|
||||||
|
|
||||||
|
:param iso: Input date
|
||||||
|
:return: Datetime object
|
||||||
|
"""
|
||||||
|
return datetime.datetime(int(iso[:4]), int(iso[5:7]), int(iso[8:10]))
|
||||||
|
|
||||||
|
|
||||||
|
def md5(file: Path | str) -> str:
|
||||||
|
"""
|
||||||
|
Compute md5 of a file
|
||||||
|
|
||||||
|
:param file: File path
|
||||||
|
:return: md5 string
|
||||||
|
"""
|
||||||
|
file = Path(file)
|
||||||
|
hash_md5 = hashlib.md5()
|
||||||
|
with open(file, "rb") as f:
|
||||||
|
for chunk in iter(lambda: f.read(4096), b""):
|
||||||
|
hash_md5.update(chunk)
|
||||||
|
return hash_md5.hexdigest()
|
||||||
|
|||||||
Reference in New Issue
Block a user