ryg/royalnet
ryg/royalnet
1
Fork 0
mirror of https://github.com/RYGhub/royalnet.git synced 2024-11-29 22:44:19 +00:00
Code Activity

🗑 Remove campaigns module, as it has been replaced by engineer

Browse source
This commit is contained in:
Steffo 2020-12-29 16:36:55 +01:00
parent e2781c3743
commit d5b409489d
9 changed files with 1 additions and 369 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

19
royalnet/campaigns/__init__.py
View file

@ -1,19 +0,0 @@
"""
This module defines multiple classes that can be used to facilitate the implementation of a conversational interface
in chat bots.
"""
from .campaign import *
from .asynccampaign import *
from .challenge import *
from .asyncchallenge import *
__all__ = (
"Campaign",
"AsyncCampaign",
"Challenge",
"TrueChallenge",
"AsyncChallenge",
"TrueAsyncChallenge",
)

99
royalnet/campaigns/asynccampaign.py
View file

@ -1,99 +0,0 @@
from __future__ import annotations
from royalnet.royaltyping import *
import logging
import inspect
import datetime
from .asyncchallenge import AsyncChallenge, TrueAsyncChallenge
from .exc import *
log = logging.getLogger(__name__)
__all__ = (
"AsyncCampaign",
)
class AsyncCampaign:
"""
The AsyncCampaign module allows for branching asyncgenerator-based back-and-forths between the software and the
user.
An :class:`.AsyncCampaign` consists of multiple chained AsyncAdventures, which are AsyncGenerators yielding tuples
with an AsyncChallenge and optional data.
"""
def __init__(self, start: AsyncAdventure, challenge: Optional[AsyncChallenge], *args, **kwargs):
"""
Initialize an :class:`.AsyncCampaign` object.
.. warning:: Do not use this, use the :meth:`.create()` method instead!
:param start: The starting adventure for the :class:`.AsyncCampaign`.
"""
self.adventure: AsyncAdventure = start
self.challenge: AsyncChallenge = challenge or TrueAsyncChallenge()
self.last_update: datetime.datetime = ...
@classmethod
async def create(cls, start: AsyncAdventure, challenge: Optional[AsyncChallenge] = None, *args, **kwargs) -> AsyncCampaign:
"""
Create a new :class:`.AsyncCampaign` object.
:param start: The starting Adventure for the :class:`.AsyncCampaign`.
:param challenge: The AsyncChallenge the campaign should start with.
:return: The created :class:`.AsyncCampaign`.
"""
campaign = cls(start=start, challenge=challenge, *args, **kwargs)
await campaign._asend(None)
return campaign
async def _asend(self, data: Any) -> Any:
try:
return await self.adventure.asend(data)
except RuntimeError:
log.error(f"{self.adventure} is being used unexpectedly by something else!")
raise
async def _athrow(self, typ: Type[BaseException], val: Optional[BaseException], tb: Any) -> Any:
try:
return await self.adventure.athrow(typ, val, tb)
except RuntimeError:
log.error(f"{self.adventure} is being used unexpectedly by something else!")
raise
async def _aclose(self) -> None:
try:
await self.adventure.aclose()
except RuntimeError:
log.error(f"{self.adventure} is being used unexpectedly by something else!")
raise
async def next(self, data: Any = None) -> List:
"""
Try to advance the :class:`.AsyncCampaign` with the passed data.
:param data: The data to pass to the current AsyncAdventure.
:return: Optional additional data returned by the AsyncAdventure.
:raises ChallengeFailedError: if the data passed fails the AsyncChallenge check.
"""
self.last_update = datetime.datetime.now()
if not await self.challenge.filter(data):
raise ChallengeFailedError(f"{data} failed the {self.challenge} challenge")
result = await self._asend(data)
if inspect.isasyncgen(result):
await self._aclose()
self.adventure = result
await self._asend(None)
return await self.next(data)
elif isinstance(result, AsyncChallenge):
self.challenge = result
return []
elif result is None:
return []
elif isinstance(result, tuple) and len(result) > 0:
if isinstance(result[0], AsyncChallenge):
self.challenge, *output = result
return output
elif result[0] is None:
_, *output = result
return output
else:
raise TypeError(f"AsyncAdventure yielded an invalid type: {result.__class_}")

24
royalnet/campaigns/asyncchallenge.py
View file

@ -1,24 +0,0 @@
from royalnet.royaltyping import *
import abc
__all__ = (
"AsyncChallenge",
"TrueAsyncChallenge",
)
class AsyncChallenge(metaclass=abc.ABCMeta):
"""A filter for inputs passed to an :class:`.AsyncCampaign`."""
@abc.abstractmethod
async def filter(self, data: Any) -> bool:
"""Decide if the data should be skipped or not."""
raise NotImplementedError()
class TrueAsyncChallenge(AsyncChallenge):
"""An :class:`.AsyncChallenge` which always returns :data:`True`."""
async def filter(self, data: Any) -> bool:
"""Decide if the data should be skipped or not."""
return True

77
royalnet/campaigns/campaign.py
View file

