Skip to content

feeds

Classes:

Attributes:

GitTag module-attribute 

GitTag = NewType('GitTag', str)

URL module-attribute 

URL = NewType('URL', str)

logger module-attribute 

logger = getLogger(__name__)

ArtCNNFeed

Bases: Feed

Methods:

Attributes:

base_url class-attribute instance-attribute 

base_url = (
    f"https://api.github.com/repos/{repo}/releases?per_page=100&page={page}"
)

display_name class-attribute instance-attribute 

display_name = 'ArtCNN'

headers property 

headers: dict[str, str]

repo class-attribute instance-attribute 

repo = 'Artoriuz/ArtCNN'

all_feeds classmethod 

all_feeds() -> dict[str, type[Feed]]
Source code in vsscale/mlrt/feeds.py 
55
56
57
58
@classproperty.cached
@classmethod
def all_feeds(cls) -> dict[str, type[Feed]]:
    return dict(cls._registry)

releases async 

releases(session: AsyncSession) -> list[Release]
Source code in vsscale/mlrt/feeds.py 
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
async def releases(self, session: niquests.AsyncSession) -> list[Release]:
    raw_releases = list[dict[str, Any]]()

    page = 1
    while True:
        batch = (await session.get(self.base_url.format(page=page), headers=self.headers)).raise_for_status().json()
        if not batch:
            break
        raw_releases.extend(batch)
        page += 1

    result = list[Release]()
    for r in raw_releases:
        tag = GitTag(r["tag_name"])
        published = r.get("published_at", "")
        raw_assets = r.get("assets", [])

        if not raw_assets:
            continue

        onnx_assets = [
            Asset(
                name=a["name"],
                url=URL(a["browser_download_url"]),
                size=a["size"],
                sha256=a["digest"].removeprefix("sha256:"),
            )
            for a in raw_assets
            if a["name"].endswith(".onnx") and a.get("digest")
        ]

        if onnx_assets:
            result.append(Release(tag, published, onnx_assets))

    return result

Asset dataclass 

Asset(name: str, url: URL, size: int, sha256: str)

Attributes:

name instance-attribute 

name: str

sha256 instance-attribute 

sha256: str

size instance-attribute 

size: int

url instance-attribute 

url: URL

DPIRFeed

Bases: Feed

Methods:

Attributes:

base_url class-attribute instance-attribute 

base_url = (
    f"https://api.github.com/repos/{repo}/releases?per_page=100&page={page}"
)

display_name class-attribute instance-attribute 

display_name = 'DPIR'

headers property 

headers: dict[str, str]

repo class-attribute instance-attribute 

repo = 'Jaded-Encoding-Thaumaturgy/KAIR'

all_feeds classmethod 

all_feeds() -> dict[str, type[Feed]]
Source code in vsscale/mlrt/feeds.py 
55
56
57
58
@classproperty.cached
@classmethod
def all_feeds(cls) -> dict[str, type[Feed]]:
    return dict(cls._registry)

releases async 

releases(session: AsyncSession) -> list[Release]
Source code in vsscale/mlrt/feeds.py 
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
async def releases(self, session: niquests.AsyncSession) -> list[Release]:
    raw_releases = list[dict[str, Any]]()

    page = 1
    while True:
        batch = (await session.get(self.base_url.format(page=page), headers=self.headers)).raise_for_status().json()
        if not batch:
            break
        raw_releases.extend(batch)
        page += 1

    result = list[Release]()
    for r in raw_releases:
        tag = GitTag(r["tag_name"])
        published = r.get("published_at", "")
        raw_assets = r.get("assets", [])

        if not raw_assets:
            continue

        onnx_assets = [
            Asset(
                name=a["name"],
                url=URL(a["browser_download_url"]),
                size=a["size"],
                sha256=a["digest"].removeprefix("sha256:"),
            )
            for a in raw_assets
            if a["name"].endswith(".onnx") and a.get("digest")
        ]

        if onnx_assets:
            result.append(Release(tag, published, onnx_assets))

    return result

Feed

Bases: ABC

Base class for model feeds backed by GitHub releases.

Methods:

Attributes:

base_url class-attribute 

base_url: str

display_name class-attribute 

display_name: str

headers property 

headers: dict[str, str]

all_feeds classmethod 

all_feeds() -> dict[str, type[Feed]]
Source code in vsscale/mlrt/feeds.py 
55
56
57
58
@classproperty.cached
@classmethod
def all_feeds(cls) -> dict[str, type[Feed]]:
    return dict(cls._registry)

releases async 

releases(session: AsyncSession) -> list[Release]
Source code in vsscale/mlrt/feeds.py 
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
async def releases(self, session: niquests.AsyncSession) -> list[Release]:
    raw_releases = list[dict[str, Any]]()

    page = 1
    while True:
        batch = (await session.get(self.base_url.format(page=page), headers=self.headers)).raise_for_status().json()
        if not batch:
            break
        raw_releases.extend(batch)
        page += 1

    result = list[Release]()
    for r in raw_releases:
        tag = GitTag(r["tag_name"])
        published = r.get("published_at", "")
        raw_assets = r.get("assets", [])

        if not raw_assets:
            continue

        onnx_assets = [
            Asset(
                name=a["name"],
                url=URL(a["browser_download_url"]),
                size=a["size"],
                sha256=a["digest"].removeprefix("sha256:"),
            )
            for a in raw_assets
            if a["name"].endswith(".onnx") and a.get("digest")
        ]

        if onnx_assets:
            result.append(Release(tag, published, onnx_assets))

    return result

Release dataclass 

Release(tag: GitTag, published_at: str, assets: Sequence[Asset] = list())

Attributes:

assets class-attribute instance-attribute 

assets: Sequence[Asset] = field(default_factory=list)

published_at instance-attribute 

published_at: str

tag instance-attribute 

tag: GitTag

Waifu2XFeed

Bases: Feed

Methods:

Attributes:

base_url class-attribute instance-attribute 

base_url = (
    f"https://api.github.com/repos/{repo}/releases?per_page=100&page={page}"
)

display_name class-attribute instance-attribute 

display_name = 'Waifu2X'

headers property 

headers: dict[str, str]

repo class-attribute instance-attribute 

repo = 'Jaded-Encoding-Thaumaturgy/nunif'

all_feeds classmethod 

all_feeds() -> dict[str, type[Feed]]
Source code in vsscale/mlrt/feeds.py 
55
56
57
58
@classproperty.cached
@classmethod
def all_feeds(cls) -> dict[str, type[Feed]]:
    return dict(cls._registry)

releases async 

releases(session: AsyncSession) -> list[Release]
Source code in vsscale/mlrt/feeds.py 
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
async def releases(self, session: niquests.AsyncSession) -> list[Release]:
    raw_releases = list[dict[str, Any]]()

    page = 1
    while True:
        batch = (await session.get(self.base_url.format(page=page), headers=self.headers)).raise_for_status().json()
        if not batch:
            break
        raw_releases.extend(batch)
        page += 1

    result = list[Release]()
    for r in raw_releases:
        tag = GitTag(r["tag_name"])
        published = r.get("published_at", "")
        raw_assets = r.get("assets", [])

        if not raw_assets:
            continue

        onnx_assets = [
            Asset(
                name=a["name"],
                url=URL(a["browser_download_url"]),
                size=a["size"],
                sha256=a["digest"].removeprefix("sha256:"),
            )
            for a in raw_assets
            if a["name"].endswith(".onnx") and a.get("digest")
        ]

        if onnx_assets:
            result.append(Release(tag, published, onnx_assets))

    return result