From 99f7bd625041eb9dcc2b144398332f086ab41500 Mon Sep 17 00:00:00 2001 From: "Michael Hirsch, Ph.D" Date: Tue, 9 Feb 2021 22:54:14 -0500 Subject: [PATCH] modernize anno --- .github/CODE_OF_CONDUCT.md | 76 -------------------------------------- .github/FUNDING.yml | 4 -- setup.cfg | 2 +- src/pylivestream/api.py | 18 ++++----- src/pylivestream/base.py | 62 +++++++++++++++---------------- src/pylivestream/ffmpeg.py | 11 +++--- src/pylivestream/glob.py | 10 ++--- src/pylivestream/stream.py | 39 ++++++++++--------- src/pylivestream/utils.py | 8 ++-- 9 files changed, 75 insertions(+), 155 deletions(-) delete mode 100644 .github/CODE_OF_CONDUCT.md delete mode 100644 .github/FUNDING.yml diff --git a/.github/CODE_OF_CONDUCT.md b/.github/CODE_OF_CONDUCT.md deleted file mode 100644 index db301c4..0000000 --- a/.github/CODE_OF_CONDUCT.md +++ /dev/null @@ -1,76 +0,0 @@ -# Contributor Covenant Code of Conduct - -## Our Pledge - -In the interest of fostering an open and welcoming environment, we as -contributors and maintainers pledge to making participation in our project and -our community a harassment-free experience for everyone, regardless of age, body -size, disability, ethnicity, sex characteristics, gender identity and expression, -level of experience, education, socio-economic status, nationality, personal -appearance, race, religion, or sexual identity and orientation. - -## Our Standards - -Examples of behavior that contributes to creating a positive environment -include: - -* Using welcoming and inclusive language -* Being respectful of differing viewpoints and experiences -* Gracefully accepting constructive criticism -* Focusing on what is best for the community -* Showing empathy towards other community members - -Examples of unacceptable behavior by participants include: - -* The use of sexualized language or imagery and unwelcome sexual attention or - advances -* Trolling, insulting/derogatory comments, and personal or political attacks -* Public or private harassment -* Publishing others' private information, such as a physical or electronic - address, without explicit permission -* Other conduct which could reasonably be considered inappropriate in a - professional setting - -## Our Responsibilities - -Project maintainers are responsible for clarifying the standards of acceptable -behavior and are expected to take appropriate and fair corrective action in -response to any instances of unacceptable behavior. - -Project maintainers have the right and responsibility to remove, edit, or -reject comments, commits, code, wiki edits, issues, and other contributions -that are not aligned to this Code of Conduct, or to ban temporarily or -permanently any contributor for other behaviors that they deem inappropriate, -threatening, offensive, or harmful. - -## Scope - -This Code of Conduct applies both within project spaces and in public spaces -when an individual is representing the project or its community. Examples of -representing a project or community include using an official project e-mail -address, posting via an official social media account, or acting as an appointed -representative at an online or offline event. Representation of a project may be -further defined and clarified by project maintainers. - -## Enforcement - -Instances of abusive, harassing, or otherwise unacceptable behavior may be -reported by contacting the project team at info@scivision.dev. All -complaints will be reviewed and investigated and will result in a response that -is deemed necessary and appropriate to the circumstances. The project team is -obligated to maintain confidentiality with regard to the reporter of an incident. -Further details of specific enforcement policies may be posted separately. - -Project maintainers who do not follow or enforce the Code of Conduct in good -faith may face temporary or permanent repercussions as determined by other -members of the project's leadership. - -## Attribution - -This Code of Conduct is adapted from the [Contributor Covenant][homepage], version 1.4, -available at https://www.contributor-covenant.org/version/1/4/code-of-conduct.html - -[homepage]: https://www.contributor-covenant.org - -For answers to common questions about this code of conduct, see -https://www.contributor-covenant.org/faq diff --git a/.github/FUNDING.yml b/.github/FUNDING.yml deleted file mode 100644 index 702f5b3..0000000 --- a/.github/FUNDING.yml +++ /dev/null @@ -1,4 +0,0 @@ -# These are supported funding model platforms - -github: [scivision] -ko_fi: scivision diff --git a/setup.cfg b/setup.cfg index 25ab172..7e30163 100644 --- a/setup.cfg +++ b/setup.cfg @@ -46,7 +46,7 @@ lint = flake8-bugbear flake8-builtins flake8-blind-except - mypy + mypy >= 0.800 captions = tinytag diff --git a/src/pylivestream/api.py b/src/pylivestream/api.py index f4f2b54..c663e48 100644 --- a/src/pylivestream/api.py +++ b/src/pylivestream/api.py @@ -9,7 +9,7 @@ pls.microphone('twitch', key='~/twitch.key') """ -import typing as T +from __future__ import annotations from pathlib import Path from .base import FileIn, Microphone, Screenshare, SaveDisk, Webcam @@ -27,14 +27,14 @@ def stream_file( ini_file: Path, - websites: T.Sequence[str], + websites: list[str], video_file: Path, loop: bool = None, assume_yes: bool = False, timeout: float = None, ): S = FileIn(ini_file, websites, infn=video_file, loop=loop, yes=assume_yes, timeout=timeout) - sites: T.List[str] = list(S.streams.keys()) + sites: list[str] = list(S.streams.keys()) # %% Go live if assume_yes: print(f"going live on {sites} looping file {video_file}") @@ -47,7 +47,7 @@ def stream_file( def stream_files( ini_file: Path, - websites: T.Sequence[str], + websites: list[str], *, video_path: Path, glob: str = None, @@ -81,7 +81,7 @@ def stream_files( def stream_microphone( ini_file: Path, - websites: T.Sequence[str], + websites: list[str], *, still_image: Path = None, assume_yes: bool = False, @@ -103,11 +103,11 @@ def stream_microphone( def stream_screen( - ini_file: Path, websites: T.Sequence[str], *, assume_yes: bool = False, timeout: float = None + ini_file: Path, websites: list[str], *, assume_yes: bool = False, timeout: float = None ): S = Screenshare(ini_file, websites, yes=assume_yes, timeout=timeout) - sites: T.List[str] = list(S.streams.keys()) + sites: list[str] = list(S.streams.keys()) # %% Go live if assume_yes: print("going live on", sites) @@ -131,9 +131,9 @@ def capture_screen( s.save() -def stream_webcam(ini_file: Path, websites: T.Sequence[str], *, assume_yes: bool, timeout: float): +def stream_webcam(ini_file: Path, websites: list[str], *, assume_yes: bool, timeout: float): S = Webcam(ini_file, websites, yes=assume_yes, timeout=timeout) - sites: T.List[str] = list(S.streams.keys()) + sites: list[str] = list(S.streams.keys()) # %% Go live if assume_yes: print("going live on", sites) diff --git a/src/pylivestream/base.py b/src/pylivestream/base.py index 94a30f3..49abf23 100644 --- a/src/pylivestream/base.py +++ b/src/pylivestream/base.py @@ -1,5 +1,5 @@ +from __future__ import annotations from pathlib import Path -import typing import logging import os @@ -22,15 +22,15 @@ def __init__(self, inifn: Path, site: str, **kwargs) -> None: self.video_bitrate() - vidIn: typing.List[str] = self.videoIn() - vidOut: typing.List[str] = self.videoOut() + vidIn: list[str] = self.videoIn() + vidOut: list[str] = self.videoOut() - audIn: typing.List[str] = self.audioIn() - audOut: typing.List[str] = self.audioOut() + audIn: list[str] = self.audioIn() + audOut: list[str] = self.audioOut() - buf: typing.List[str] = self.buffer(self.server) + buf: list[str] = self.buffer(self.server) # %% begin to setup command line - cmd: typing.List[str] = [] + cmd: list[str] = [] cmd.append(self.exe) cmd += self.loglevel @@ -61,12 +61,12 @@ def __init__(self, inifn: Path, site: str, **kwargs) -> None: self.sink = sink cmd.append(sink) - self.cmd: typing.List[str] = cmd + self.cmd: list[str] = cmd # %% quick check command, to verify device exists # 0.1 seems OK, spurious buffer error on Windows that wasn't helped by any bigger size CHECKTIMEOUT = "0.1" - self.checkcmd: typing.List[str] = ( + self.checkcmd: list[str] = ( [self.exe] + self.loglevel + ["-t", CHECKTIMEOUT] @@ -76,7 +76,7 @@ def __init__(self, inifn: Path, site: str, **kwargs) -> None: + ["-f", "null", "-"] # webcam needs at output ) - def startlive(self, sinks: typing.Sequence[str] = None): + def startlive(self, sinks: list[str] = None): """finally start the stream(s)""" if self.docheck: @@ -108,9 +108,9 @@ def startlive(self, sinks: typing.Sequence[str] = None): elif len(sinks) == 1: run(self.cmd) else: # multi-stream output tee - cmdstem: typing.List[str] = self.cmd[:-3] + cmdstem: list[str] = self.cmd[:-3] # +global_header is necessary to tee to multiple services - cmd: typing.List[str] = cmdstem + ["-flags:v", "+global_header", "-f", "tee"] + cmd: list[str] = cmdstem + ["-flags:v", "+global_header", "-f", "tee"] if self.image: # connect image to video stream, audio file to audio stream @@ -166,7 +166,7 @@ def check_device(self, site: str = None) -> bool: # %% operators class Screenshare: - def __init__(self, inifn: Path, websites: typing.Sequence[str], **kwargs): + def __init__(self, inifn: Path, websites: list[str], **kwargs): if isinstance(websites, str): websites = [websites] @@ -175,11 +175,11 @@ def __init__(self, inifn: Path, websites: typing.Sequence[str], **kwargs): for site in websites: streams[site] = Livestream(inifn, site, vidsource="screen", **kwargs) - self.streams: typing.Dict[str, Livestream] = streams + self.streams: dict[str, Livestream] = streams def golive(self): - sinks: typing.List[str] = [self.streams[stream].sink for stream in self.streams] + sinks: list[str] = [self.streams[stream].sink for stream in self.streams] try: next(self.streams[unify_streams(self.streams)].startlive(sinks)) @@ -188,7 +188,7 @@ def golive(self): class Webcam: - def __init__(self, inifn: Path, websites: typing.Sequence[str], **kwargs): + def __init__(self, inifn: Path, websites: list[str], **kwargs): if isinstance(websites, str): websites = [websites] @@ -197,11 +197,11 @@ def __init__(self, inifn: Path, websites: typing.Sequence[str], **kwargs): for site in websites: streams[site] = Livestream(inifn, site, vidsource="camera", **kwargs) - self.streams: typing.Dict[str, Livestream] = streams + self.streams: dict[str, Livestream] = streams def golive(self): - sinks: typing.List[str] = [self.streams[stream].sink for stream in self.streams] + sinks: list[str] = [self.streams[stream].sink for stream in self.streams] try: next(self.streams[unify_streams(self.streams)].startlive(sinks)) @@ -210,7 +210,7 @@ def golive(self): class Microphone: - def __init__(self, inifn: Path, websites: typing.Sequence[str], **kwargs): + def __init__(self, inifn: Path, websites: list[str], **kwargs): if isinstance(websites, str): websites = [websites] @@ -219,11 +219,11 @@ def __init__(self, inifn: Path, websites: typing.Sequence[str], **kwargs): for site in websites: streams[site] = Livestream(inifn, site, **kwargs) - self.streams: typing.Dict[str, Livestream] = streams + self.streams: dict[str, Livestream] = streams def golive(self): - sinks: typing.List[str] = [self.streams[stream].sink for stream in self.streams] + sinks: list[str] = [self.streams[stream].sink for stream in self.streams] try: next(self.streams[unify_streams(self.streams)].startlive(sinks)) @@ -233,7 +233,7 @@ def golive(self): # %% File-based inputs class FileIn: - def __init__(self, inifn: Path, websites: typing.Sequence[str], **kwargs): + def __init__(self, inifn: Path, websites: list[str], **kwargs): if isinstance(websites, str): websites = [websites] @@ -242,11 +242,11 @@ def __init__(self, inifn: Path, websites: typing.Sequence[str], **kwargs): for site in websites: streams[site] = Livestream(inifn, site, vidsource="file", **kwargs) - self.streams: typing.Dict[str, Livestream] = streams + self.streams: dict[str, Livestream] = streams def golive(self): - sinks: typing.List[str] = [self.streams[stream].sink for stream in self.streams] + sinks: list[str] = [self.streams[stream].sink for stream in self.streams] try: next(self.streams[unify_streams(self.streams)].startlive(sinks)) @@ -267,13 +267,13 @@ def __init__(self, inifn: Path, outfn: Path = None, **kwargs): self.osparam(kwargs.get("key")) - vidIn: typing.List[str] = self.videoIn() - vidOut: typing.List[str] = self.videoOut() + vidIn: list[str] = self.videoIn() + vidOut: list[str] = self.videoOut() - audIn: typing.List[str] = self.audioIn() - audOut: typing.List[str] = self.audioOut() + audIn: list[str] = self.audioIn() + audOut: list[str] = self.audioOut() - self.cmd: typing.List[str] = [str(self.exe)] + self.cmd: list[str] = [str(self.exe)] self.cmd += vidIn + audIn self.cmd += vidOut + audOut @@ -295,7 +295,7 @@ def save(self): print("specify filename to save screen capture w/ audio to disk.") -def unify_streams(streams: typing.Dict[str, Stream]) -> str: +def unify_streams(streams: dict[str, Stream]) -> str: """ find least common denominator stream settings, so "tee" output can generate multiple streams. @@ -306,7 +306,7 @@ def unify_streams(streams: typing.Dict[str, Stream]) -> str: fast native Python argmin() https://stackoverflow.com/a/11825864 """ - vid_bw: typing.List[int] = [streams[s].video_kbps for s in streams] + vid_bw: list[int] = [streams[s].video_kbps for s in streams] argmin: int = min(range(len(vid_bw)), key=vid_bw.__getitem__) diff --git a/src/pylivestream/ffmpeg.py b/src/pylivestream/ffmpeg.py index b55ecaa..84a5ed7 100644 --- a/src/pylivestream/ffmpeg.py +++ b/src/pylivestream/ffmpeg.py @@ -1,5 +1,6 @@ +from __future__ import annotations +import typing as T import subprocess -from typing import List, Union from time import sleep import os from pathlib import Path @@ -21,7 +22,7 @@ def __init__(self): self.THROTTLE = "-re" - def timelimit(self, t: Union[str, int, float]) -> List[str]: + def timelimit(self, t: str | int | float) -> list[str]: if t is None: return [] @@ -34,7 +35,7 @@ def timelimit(self, t: Union[str, int, float]) -> List[str]: else: return [] - def drawtext(self, text: str = None) -> List[str]: + def drawtext(self, text: str = None) -> list[str]: # fontfile=/path/to/font.ttf: if not text: # None or '' or [] etc. return [] @@ -87,7 +88,7 @@ def listener(self): return proc - def movingBG(self, bgfn: Path = None) -> List[str]: + def movingBG(self, bgfn: Path = None) -> list[str]: if not bgfn: return [] @@ -117,7 +118,7 @@ def get_exe(exein: str) -> str: return exe -def get_meta(fn: Path, exein: str = None) -> Union[None, dict]: +def get_meta(fn: Path, exein: str = None) -> dict[str, T.Any]: if not fn: # audio-only return None diff --git a/src/pylivestream/glob.py b/src/pylivestream/glob.py index a380551..e3f3177 100644 --- a/src/pylivestream/glob.py +++ b/src/pylivestream/glob.py @@ -1,6 +1,6 @@ +from __future__ import annotations import random from pathlib import Path -import typing as T from .base import FileIn from .utils import meta_caption @@ -12,9 +12,9 @@ def playonce( - flist: T.List[Path], + flist: list[Path], image: Path, - sites: T.Sequence[str], + sites: list[str], inifn: Path, shuffle: bool, usemeta: bool, @@ -24,7 +24,7 @@ def playonce( if shuffle: random.shuffle(flist) - caption: T.Union[str, None] + caption: str for f in flist: if usemeta and TinyTag: @@ -41,7 +41,7 @@ def playonce( s.golive() -def fileglob(path: Path, glob: str) -> T.List[Path]: +def fileglob(path: Path, glob: str) -> list[Path]: path = Path(path).expanduser() diff --git a/src/pylivestream/stream.py b/src/pylivestream/stream.py index 58b9800..d9f1614 100644 --- a/src/pylivestream/stream.py +++ b/src/pylivestream/stream.py @@ -4,7 +4,6 @@ import os import sys from configparser import ConfigParser -from typing import List # from . import utils @@ -30,7 +29,7 @@ def __init__(self, inifn: Path, site: str, **kwargs): self.F = Ffmpeg() - self.loglevel: List[str] = self.F.INFO if kwargs.get("verbose") else self.F.ERROR + self.loglevel: list[str] = self.F.INFO if kwargs.get("verbose") else self.F.ERROR self.inifn: Path = Path(inifn).expanduser() if inifn else None @@ -45,13 +44,13 @@ def __init__(self, inifn: Path, site: str, **kwargs): self.loop: bool = kwargs.get("loop") self.infn = Path(kwargs["infn"]).expanduser() if kwargs.get("infn") else None - self.yes: List[str] = self.F.YES if kwargs.get("yes") else [] + self.yes: list[str] = self.F.YES if kwargs.get("yes") else [] - self.queue: List[str] = [] # self.F.QUEUE + self.queue: list[str] = [] # self.F.QUEUE self.caption: str = kwargs.get("caption") - self.timelimit: List[str] = self.F.timelimit(kwargs.get("timeout")) + self.timelimit: list[str] = self.F.timelimit(kwargs.get("timeout")) def osparam(self, key: str): """load OS specific config""" @@ -78,13 +77,13 @@ def osparam(self, key: str): logging.error("Wayland may only give black output. Try X11") if self.vidsource == "camera": - self.res: List[str] = C.get(self.site, "webcam_res").split("x") + self.res: list[str] = C.get(self.site, "webcam_res").split("x") self.fps: float = C.getint(self.site, "webcam_fps") self.movingimage = self.staticimage = False elif self.vidsource == "screen": self.res = C.get(self.site, "screencap_res").split("x") self.fps = C.getint(self.site, "screencap_fps") - self.origin: List[str] = C.get(self.site, "screencap_origin").split(",") + self.origin: list[str] = C.get(self.site, "screencap_origin").split(",") self.movingimage = self.staticimage = False elif self.vidsource == "file": # streaming video from a file self.res = utils.get_resolution(self.infn, self.probeexe) @@ -128,7 +127,7 @@ def osparam(self, key: str): else: self.key = utils.getstreamkey(C.get(self.site, "key", fallback=None)) - def videoIn(self, quick: bool = False) -> List[str]: + def videoIn(self, quick: bool = False) -> list[str]: """ config video input """ @@ -150,13 +149,13 @@ def videoIn(self, quick: bool = False) -> List[str]: return v - def videoOut(self) -> List[str]: + def videoOut(self) -> list[str]: """ configure video output """ vid_format = "uyvy422" if sys.platform == "darwin" else "yuv420p" - v: List[str] = ["-codec:v", "libx264", "-pix_fmt", vid_format] + v: list[str] = ["-codec:v", "libx264", "-pix_fmt", vid_format] # %% set frames/sec, bitrate and keyframe interval """ DON'T DO THIS. @@ -180,7 +179,7 @@ def videoOut(self) -> List[str]: return v - def audioIn(self, quick: bool = False) -> List[str]: + def audioIn(self, quick: bool = False) -> list[str]: """ -ac 2 doesn't seem to be needed, so it was removed. @@ -198,7 +197,7 @@ def audioIn(self, quick: bool = False) -> List[str]: self.audiochan = f"anullsrc=sample_rate={self.audiofs}:channel_layout=stereo" if self.vidsource == "file": - a: List[str] = [] + a: list[str] = [] elif self.acap == "null": a = ["-f", "lavfi", "-i", self.audiochan] else: @@ -206,7 +205,7 @@ def audioIn(self, quick: bool = False) -> List[str]: return a - def audioOut(self) -> List[str]: + def audioOut(self) -> list[str]: """ select audio codec @@ -246,13 +245,13 @@ def video_bitrate(self): else: self.video_kbps: int = list(BR60.values())[bisect.bisect_left(list(BR60.keys()), x)] - def screengrab(self, quick: bool = False) -> List[str]: + def screengrab(self, quick: bool = False) -> list[str]: """ grab video from desktop. May not work for Wayland desktop. """ - v: List[str] = ["-f", self.vcap] + v: list[str] = ["-f", self.vcap] # FIXME: explict frame rate is problematic for MacOS with screenshare. Just leave it off? # if not quick: @@ -277,7 +276,7 @@ def screengrab(self, quick: bool = False) -> List[str]: return v - def webcam(self, quick: bool = False) -> List[str]: + def webcam(self, quick: bool = False) -> list[str]: """ configure webcam @@ -289,13 +288,13 @@ def webcam(self, quick: bool = False) -> List[str]: if not webcam_chan: webcam_chan = "default" - v: List[str] = ["-f", self.hcam, "-i", webcam_chan] + v: list[str] = ["-f", self.hcam, "-i", webcam_chan] # '-r', str(self.fps), # -r causes bad dropouts return v - def filein(self, quick: bool = False) -> List[str]: + def filein(self, quick: bool = False) -> list[str]: """ used for: @@ -303,7 +302,7 @@ def filein(self, quick: bool = False) -> List[str]: * microphone-only """ - v: List[str] = [] + v: list[str] = [] """ -re is NOT for actual streaming devices (webcam, microphone) @@ -334,7 +333,7 @@ def filein(self, quick: bool = False) -> List[str]: return v - def buffer(self, server: str) -> List[str]: + def buffer(self, server: str) -> list[str]: """configure network buffer. Tradeoff: latency vs. robustness""" # constrain to single thread, default is multi-thread # buf = ['-threads', '1'] diff --git a/src/pylivestream/utils.py b/src/pylivestream/utils.py index 1a9138b..276abf4 100644 --- a/src/pylivestream/utils.py +++ b/src/pylivestream/utils.py @@ -1,15 +1,15 @@ +from __future__ import annotations import logging import subprocess from pathlib import Path import sys -import typing as T import importlib.resources import shutil from .ffmpeg import get_meta -def run(cmd: T.Sequence[str]): +def run(cmd: list[str]): """ FIXME: shell=True for Windows seems necessary to specify devices enclosed by "" quotes """ @@ -37,7 +37,7 @@ def run(cmd: T.Sequence[str]): """ -def check_device(cmd: T.Sequence[str]) -> bool: +def check_device(cmd: list[str]) -> bool: try: run(cmd) ok = True @@ -98,7 +98,7 @@ def meta_caption(meta) -> str: return caption -def get_resolution(fn: Path, exe: str = None) -> T.List[str]: +def get_resolution(fn: Path, exe: str = None) -> list[str]: """ get resolution (widthxheight) of video file http://trac.ffmpeg.org/wiki/FFprobeTips#WidthxHeight