@ -1,77 +0,0 @@
from __future__ import annotations
from royalnet.royaltyping import *
import logging
import inspect
import datetime
from .challenge import Challenge, TrueChallenge
from .exc import *
log = logging.getLogger(__name__)
__all__ = (
"Campaign",
)
class Campaign:
"""
The Campaign module allows for branching generator-based back-and-forths between the software and the
user.
A :class:`.Campaign` consists of multiple chained Adventures, which are Generators yielding tuples with a
:class:`.Campaign` and optional data.
"""
def __init__(self, start: Adventure, challenge: Optional[Challenge] = None, *args, **kwargs):
"""
Initialize a :class:`.Campaign` object.
.. warning:: Do not use this, use the :meth:`.create` method instead!
"""
self.adventure: Adventure = start
self.challenge: Challenge = challenge or TrueChallenge()
self.last_update: datetime.datetime = ...
@classmethod
def create(cls, start: Adventure, challenge: Optional[Challenge] = None, *args, **kwargs) -> Campaign:
"""
Create a new :class:`.Campaign` object.
:param start: The starting Adventure for the :class:`.Campaign`.
:param challenge: The Challenge the :class:`.Campaign` should start with.
:return: The created :class:`.Campaign`.
"""
campaign = cls(start=start, challenge=challenge, *args, **kwargs)
campaign.adventure.send(None)
return campaign
def next(self, data: Any = None) -> List:
"""
Try to advance the :class:`.Campaign` with the passed data.
:param data: The data to pass to the current Adventure.
:return: Optional additional data returned by the Adventure.
:raises .ChallengeFailedError: if the data passed fails the Challenge check.
"""
self.last_update = datetime.datetime.now()
if not self.challenge.filter(data):
raise ChallengeFailedError(f"{data} failed the {self.challenge} challenge")
result = self.adventure.send(data)
if inspect.isgenerator(result):
self.adventure.close()
self.adventure = result
self.adventure.send(None)
return self.next(data)
elif isinstance(result, Challenge):
self.challenge = result
return []
elif result is None:
return []
elif isinstance(result, tuple) and len(result) > 0:
if isinstance(result[0], Challenge):
self.challenge, *output = result
return output
elif result[0] is None:
_, *output = result
return output
else:
raise TypeError(f"Adventure yielded an invalid type: {result.__class_}")

24
royalnet/campaigns/challenge.py
View file

@ -1,24 +0,0 @@
from royalnet.royaltyping import *
import abc
__all__ = (
"Challenge",
"TrueChallenge",
)
class Challenge(metaclass=abc.ABCMeta):
"""A filter for inputs passed to a :class:`.Campaign`."""
@abc.abstractmethod
def filter(self, data: Any) -> bool:
"""Decide if the data should be skipped or not."""
raise NotImplementedError()
class TrueChallenge(Challenge):
"""A :class:`.Challenge` which always returns :data:`True`."""
async def filter(self, data: Any) -> bool:
"""Decide if the data should be skipped or not."""
return True

19
royalnet/campaigns/exc.py
View file

@ -1,19 +0,0 @@
from ..exc import RoyalnetException
class CampaignsError(RoyalnetException):
"""
An error related to the campaigns module.
"""
class ChallengeFailedError(CampaignsError):
"""
The data passed to the Campaign (or its async equivalent) failed the challenge.
"""
__all__ = (
"CampaignsError",
"ChallengeFailedError",
)

55
royalnet/campaigns/tests/test_asynccampaign.py
View file

@ -1,55 +0,0 @@
import pytest
from ..asynccampaign import AsyncCampaign
from ..asyncchallenge import AsyncChallenge
from ..exc import *
@pytest.mark.asyncio
async def test_creation():
async def gen():
yield
await AsyncCampaign.create(start=gen())
@pytest.mark.asyncio
async def test_send_receive():
async def gen():
ping = yield
assert ping == "Ping!"
yield None, "Pong!"
campaign = await AsyncCampaign.create(start=gen())
pong, = await campaign.next("Ping!")
assert pong == "Pong!"
class FalseChallenge(AsyncChallenge):
async def filter(self, data) -> bool:
return False
@pytest.mark.asyncio
async def test_failing_check():
async def gen():
yield
campaign = await AsyncCampaign.create(start=gen(), challenge=FalseChallenge())
with pytest.raises(ChallengeFailedError):
await campaign.next()
@pytest.mark.asyncio
async def test_switching():
async def gen_1():
yield
yield gen_2()
async def gen_2():
yield
yield None, "Second message!"
yield None
campaign = await AsyncCampaign.create(start=gen_1())
data, = await campaign.next()
assert data == "Second message!"

51
royalnet/campaigns/tests/test_campaign.py
View file

@ -1,51 +0,0 @@
import pytest
from ..campaign import Campaign
from ..challenge import Challenge, TrueChallenge
from ..exc import *
def test_creation():
def gen():
yield
Campaign.create(start=gen())
def test_send_receive():
def gen():
ping = yield
assert ping == "Ping!"
yield None, "Pong!"
campaign = Campaign.create(start=gen())
pong, = campaign.next("Ping!")
assert pong == "Pong!"
class FalseChallenge(Challenge):
def filter(self, data) -> bool:
return False
def test_failing_check():
def gen():
yield
campaign = Campaign.create(start=gen(), challenge=FalseChallenge())
with pytest.raises(ChallengeFailedError):
campaign.next()
def test_switching():
def gen_1():
yield
yield gen_2()
def gen_2():
yield
yield None, "Second message!"
yield None
campaign = Campaign.create(start=gen_1())
data, = campaign.next()
assert data == "Second message!"

2
royalnet/engineer/bullet.py
View file

@ -1,5 +1,5 @@
"""
Bullets are parts of the data model
"""
from __future__ import annotations