asreva
dc59b8affb
- New filesystem.py: make_fs() factory (returns s3fs.S3FileSystem or None),
plus fsjoin/fsstem/fsname path helpers
- config.py: storage field is now required ('local' or 's3'); load_config
raises a clear ValueError when it is missing
- video_loader, clip_selector, annotator: thread fs through all file I/O;
local paths unchanged, S3 paths use fs.open/fs.exists/fs.pipe
- annotation_script: load .env via python-dotenv at startup, create fs from
config and pass to Annotator
- Add .env.example with SwitchEngines endpoint and AWS checksum env vars
- pyproject.toml: add s3fs and python-dotenv dependencies
- Reduce default mask alpha from 40% to 15%
- Update example clip names to colon-separated timestamps
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
83 lines
2.4 KiB
Python
83 lines
2.4 KiB
Python
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
|
|
@dataclass
|
|
class FilenameConfig:
|
|
video_in_zip: str = "left.mp4"
|
|
video_tmp_suffix: str = ".mp4"
|
|
zip_extension: str = ".zip"
|
|
mask: str = "mask.png"
|
|
metadata: str = "metadata.json"
|
|
frame: str = "frame.png"
|
|
overlay: str = "overlay.png"
|
|
mask_vis: str = "mask_vis.png"
|
|
gif_original_hires: str = "video_original_hires.gif"
|
|
gif_original_lowres: str = "video_original_lowres.gif"
|
|
gif_overlay_hires: str = "video_overlay_hires.gif"
|
|
gif_overlay_lowres: str = "video_overlay_lowres.gif"
|
|
|
|
|
|
@dataclass
|
|
class AppConfig:
|
|
storage: str # required: 'local' or 's3'
|
|
display_max: int = 480
|
|
fps_fallback: int = 25
|
|
max_frames: int = 100
|
|
data_dir: str = "data/clips"
|
|
out_dir: str = "data/annotation_results"
|
|
clips_file: str = "config/clips.txt"
|
|
optical_flow_config_file: str = ""
|
|
questions: list = field(default_factory=list)
|
|
filenames: FilenameConfig = field(default_factory=FilenameConfig)
|
|
|
|
def get_questions(self):
|
|
return [
|
|
(
|
|
s["section"],
|
|
[
|
|
(
|
|
item["key"],
|
|
item["label"],
|
|
[str(o) for o in item["options"]],
|
|
str(item["default"])
|
|
if item.get("default") is not None
|
|
else None,
|
|
)
|
|
for item in s["items"]
|
|
],
|
|
)
|
|
for s in self.questions
|
|
]
|
|
|
|
|
|
@dataclass
|
|
class OpticalFlowConfig:
|
|
enabled: bool = False
|
|
norm_squared_threshold: float = 0.3
|
|
gaussian_kernel: tuple[int, int] = (5, 5)
|
|
brightness_range: tuple[int, int] = (20, 235)
|
|
|
|
|
|
def load_optical_flow_config(path: Path) -> OpticalFlowConfig:
|
|
with open(path) as f:
|
|
data = yaml.safe_load(f)
|
|
data["gaussian_kernel"] = tuple(data["gaussian_kernel"])
|
|
data["brightness_range"] = tuple(data["brightness_range"])
|
|
return OpticalFlowConfig(**data)
|
|
|
|
|
|
def load_config(path: Path) -> AppConfig:
|
|
with open(path) as f:
|
|
data = yaml.safe_load(f)
|
|
if "storage" not in data:
|
|
raise ValueError(
|
|
f"{path}: missing required field 'storage'. Set it to 'local' or 's3'."
|
|
)
|
|
fn_data = data.pop("filenames", {})
|
|
cfg = AppConfig(**data)
|
|
cfg.filenames = FilenameConfig(**fn_data)
|
|
return cfg
|