# pyfuncschedule
A package that allows python functions to be repeatedly run according to a given schedule composed of a series of time intervals.
## Usage & Syntax
Function schedules are written as config files with a custom format. There is not as yet a standardised file extention for such configuration files, we might use `.fsch`.
The simplest func schedule has the following syntax:
```
ACTION(ARG1, ARG2, ...) @ [I1, I2, ...]:N
```
White spaces (including newlines and tabs) are ignored (unless part of `str` literals).
Comments work similar to python and use `#` to being.
### Example 1 - Simple Schedule
```
# this is a comment
foo(1) @ [1,2,3]:1
```
Running this schedule will execute the function `foo` with arguments `1` at `t=1,3,6`. Notice that the schedule is specified in relative time in __seconds__ since the previous function execution (`int` or `float`).
### Example 2 - Repeating Schedule Blocks
```
foo() @ [1,2]:2
```
`foo` will be executed at `t=1,3,4,6`. The `:N` value specifies the number of times to repeat the given schedule block.
### Example 3 - Repeating Forever
```
foo() @ [1]:*
```
`foo` will be executed at `t=1,2,3,4,...` and continue until the schedule runner is terminated.
### Example 4 - Nesting Schedules
It is possible to nest schedules to acheive more complex timings.
```
foo() @ [[1,2]:2]:2
```
We can unpack this schedule as follows:
```
[1,2]:2 -> [1,2,1,2] -> t=[1,3,4,6]
[[1,2]:2]:2 -> [1,2,1,2,1,2,1,2] -> t=[1,3,4,6,7,9,10,12]
```
Nested schedules are unpacked _lazily_.
### Example 5 - Function Calls
Using static timings for schedules is useful in many scenarios, but in some cases we may like to determine these timings during execution. This can be done by registering functions to our schedule (see section TODO).
The syntax for this is the same as a function call in python.
```
foo() @ [uniform(0,1)]:*
```
This will execute `foo()` at some time sampled uniformly in the interval `[0,1]`. The `uniform` function is called repeatedly to produce new random intervals. The function is called to produce its value at the moment of the last execution of `foo`. If the function call is the first element in the schedule, it will be called at time `0` (`foo` is typically not executed at this time).
### Example 6 - Function Data
Many of the Python built-in data types: 'int','float','bool','str','list','dict' can be used as arguments to function calls. It is also possible to nest function calls, making it possible to pass around custom data types.
```
foo(["a","b"]) @ [bar(baz(1), baz(2))]:*
```
## Parsing func schedules
parsing a schedule is easy:
1. create a parser
```python
from pyfuncschedule import ScheduleParser
parser = ScheduleParser()
```
2. register the functions that you want to use:
```python
def foo():
...
def bar():
...
# use register_action for the function you wish to call after each interval.
parser.register_action(foo)
# use register function for functions that are used in the intervals.
parser.register_function(bar)
```
3. define or load your schedules
```python
schedules_str = """foo1()@[bar(),2]:1"""
```
NOTE: as of version 0.1 schedules must be provided as a `str` there is no option for loading directly from a file.
4. parse and resolve
```python
schedules = parser.resolve(parser.parse(schedule_str))
```
This will parse the schedule and resolve any functions that have been registered.
## Running func schedules
The result is a list of schedule objects which act like iterables providing `(interval, func)`. One way to run a given schedule is to iterate over it and to wait in a new thread, for example:
```python
for (interval, action) in schedules[0]:
time.sleep(interval)
action()
```
Instead, we can use `asyncio` and the `schedule.stream()` method, which will run the schedule in an `async` context.
```python
import time
import asyncio
from pyfuncschedule import ScheduleParser
async def main():
start_time = time.time()
schedules_str = """foo()@[1]:3 \n bar()@[2]:2"""
parser = ScheduleParser()
def foo():
return f"foo: {time.time() - start_time}"
def bar():
return f"bar: {time.time() - start_time}"
parser.register_action(foo)
parser.register_action(bar)
schedules = parser.resolve(parser.parse(schedules_str))
async def task(schedule):
async for x in schedule.stream():
print(x)
tasks = []
for schedule in schedules:
tasks.append(asyncio.create_task(task(schedule)))
await asyncio.gather(*tasks)
asyncio.run(main())
```
The above creates an `asyncio` task for each schedule, this gives more control over how each schedule will be run. However if we have many schedules we might instead want all schedules to be "merged" into one. This can be acheived by using the `parser.stream(schedules)` method, see below:
```python
import time
import asyncio
from pyfuncschedule import ScheduleParser
async def main():
start_time = time.time()
schedules_str = """foo()@[1]:3 \n bar()@[2]:2"""
parser = ScheduleParser()
def foo():
return f"foo: {time.time() - start_time}"
def bar():
return f"bar: {time.time() - start_time}"
parser.register_action(foo)
parser.register_action(bar)
schedules = parser.resolve(parser.parse(schedules_str))
# safe async iter context
async with parser.stream(schedules) as stream:
# iterate asynchronously
async for x in stream:
print(x)
asyncio.run(main())
```
## Contributing
If you discover a bug or feel something is missing from this package please create an issue and feel free to contribute!
Raw data
{
"_id": null,
"home_page": "https://github.com/BenedictWilkins/pyfuncschedule",
"name": "pyfuncschedule",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.10",
"maintainer_email": null,
"keywords": null,
"author": "Benedict Wilkins",
"author_email": "benrjw@gmail.com",
"download_url": "https://files.pythonhosted.org/packages/a8/8e/a73a19e4156fca7700b23a618853bde24f422c2b2aabcc326b2f3100f925/pyfuncschedule-0.1.tar.gz",
"platform": null,
"description": "# pyfuncschedule\n\nA package that allows python functions to be repeatedly run according to a given schedule composed of a series of time intervals.\n\n## Usage & Syntax\n\nFunction schedules are written as config files with a custom format. There is not as yet a standardised file extention for such configuration files, we might use `.fsch`.\n\nThe simplest func schedule has the following syntax: \n```\nACTION(ARG1, ARG2, ...) @ [I1, I2, ...]:N\n```\nWhite spaces (including newlines and tabs) are ignored (unless part of `str` literals).\nComments work similar to python and use `#` to being.\n\n### Example 1 - Simple Schedule\n```\n# this is a comment\nfoo(1) @ [1,2,3]:1\n```\n\nRunning this schedule will execute the function `foo` with arguments `1` at `t=1,3,6`. Notice that the schedule is specified in relative time in __seconds__ since the previous function execution (`int` or `float`).\n\n### Example 2 - Repeating Schedule Blocks\n```\nfoo() @ [1,2]:2\n```\n\n`foo` will be executed at `t=1,3,4,6`. The `:N` value specifies the number of times to repeat the given schedule block.\n\n### Example 3 - Repeating Forever\n```\nfoo() @ [1]:*\n```\n\n`foo` will be executed at `t=1,2,3,4,...` and continue until the schedule runner is terminated.\n\n### Example 4 - Nesting Schedules\n\nIt is possible to nest schedules to acheive more complex timings.\n```\nfoo() @ [[1,2]:2]:2\n```\nWe can unpack this schedule as follows:\n```\n[1,2]:2 -> [1,2,1,2] -> t=[1,3,4,6]\n[[1,2]:2]:2 -> [1,2,1,2,1,2,1,2] -> t=[1,3,4,6,7,9,10,12]\n```\nNested schedules are unpacked _lazily_.\n\n### Example 5 - Function Calls\n\nUsing static timings for schedules is useful in many scenarios, but in some cases we may like to determine these timings during execution. This can be done by registering functions to our schedule (see section TODO).\nThe syntax for this is the same as a function call in python.\n\n```\nfoo() @ [uniform(0,1)]:*\n```\n\nThis will execute `foo()` at some time sampled uniformly in the interval `[0,1]`. The `uniform` function is called repeatedly to produce new random intervals. The function is called to produce its value at the moment of the last execution of `foo`. If the function call is the first element in the schedule, it will be called at time `0` (`foo` is typically not executed at this time).\n\n### Example 6 - Function Data\n\nMany of the Python built-in data types: 'int','float','bool','str','list','dict' can be used as arguments to function calls. It is also possible to nest function calls, making it possible to pass around custom data types. \n```\nfoo([\"a\",\"b\"]) @ [bar(baz(1), baz(2))]:*\n```\n\n## Parsing func schedules\n\nparsing a schedule is easy:\n\n1. create a parser\n```python\nfrom pyfuncschedule import ScheduleParser\nparser = ScheduleParser()\n```\n\n2. register the functions that you want to use:\n```python\n\ndef foo():\n ...\ndef bar():\n ...\n# use register_action for the function you wish to call after each interval.\nparser.register_action(foo)\n# use register function for functions that are used in the intervals.\nparser.register_function(bar)\n```\n3. define or load your schedules\n```python\nschedules_str = \"\"\"foo1()@[bar(),2]:1\"\"\"\n```\nNOTE: as of version 0.1 schedules must be provided as a `str` there is no option for loading directly from a file.\n\n4. parse and resolve\n```python\nschedules = parser.resolve(parser.parse(schedule_str))\n```\nThis will parse the schedule and resolve any functions that have been registered. \n\n## Running func schedules\n\nThe result is a list of schedule objects which act like iterables providing `(interval, func)`. One way to run a given schedule is to iterate over it and to wait in a new thread, for example:\n\n```python\nfor (interval, action) in schedules[0]:\n time.sleep(interval)\n action()\n```\n\nInstead, we can use `asyncio` and the `schedule.stream()` method, which will run the schedule in an `async` context.\n```python\nimport time\nimport asyncio\nfrom pyfuncschedule import ScheduleParser\n\nasync def main():\n start_time = time.time()\n schedules_str = \"\"\"foo()@[1]:3 \\n bar()@[2]:2\"\"\"\n parser = ScheduleParser()\n\n def foo():\n return f\"foo: {time.time() - start_time}\"\n\n def bar():\n return f\"bar: {time.time() - start_time}\"\n\n parser.register_action(foo)\n parser.register_action(bar)\n schedules = parser.resolve(parser.parse(schedules_str))\n\n async def task(schedule):\n async for x in schedule.stream():\n print(x)\n\n tasks = []\n for schedule in schedules:\n tasks.append(asyncio.create_task(task(schedule)))\n await asyncio.gather(*tasks)\n\nasyncio.run(main())\n```\n\nThe above creates an `asyncio` task for each schedule, this gives more control over how each schedule will be run. However if we have many schedules we might instead want all schedules to be \"merged\" into one. This can be acheived by using the `parser.stream(schedules)` method, see below:\n\n```python\nimport time\nimport asyncio\nfrom pyfuncschedule import ScheduleParser\n\nasync def main():\n start_time = time.time()\n schedules_str = \"\"\"foo()@[1]:3 \\n bar()@[2]:2\"\"\"\n parser = ScheduleParser()\n\n def foo():\n return f\"foo: {time.time() - start_time}\"\n\n def bar():\n return f\"bar: {time.time() - start_time}\"\n\n parser.register_action(foo)\n parser.register_action(bar)\n schedules = parser.resolve(parser.parse(schedules_str))\n # safe async iter context\n async with parser.stream(schedules) as stream:\n # iterate asynchronously\n async for x in stream:\n print(x)\n\nasyncio.run(main())\n```\n\n## Contributing\n\nIf you discover a bug or feel something is missing from this package please create an issue and feel free to contribute!\n\n\n\n\n\n",
"bugtrack_url": null,
"license": null,
"summary": "A simple schedule parser/runner with a convienient syntax.",
"version": "0.1",
"project_urls": {
"Homepage": "https://github.com/BenedictWilkins/pyfuncschedule"
},
"split_keywords": [],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "9d64cd3cc410f84495a84839f4dd3c289e7899e57e738967952cb25d958addae",
"md5": "af540de102b69d75cea8e2b737abea7b",
"sha256": "03ada132c8d9bcca19b2b55715699d675231d1845cc9e571d37ef410ff88d91b"
},
"downloads": -1,
"filename": "pyfuncschedule-0.1-py3-none-any.whl",
"has_sig": false,
"md5_digest": "af540de102b69d75cea8e2b737abea7b",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.10",
"size": 20769,
"upload_time": "2024-06-13T10:07:07",
"upload_time_iso_8601": "2024-06-13T10:07:07.744880Z",
"url": "https://files.pythonhosted.org/packages/9d/64/cd3cc410f84495a84839f4dd3c289e7899e57e738967952cb25d958addae/pyfuncschedule-0.1-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "a88ea73a19e4156fca7700b23a618853bde24f422c2b2aabcc326b2f3100f925",
"md5": "493e7a8818332e79ed70b8a6e22fd7c5",
"sha256": "2e4d5dffa90b7d16d8a0826446b6e13c3f553ae31bfdada923b18748eea2d22f"
},
"downloads": -1,
"filename": "pyfuncschedule-0.1.tar.gz",
"has_sig": false,
"md5_digest": "493e7a8818332e79ed70b8a6e22fd7c5",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.10",
"size": 23556,
"upload_time": "2024-06-13T10:07:09",
"upload_time_iso_8601": "2024-06-13T10:07:09.536464Z",
"url": "https://files.pythonhosted.org/packages/a8/8e/a73a19e4156fca7700b23a618853bde24f422c2b2aabcc326b2f3100f925/pyfuncschedule-0.1.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-06-13 10:07:09",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "BenedictWilkins",
"github_project": "pyfuncschedule",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"lcname": "pyfuncschedule"
}
JSON
