File size: 3,907 Bytes
78aefa8 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 | """Utilitas umum: logging, I/O pickle/json, parsing metadata dari nama file."""
import glob
import json
import logging
import os
import pickle
import re
import sys
import config
def get_logger(name="pipeline"):
"""Logger tunggal yang menulis ke konsol dan ke outputs/logs/pipeline.log."""
os.makedirs(config.LOGS_DIR, exist_ok=True)
logger = logging.getLogger(name)
if logger.handlers:
return logger # sudah dikonfigurasi sebelumnya, hindari duplikasi handler
logger.setLevel(getattr(logging, config.LOG_LEVEL, logging.INFO))
fmt = logging.Formatter(
"%(asctime)s | %(levelname)-7s | %(name)s | %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
stream_handler = logging.StreamHandler(sys.stdout)
stream_handler.setFormatter(fmt)
logger.addHandler(stream_handler)
file_handler = logging.FileHandler(
os.path.join(config.LOGS_DIR, "pipeline.log"), encoding="utf-8"
)
file_handler.setFormatter(fmt)
logger.addHandler(file_handler)
return logger
def ensure_dirs():
"""Membuat seluruh direktori output yang dibutuhkan pipeline jika belum ada."""
for d in (
config.OUTPUT_DIR,
config.INTERIM_DIR,
config.MASKS_DIR,
config.PROCESSED_DIR,
config.FIGURES_DIR,
config.TABLES_DIR,
config.LOGS_DIR,
):
os.makedirs(d, exist_ok=True)
def save_pickle(obj, path):
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "wb") as f:
pickle.dump(obj, f, protocol=pickle.HIGHEST_PROTOCOL)
def load_pickle(path):
with open(path, "rb") as f:
return pickle.load(f)
def save_json(obj, path):
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "w", encoding="utf-8") as f:
json.dump(obj, f, indent=2, ensure_ascii=False)
def list_images(raw_data_dir, extensions):
"""Mencari seluruh file citra secara rekursif di bawah raw_data_dir."""
paths = []
for ext in extensions:
pattern = os.path.join(raw_data_dir, "**", "*" + ext)
paths.extend(glob.glob(pattern, recursive=True))
return sorted(set(paths))
def parse_filename_metadata(filepath, pattern):
"""
Mem-parsing metadata dari nama file sesuai konvensi:
[HARI]_[DOMAIN]_[KODE_DAGING]_[NOMOR_URUT].JPG
Contoh: DAY-1_RPH_HLR_001.JPG
Mengembalikan dict berisi day, domain, cut_code, seq_number, filename.
Jika nama file tidak cocok pola, field yang tidak terparsing diisi None
dan pipeline tetap berjalan (tidak crash), namun dicatat sebagai warning
oleh caller.
"""
filename = os.path.basename(filepath)
match = re.match(pattern, filename)
if match is None:
return {
"filename": filename,
"filepath": filepath,
"day": None,
"domain": None,
"cut_code": None,
"seq_number": None,
"parsed_ok": False,
}
day, domain, cut_code, seq_number = match.groups()
return {
"filename": filename,
"filepath": filepath,
"day": day,
"domain": domain.upper(),
"cut_code": cut_code.upper(),
"seq_number": int(seq_number),
"parsed_ok": True,
}
def load_labels_csv(path):
"""
Memuat label ground truth opsional dari CSV (kolom: filename,label).
Mengembalikan dict {filename: label}, atau {} jika file tidak ditemukan.
"""
if path is None or not os.path.isfile(path):
return {}
import pandas as pd
df = pd.read_csv(path)
df.columns = [c.strip().lower() for c in df.columns]
if "filename" not in df.columns or "label" not in df.columns:
raise ValueError(
"labels.csv harus memiliki kolom 'filename' dan 'label'. "
"Kolom yang ditemukan: {}".format(list(df.columns))
)
return dict(zip(df["filename"], df["label"]))
|