Skip to content

Commit

Permalink
modernize anno
Browse files Browse the repository at this point in the history
  • Loading branch information
@scivision
scivision committed Feb 10, 2021
1 parent dc5c3b9 commit 99f7bd6
Show file tree
Hide file tree
Showing 9 changed files with 75 additions and 155 deletions.
76 changes: 0 additions & 76 deletions .github/CODE_OF_CONDUCT.md
View file

This file was deleted.

4 changes: 0 additions & 4 deletions .github/FUNDING.yml
View file

This file was deleted.

2 changes: 1 addition & 1 deletion setup.cfg
View file
Original file line number Diff line number Diff line change
Expand Up @@ -46,7 +46,7 @@ lint =
flake8-bugbear
flake8-builtins
flake8-blind-except
mypy
mypy >= 0.800
captions =
tinytag

Expand Down
18 changes: 9 additions & 9 deletions src/pylivestream/api.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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}")
Expand All @@ -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,
Expand Down Expand Up @@ -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,
Expand All @@ -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)
Expand All @@ -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)
Expand Down
62 changes: 31 additions & 31 deletions src/pylivestream/base.py
View file
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
from __future__ import annotations
from pathlib import Path
import typing
import logging
import os

Expand All @@ -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
Expand Down Expand Up @@ -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]
Expand All @@ -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:
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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]
Expand All @@ -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))
Expand All @@ -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]
Expand All @@ -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))
Expand All @@ -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]
Expand All @@ -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))
Expand All @@ -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]
Expand All @@ -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))
Expand All @@ -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

Expand All @@ -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.
Expand All @@ -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__)

Expand Down
11 changes: 6 additions & 5 deletions src/pylivestream/ffmpeg.py
View file
Original file line number Diff line number Diff line change
@@ -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
Expand All @@ -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 []

Expand All @@ -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 []
Expand Down Expand Up @@ -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 []

Expand Down Expand Up @@ -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

Expand Down
Loading

0 comments on commit 99f7bd6

Please sign in to comment.