Commit 0c657954 authored by Jackson Garner's avatar Jackson Garner

Add more type stuff

parent 7e6ff734
Pipeline #51077684 failed with stage
in 2 minutes and 39 seconds
......@@ -120,7 +120,8 @@ def loadConf(config: int) -> Tuple[List[Source], Iterator[uri.URI]]:
raise KeyError('No sources found in configuration file.')
with open(Config.sources) as source_file:
sources: List[Source] = get_sources_list(source_file)
sources: List[Source] = get_sources_list(
generator: Iterator[Any] = playable_generator(sources)
a = (sources, generator)
return a
......@@ -15,7 +15,7 @@
import os
from os import environ, fspath, makedirs, path
from typing import Any, Dict
from typing import Dict, Any
import yaml
from import PatternMatchingEventHandler
......@@ -26,20 +26,20 @@ class Singleton(type):
"""Technique from here:"""
_instances: Dict[type, Any] = {}
def __call__(cls, *args, **kwargs):
def __call__(cls, *args, **kwargs) -> Any:
if cls not in cls._instances:
cls._instances[cls] = super(Singleton,
cls).__call__(*args, **kwargs)
return cls._instances[cls]
def __getattr__(cls, name):
def __getattr__(cls, name) -> Any:
"""Overrides dot operator access on the type."""
if hasattr(cls._instances.get(cls), '__classgetattr__'):
return cls._instances[cls].__classgetattr__(name)
return super().__getattribute__(name)
def __getitem__(cls, name):
def __getitem__(cls, name) -> Any:
"""Overrides blacket operator access on the type."""
if hasattr(cls._instances.get(cls), '__getitem__'):
return cls._instances[cls].__getitem__(name)
# This file is part of Visplay.
# Visplay is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
# Visplay is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# GNU General Public License for more details.
# You should have received a copy of the GNU General Public License
# along with Visplay. If not, see <>.
import uri
import yaml
import random
def get_sources_list(source_stream):
from visplay.sources import source_constructors as sc
all_sources = []
sources_yaml = yaml.load(source_stream)
# Handle imports
if 'import' in sources_yaml and sources_yaml['import']:
for name in sources_yaml['import']:
source = sources_yaml['import'][name]
path = uri.URI(source)
new_source = sc[str(path.scheme)](name, path, is_import=True)
if 'add' in sources_yaml and sources_yaml['add']:
# calls the corresponding source constructor (LocalSource/HTTPSource)
# with the arguments provided, then appends to the list
for source in sources_yaml['add']:
path = uri.URI(source)
new_source = sc[str(path.scheme)](source, path)
if 'addTo' in path.query:
playlist_name = path.query['addTo']
playlist = list(filter(lambda a: type(a) == str, new_source.assets))
if 'shuffle' in path.query:
new_source.assets[playlist_name] = playlist
return all_sources
# Create the necessary namespaces
def sources_to_asset(name, sources):
assets = {}
for source in sources:
for asset in source.assets:
if type(source.assets[asset]) is list:
source_asset = source.assets[asset]
source_asset[:] = [name + ':' + item for item in source_asset]
assets[name + ':' + asset] = source.assets[asset]
return assets
......@@ -16,7 +16,7 @@
import os
import random
from pathlib import Path
from typing import IO, Dict, List, NewType
from typing import Dict, List, cast
import aiohttp
import requests
......@@ -26,17 +26,13 @@ from uri import URI
from import PatternMatchingEventHandler
from watchdog.observers import Observer
from visplay.setup_sources import get_sources_list, sources_to_asset
from typing import Dict, List, cast
from aiohttp import ClientSession
# Every source needs to say whether the files it gets survive after an error,
# how to get an asset file, and how to get a playlist file.
Assets = NewType('Asset', Dict[str, URI])
def get_local_yaml(yaml_path: str) -> Assets:
def get_local_yaml(yaml_path: URI) -> Dict[str, URI]:
if yaml_path is None:
return {}
......@@ -47,12 +43,11 @@ def get_local_yaml(yaml_path: str) -> Assets:
raise Exception(f'{yaml_path} does not exist.')
def get_sources_list(source_stream: IO) -> List['Source']:
def get_sources_list(source_stream: str) -> List['Source']:
from visplay.sources import source_constructors as sc
all_sources = []
sources_yaml = yaml.load(source_stream)
# Handle imports
if 'import' in sources_yaml and sources_yaml['import']:
......@@ -130,7 +125,7 @@ class HTTPSource(Source):
if self.is_import:
with requests.get(uri.base, verify=False) as remote_file:
self.sources = get_sources_list(remote_file.content)
self.sources = get_sources_list(remote_file.text)
self.assets = sources_to_asset(name, self.sources)
self.layout = None
......@@ -210,7 +205,8 @@ class PathSource(Source):
if self.is_import and '.sources' in file_path.suffixes:
with open(file_path) as source_file:
# Recursively discover all sources
self.sources += get_sources_list(source_file)
self.sources += get_sources_list(
# Namespace the assets
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment