This commit is contained in:
AF 2023-10-02 16:58:30 +00:00
parent 54a74ffa32
commit f4a3f9a59a
12 changed files with 48 additions and 42 deletions

View File

@ -1,12 +1,12 @@
from setuptools import setup from setuptools import setup
setup( setup(
name='v6d2ctx', name="v6d2ctx",
version='', version="",
packages=['v6d2ctx', 'v6d2ctx.integration'], packages=["v6d2ctx", "v6d2ctx.integration"],
url='', url="",
license='', license="",
author='PARRRATE T&V', author="PARRRATE T&V",
author_email='', author_email="",
description='' description="",
) )

View File

@ -2,11 +2,11 @@ from typing import Callable, Generic, TypeVar
from .context import * from .context import *
__all__ = ('AtOf',) __all__ = ("AtOf",)
K = TypeVar('K') K = TypeVar("K")
V = TypeVar('V') V = TypeVar("V")
class AtOf(Generic[K, V]): class AtOf(Generic[K, V]):
@ -17,6 +17,7 @@ class AtOf(Generic[K, V]):
def wrap(value: V) -> V: def wrap(value: V) -> V:
bucket[key] = value bucket[key] = value
return value return value
return wrap return wrap
def of(key: K) -> V: def of(key: K) -> V:

View File

@ -3,7 +3,7 @@ from typing import Awaitable, Callable, Optional, Union
import discord import discord
__all__ = ('usertype', 'Context', 'escape', 'command_type', 'Explicit', 'Implicit') __all__ = ("usertype", "Context", "escape", "command_type", "Explicit", "Implicit")
usertype = Union[discord.abc.User, discord.user.BaseUser, discord.Member, discord.User] usertype = Union[discord.abc.User, discord.user.BaseUser, discord.Member, discord.User]
@ -23,7 +23,7 @@ class Context:
async def long(self, s: str): async def long(self, s: str):
resio = StringIO(s) resio = StringIO(s)
res = '' res = ""
for line in resio: for line in resio:
if len(res) + len(line) < 2000: if len(res) + len(line) < 2000:
res += line res += line
@ -34,14 +34,14 @@ class Context:
await self.reply(res) await self.reply(res)
ESCAPED = '`_*\'"\\' ESCAPED = "`_*'\"\\"
def escape(s: str): def escape(s: str):
res = StringIO() res = StringIO()
for c in s: for c in s:
if c in ESCAPED: if c in ESCAPED:
c = '\\' + c c = "\\" + c
res.write(c) res.write(c)
return res.getvalue() return res.getvalue()

View File

@ -5,10 +5,12 @@ import discord
from v6d2ctx.context import * from v6d2ctx.context import *
from v6d2ctx.handle_command import * from v6d2ctx.handle_command import *
__all__ = ('handle_args',) __all__ = ("handle_args",)
async def handle_args(of: Callable[[str], command_type], message: discord.Message, args: list[str], client: discord.Client): async def handle_args(
of: Callable[[str], command_type], message: discord.Message, args: list[str], client: discord.Client
):
match args: match args:
case []: case []:
return return

View File

@ -2,7 +2,7 @@ from typing import Callable
from v6d2ctx.context import * from v6d2ctx.context import *
__all__ = ('handle_command',) __all__ = ("handle_command",)
async def handle_command(of: Callable[[str], command_type], ctx: Context, name: str, args: list[str]) -> None: async def handle_command(of: Callable[[str], command_type], ctx: Context, name: str, args: list[str]) -> None:

View File

@ -6,10 +6,12 @@ import discord
from v6d2ctx.context import * from v6d2ctx.context import *
from v6d2ctx.handle_args import * from v6d2ctx.handle_args import *
__all__ = ('handle_content',) __all__ = ("handle_content",)
async def handle_content(of: Callable[[str], command_type], message: discord.Message, content: str, prefix: str, client: discord.Client): async def handle_content(
of: Callable[[str], command_type], message: discord.Message, content: str, prefix: str, client: discord.Client
):
if message.author.bot: if message.author.bot:
return return
if not content.startswith(prefix): if not content.startswith(prefix):

View File

@ -1,4 +1,4 @@
__all__ = ('Event', 'SendableEvents', 'ReceivableEvents', 'Events', 'Receiver') __all__ = ("Event", "SendableEvents", "ReceivableEvents", "Events", "Receiver")
import asyncio import asyncio
from typing import Callable, Generic, TypeVar from typing import Callable, Generic, TypeVar
@ -13,9 +13,9 @@ class Event:
raise NotImplementedError raise NotImplementedError
T = TypeVar('T', bound=Event) T = TypeVar("T", bound=Event)
T_co = TypeVar('T_co', bound=Event, covariant=True) T_co = TypeVar("T_co", bound=Event, covariant=True)
T_contra = TypeVar('T_contra', bound=Event, contravariant=True) T_contra = TypeVar("T_contra", bound=Event, contravariant=True)
class Receiver(Generic[T_contra]): class Receiver(Generic[T_contra]):

View File

@ -1,8 +1,8 @@
__all__ = ('ResponseType', 'cast_to_response') __all__ = ("ResponseType", "cast_to_response")
from typing import Any, TypeAlias from typing import Any, TypeAlias
ResponseType: TypeAlias = list['ResponseType'] | dict[str, 'ResponseType'] | float | int | bool | str | None ResponseType: TypeAlias = list["ResponseType"] | dict[str, "ResponseType"] | float | int | bool | str | None
def cast_to_response(target: Any) -> ResponseType: def cast_to_response(target: Any) -> ResponseType:

View File

@ -1,4 +1,4 @@
__all__ = ('Targets', 'JsonLike', 'Async') __all__ = ("Targets", "JsonLike", "Async")
import abc import abc
from typing import Any, Callable, Generic, TypeVar from typing import Any, Callable, Generic, TypeVar
@ -9,10 +9,10 @@ from .responsetype import *
def qualname(t: type) -> str: def qualname(t: type) -> str:
return f'{t.__module__}.{t.__qualname__}' return f"{t.__module__}.{t.__qualname__}"
T = TypeVar('T') T = TypeVar("T")
class Flagful(Generic[T]): class Flagful(Generic[T]):
@ -29,13 +29,13 @@ class Targets:
def register_target(self, targetname: str, target: Any, methodname: str, /, *flags: object) -> None: def register_target(self, targetname: str, target: Any, methodname: str, /, *flags: object) -> None:
self.targets[targetname] = Flagful((target, methodname), set(flags)) self.targets[targetname] = Flagful((target, methodname), set(flags))
print(f'registered target: {targetname}') print(f"registered target: {targetname}")
def register_type(self, target: type, methodname: str, /, *flags: object) -> None: def register_type(self, target: type, methodname: str, /, *flags: object) -> None:
self.register_target(f'{qualname(target)}.{methodname}', target, methodname, *flags) self.register_target(f"{qualname(target)}.{methodname}", target, methodname, *flags)
def register_instance(self, target: object, methodname: str, /, *flags: object) -> None: def register_instance(self, target: object, methodname: str, /, *flags: object) -> None:
self.register_target(f'{qualname(target.__class__)}().{methodname}', target, methodname, *flags) self.register_target(f"{qualname(target.__class__)}().{methodname}", target, methodname, *flags)
def register_instrumentation( def register_instrumentation(
self, self,
@ -45,7 +45,7 @@ class Targets:
*flags: object, *flags: object,
) -> None: ) -> None:
self.instrumentations[instrumentationname] = Flagful(instrumentation_factory, set(flags)) self.instrumentations[instrumentationname] = Flagful(instrumentation_factory, set(flags))
print(f'registered instrumentation: {instrumentationname}') print(f"registered instrumentation: {instrumentationname}")
def get_factory( def get_factory(
self, self,
@ -54,16 +54,14 @@ class Targets:
methodname: str, methodname: str,
instrumentationname: str, instrumentationname: str,
instrumentation_factory: Callable[[Any, str], Instrumentation], instrumentation_factory: Callable[[Any, str], Instrumentation],
/ /,
) -> Callable[[], Instrumentation]: ) -> Callable[[], Instrumentation]:
if (targetname, instrumentationname) not in self.factories: if (targetname, instrumentationname) not in self.factories:
flags_required = self.instrumentations[instrumentationname].flags flags_required = self.instrumentations[instrumentationname].flags
flags_present = self.targets[targetname].flags flags_present = self.targets[targetname].flags
if not flags_required.issubset(flags_present): if not flags_required.issubset(flags_present):
raise KeyError('target lacks flags required by instrumentation') raise KeyError("target lacks flags required by instrumentation")
self.factories[targetname, instrumentationname] = ( self.factories[targetname, instrumentationname] = lambda: instrumentation_factory(target, methodname)
lambda: instrumentation_factory(target, methodname)
)
return self.factories[targetname, instrumentationname] return self.factories[targetname, instrumentationname]

View File

@ -3,7 +3,10 @@ from typing import Hashable
from v6d2ctx.context import * from v6d2ctx.context import *
__all__ = ('lock_for', 'Locks',) __all__ = (
"lock_for",
"Locks",
)
class Locks: class Locks:

View File

@ -5,7 +5,7 @@ import time
from rainbowadn.instrument import Instrumentation from rainbowadn.instrument import Instrumentation
__all__ = ('ALog', 'SLog', 'FrameTrace', 'ABlockMonitor') __all__ = ("ALog", "SLog", "FrameTrace", "ABlockMonitor")
class ALog(Instrumentation): class ALog(Instrumentation):
@ -70,7 +70,7 @@ class ABlockMonitor:
delay = spent - delta delay = spent - delta
if delay > self.threshold: if delay > self.threshold:
self.threshold = delay self.threshold = delay
print(f'block monitor reached new peak delay {delay:.4f}') print(f"block monitor reached new peak delay {delay:.4f}")
interval = self.interval interval = self.interval
if interval > 0: if interval > 0:
await asyncio.sleep(interval) await asyncio.sleep(interval)

View File

@ -3,7 +3,7 @@ import signal
import discord import discord
__all__ = ('serve',) __all__ = ("serve",)
def serve(main, client: discord.Client, loop: asyncio.AbstractEventLoop): def serve(main, client: discord.Client, loop: asyncio.AbstractEventLoop):