Skip to content
Snippets Groups Projects
Commit 29f3f9f6 authored by Angus Lothian's avatar Angus Lothian :dark_sunglasses:
Browse files

Revert "Deleted README.md"

This reverts commit 21124444
parent 726c8e18
No related branches found
No related tags found
No related merge requests found
AccessModifierOffset: -4
AlignAfterOpenBracket: DontAlign
AlignConsecutiveAssignments: false
AlignConsecutiveDeclarations: false
AlignConsecutiveMacros: true
AlignEscapedNewlines: DontAlign
AlignOperands: false
AlignTrailingComments: true
AllowAllArgumentsOnNextLine: true
AllowAllConstructorInitializersOnNextLine: true
AllowAllParametersOfDeclarationOnNextLine: false
AllowShortBlocksOnASingleLine: Empty
AllowShortCaseLabelsOnASingleLine: true
AllowShortFunctionsOnASingleLine: Empty
AllowShortIfStatementsOnASingleLine: Never
AllowShortLambdasOnASingleLine: Inline
AllowShortLoopsOnASingleLine: false
AlwaysBreakAfterReturnType: None
AlwaysBreakBeforeMultilineStrings: true
AlwaysBreakTemplateDeclarations: Yes
BinPackArguments: false
BinPackParameters: true
BreakBeforeBraces: Custom
BraceWrapping:
AfterCaseLabel: false
AfterClass: false
AfterControlStatement: false
AfterEnum: false
AfterFunction: false
AfterNamespace: false
AfterObjCDeclaration: false
AfterStruct: false
AfterUnion: false
AfterExternBlock: false
BeforeCatch: false
BeforeElse: false
IndentBraces: false
SplitEmptyFunction: false
SplitEmptyRecord: false
SplitEmptyNamespace: true
BreakBeforeBinaryOperators: None
BreakBeforeTernaryOperators: false
BreakConstructorInitializers: BeforeComma
BreakInheritanceList: BeforeComma
BreakStringLiterals: true
ColumnLimit: 140
CommentPragmas: ''
CompactNamespaces: false
ConstructorInitializerAllOnOneLineOrOnePerLine: true
ConstructorInitializerIndentWidth: 4
ContinuationIndentWidth: 4
Cpp11BracedListStyle: true
DerivePointerAlignment: false
DisableFormat: false
FixNamespaceComments: true
ForEachMacros:
- Q_FOREACH
- BOOST_FOREACH
- FOREACH
- FOR_EACH
IncludeBlocks: Regroup
IncludeCategories:
- Regex: '^<'
Priority: 2
- Regex: '.*'
Priority: 1
IncludeIsMainRegex: '(_test)?$'
IndentCaseLabels: true
IndentGotoLabels: false
IndentPPDirectives: None
IndentWidth: 4
IndentWrappedFunctionNames: false
KeepEmptyLinesAtTheStartOfBlocks: false
Language: Cpp
MacroBlockBegin: ''
MacroBlockEnd: ''
MaxEmptyLinesToKeep: 1
NamespaceIndentation: None
NamespaceMacros:
- NAMESPACE
PenaltyBreakAssignment: 100
PenaltyBreakBeforeFirstCallParameter: 10
PenaltyBreakComment: 10
PenaltyBreakFirstLessLess: 100
PenaltyBreakString: 10
PenaltyBreakTemplateDeclaration: 10000
PenaltyExcessCharacter: 999999
PenaltyReturnTypeOnItsOwnLine: 10000
PointerAlignment: Left
ReflowComments: false
SortIncludes: true
SortUsingDeclarations: true
SpaceAfterCStyleCast: false
SpaceAfterLogicalNot: false
SpaceAfterTemplateKeyword: true
SpaceBeforeAssignmentOperators: true
SpaceBeforeCpp11BracedList: false
SpaceBeforeCtorInitializerColon: true
SpaceBeforeInheritanceColon: true
SpaceBeforeParens: ControlStatements
SpaceBeforeRangeBasedForLoopColon: true
SpaceInEmptyBlock: false
SpaceInEmptyParentheses: false
SpacesBeforeTrailingComments: 1
SpacesInAngles: false
SpacesInCStyleCastParentheses: false
SpacesInContainerLiterals: false
SpacesInParentheses: false
SpacesInSquareBrackets: false
Standard: Cpp11
StatementMacros:
- Q_UNUSED
TabWidth: 4
TypenameMacros:
- STACK_OF
- LIST
- LIST_ENTRY
UseTab: ForContinuationAndIndentation
\ No newline at end of file
.vs/
.vscode/
build*/
bin*/
logs/
dist/
CMakeLists.txt.user*
*.autosave
*.creator
*.creator.user*
\#*\#
/.emacs.desktop
/.emacs.desktop.lock
*.elc
auto-save-list
tramp
.\#*
*~
.fuse_hudden*
.directory
.Trash-*
.nfs*
Thumbs.db
Thumbs.db:encryptable
ehthumbs.db
ehthumbs_vista.db
$RECYCLE.BIN/
*.stackdump
[Dd]esktop.ini
*.egg-info
__pycache__/
env/
venv/
\ No newline at end of file
image: python:3.6
stages:
- build
- test
before_script:
- apt-get update --yes
- apt-get install --yes build-essential cmake libfmt-dev pybind11-dev
- pip3 install pytest pytest-cov
- pip3 install .
- pip3 show b_asic
PythonBuild:
stage: build
artifacts:
untracked: true
run tests:
stage: test
script:
- apt-get update && apt-get install python3 -y
- bash build.sh
\ No newline at end of file
- pytest test
\ No newline at end of file
cmake_minimum_required(VERSION 3.8)
project(
"B-ASIC"
VERSION 0.0.1
DESCRIPTION "Better ASIC Toolbox for python3"
LANGUAGES C CXX
)
find_package(fmt 5.2.1 REQUIRED)
find_package(pybind11 CONFIG REQUIRED)
set(LIBRARY_NAME "b_asic")
set(TARGET_NAME "_${LIBRARY_NAME}")
if(NOT CMAKE_LIBRARY_OUTPUT_DIRECTORY)
include(GNUInstallDirs)
set(CMAKE_LIBRARY_OUTPUT_DIRECTORY "${CMAKE_INSTALL_LIBDIR}")
endif()
set(CMAKE_ARCHIVE_OUTPUT_DIRECTORY "${CMAKE_LIBRARY_OUTPUT_DIRECTORY}")
set(CMAKE_ARCHIVE_OUTPUT_DIRECTORY_DEBUG "${CMAKE_LIBRARY_OUTPUT_DIRECTORY}")
set(CMAKE_ARCHIVE_OUTPUT_DIRECTORY_RELEASE "${CMAKE_LIBRARY_OUTPUT_DIRECTORY}")
set(CMAKE_LIBRARY_OUTPUT_DIRECTORY_DEBUG "${CMAKE_LIBRARY_OUTPUT_DIRECTORY}")
set(CMAKE_LIBRARY_OUTPUT_DIRECTORY_RELEASE "${CMAKE_LIBRARY_OUTPUT_DIRECTORY}")
set(CMAKE_PDB_OUTPUT_DIRECTORY "${CMAKE_LIBRARY_OUTPUT_DIRECTORY}")
set(CMAKE_PDB_OUTPUT_DIRECTORY_DEBUG "${CMAKE_LIBRARY_OUTPUT_DIRECTORY}")
set(CMAKE_PDB_OUTPUT_DIRECTORY_RELEASE "${CMAKE_LIBRARY_OUTPUT_DIRECTORY}")
set(CMAKE_RUNTIME_OUTPUT_DIRECTORY "${CMAKE_LIBRARY_OUTPUT_DIRECTORY}")
set(CMAKE_RUNTIME_OUTPUT_DIRECTORY_DEBUG "${CMAKE_LIBRARY_OUTPUT_DIRECTORY}")
set(CMAKE_RUNTIME_OUTPUT_DIRECTORY_RELEASE "${CMAKE_LIBRARY_OUTPUT_DIRECTORY}")
pybind11_add_module(
"${TARGET_NAME}"
"${CMAKE_CURRENT_SOURCE_DIR}/src/main.cpp"
)
target_include_directories(
"${TARGET_NAME}"
PRIVATE
"${CMAKE_CURRENT_SOURCE_DIR}/src"
)
target_compile_features(
"${TARGET_NAME}"
PRIVATE
cxx_std_17
)
target_compile_options(
"${TARGET_NAME}"
PRIVATE
$<$<OR:$<CXX_COMPILER_ID:GNU>,$<CXX_COMPILER_ID:Clang>>:
-W -Wall -Wextra -Werror -Wno-psabi
$<$<CONFIG:Debug>:-g>
$<$<NOT:$<CONFIG:Debug>>:-O3>
>
$<$<CXX_COMPILER_ID:MSVC>:
/W3 /WX /permissive- /utf-8
$<$<CONFIG:Debug>:/Od>
$<$<NOT:$<CONFIG:Debug>>:/Ot>
>
)
target_link_libraries(
"${TARGET_NAME}"
PRIVATE
fmt::fmt
)
add_custom_target(
remove_old_python_dir ALL
COMMAND ${CMAKE_COMMAND} -E remove_directory "${CMAKE_LIBRARY_OUTPUT_DIRECTORY}/${LIBRARY_NAME}"
COMMENT "Removing old python directory ${CMAKE_LIBRARY_OUTPUT_DIRECTORY}/${LIBRARY_NAME}"
)
add_custom_target(
copy_python_dir ALL
COMMAND ${CMAKE_COMMAND} -E copy_directory "${CMAKE_CURRENT_LIST_DIR}/${LIBRARY_NAME}" "${CMAKE_LIBRARY_OUTPUT_DIRECTORY}/${LIBRARY_NAME}"
COMMENT "Copying python files to ${CMAKE_LIBRARY_OUTPUT_DIRECTORY}/${LIBRARY_NAME}"
DEPENDS "${TARGET_NAME}" remove_old_python_dir
)
\ No newline at end of file
MIT License
Copyright (c) 2020 TDDD96
Copyright (c) 2020 TDDD96 PUM4
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
......
include README.md
include LICENSE
include CMakeLists.txt
recursive-include src *.cpp *.h
<img src="https://files.slack.com/files-pri/TSHPRJY83-FTTRW9MQ8/b-asic-logo-opaque.png" width="318" height="100">
<br>
<h3>B-ASIC is an ASIC toolbox for Python 3 that simplifies circuit design and optimization.<h3>
\ No newline at end of file
<img src="logo.png" width="278" height="100">
# B-ASIC - Better ASIC Toolbox
B-ASIC is an ASIC toolbox for Python 3 that simplifies circuit design and optimization.
## Development
How to build and debug the library during development.
### Prerequisites
The following packages are required in order to build the library:
* cmake 3.8+
* gcc 7+/clang 7+/msvc 16+
* fmtlib 5.2.1+
* pybind11 2.3.0+
* python 3.6+
* setuptools
* wheel
* pybind11
* numpy
* pyside2/pyqt5
### Using CMake directly
How to build using CMake.
#### Configuring
In `B-ASIC`:
```
mkdir build
cd build
cmake ..
```
#### Building (Debug)
In `B-ASIC/build`:
```
cmake --build .
```
The output gets written to `B-ASIC/build/lib`.
#### Building (Release)
In `B-ASIC/build`:
```
cmake --build . --config Release
```
The output gets written to `B-ASIC/build/lib`.
### Using setuptools to create a package
How to create a package using setuptools that can be installed using pip.
#### Setup (Binary distribution)
In `B-ASIC`:
```
python3 setup.py bdist_wheel
```
The output gets written to `B-ASIC/dist/b_asic-<version>-<python_tag>-<abi_tag>-<platform_tag>.whl`.
#### Setup (Source distribution)
In `B-ASIC`:
```
python3 setup.py sdist
```
The output gets written to `B-ASIC/dist/b-asic-<version>.tar.gz`.
#### Installation (Binary distribution)
In `B-ASIC/dist`:
```
pip install b_asic-<version>-<python_tag>-<abi_tag>-<platform_tag>.whl
```
#### Installation (Source distribution)
In `B-ASIC/dist`:
```
pip install b-asic-<version>.tar.gz
```
### Running tests
How to run the tests using pytest in a virtual environment.
#### Linux/OS X
In `B-ASIC`:
```
python3 -m venv env
source env/bin/activate
pip install .
pytest
```
#### Windows
In `B-ASIC` (as admin):
```
python3 -m venv env
.\env\Scripts\activate.bat
pip install .
pytest
```
#### Test with coverage
```
pytest --cov=b_asic --cov-report html test
```
## Usage
How to build and use the library as a user.
### Installation
```
pip install b_asic
```
### Importing
```
python3
>>> import b_asic as asic
>>> help(asic)
```
## License
B-ASIC is distributed under the MIT license.
See the included LICENSE file for more information.
"""
Better ASIC Toolbox.
TODO: More info.
"""
from b_asic.core_operations import *
from b_asic.graph_component import *
from b_asic.graph_id import *
from b_asic.operation import *
from b_asic.precedence_chart import *
from b_asic.port import *
from b_asic.schema import *
from b_asic.signal_flow_graph import *
from b_asic.signal import *
from b_asic.simulation import *
"""@package docstring
B-ASIC Core Operations Module.
TODO: More info.
"""
from numbers import Number
from typing import Any
from numpy import conjugate, sqrt, abs as np_abs
from b_asic.port import InputPort, OutputPort
from b_asic.graph_id import GraphIDType
from b_asic.operation import AbstractOperation
from b_asic.graph_component import Name, TypeName
class Input(AbstractOperation):
"""Input operation.
TODO: More info.
"""
# TODO: Implement all functions.
@property
def type_name(self) -> TypeName:
return "in"
class Constant(AbstractOperation):
"""Constant value operation.
TODO: More info.
"""
def __init__(self, value: Number = 0, name: Name = ""):
super().__init__(name)
self._output_ports = [OutputPort(0, self)]
self._parameters["value"] = value
def evaluate(self):
return self.param("value")
@property
def type_name(self) -> TypeName:
return "c"
class Addition(AbstractOperation):
"""Binary addition operation.
TODO: More info.
"""
def __init__(self, source1: OutputPort = None, source2: OutputPort = None, name: Name = ""):
super().__init__(name)
self._input_ports = [InputPort(0, self), InputPort(1, self)]
self._output_ports = [OutputPort(0, self)]
if source1 is not None:
self._input_ports[0].connect(source1)
if source2 is not None:
self._input_ports[1].connect(source2)
def evaluate(self, a, b):
return a + b
@property
def type_name(self) -> TypeName:
return "add"
class Subtraction(AbstractOperation):
"""Binary subtraction operation.
TODO: More info.
"""
def __init__(self, source1: OutputPort = None, source2: OutputPort = None, name: Name = ""):
super().__init__(name)
self._input_ports = [InputPort(0, self), InputPort(1, self)]
self._output_ports = [OutputPort(0, self)]
if source1 is not None:
self._input_ports[0].connect(source1)
if source2 is not None:
self._input_ports[1].connect(source2)
def evaluate(self, a, b):
return a - b
@property
def type_name(self) -> TypeName:
return "sub"
class Multiplication(AbstractOperation):
"""Binary multiplication operation.
TODO: More info.
"""
def __init__(self, source1: OutputPort = None, source2: OutputPort = None, name: Name = ""):
super().__init__(name)
self._input_ports = [InputPort(0, self), InputPort(1, self)]
self._output_ports = [OutputPort(0, self)]
if source1 is not None:
self._input_ports[0].connect(source1)
if source2 is not None:
self._input_ports[1].connect(source2)
def evaluate(self, a, b):
return a * b
@property
def type_name(self) -> TypeName:
return "mul"
class Division(AbstractOperation):
"""Binary division operation.
TODO: More info.
"""
def __init__(self, source1: OutputPort = None, source2: OutputPort = None, name: Name = ""):
super().__init__(name)
self._input_ports = [InputPort(0, self), InputPort(1, self)]
self._output_ports = [OutputPort(0, self)]
if source1 is not None:
self._input_ports[0].connect(source1)
if source2 is not None:
self._input_ports[1].connect(source2)
def evaluate(self, a, b):
return a / b
@property
def type_name(self) -> TypeName:
return "div"
class SquareRoot(AbstractOperation):
"""Unary square root operation.
TODO: More info.
"""
def __init__(self, source1: OutputPort = None, name: Name = ""):
super().__init__(name)
self._input_ports = [InputPort(0, self)]
self._output_ports = [OutputPort(0, self)]
if source1 is not None:
self._input_ports[0].connect(source1)
def evaluate(self, a):
return sqrt((complex)(a))
@property
def type_name(self) -> TypeName:
return "sqrt"
class ComplexConjugate(AbstractOperation):
"""Unary complex conjugate operation.
TODO: More info.
"""
def __init__(self, source1: OutputPort = None, name: Name = ""):
super().__init__(name)
self._input_ports = [InputPort(0, self)]
self._output_ports = [OutputPort(0, self)]
if source1 is not None:
self._input_ports[0].connect(source1)
def evaluate(self, a):
return conjugate(a)
@property
def type_name(self) -> TypeName:
return "conj"
class Max(AbstractOperation):
"""Binary max operation.
TODO: More info.
"""
def __init__(self, source1: OutputPort = None, source2: OutputPort = None, name: Name = ""):
super().__init__(name)
self._input_ports = [InputPort(0, self), InputPort(1, self)]
self._output_ports = [OutputPort(0, self)]
if source1 is not None:
self._input_ports[0].connect(source1)
if source2 is not None:
self._input_ports[1].connect(source2)
def evaluate(self, a, b):
assert not isinstance(a, complex) and not isinstance(b, complex), \
("core_operations.Max does not support complex numbers.")
return a if a > b else b
@property
def type_name(self) -> TypeName:
return "max"
class Min(AbstractOperation):
"""Binary min operation.
TODO: More info.
"""
def __init__(self, source1: OutputPort = None, source2: OutputPort = None, name: Name = ""):
super().__init__(name)
self._input_ports = [InputPort(0, self), InputPort(1, self)]
self._output_ports = [OutputPort(0, self)]
if source1 is not None:
self._input_ports[0].connect(source1)
if source2 is not None:
self._input_ports[1].connect(source2)
def evaluate(self, a, b):
assert not isinstance(a, complex) and not isinstance(b, complex), \
("core_operations.Min does not support complex numbers.")
return a if a < b else b
@property
def type_name(self) -> TypeName:
return "min"
class Absolute(AbstractOperation):
"""Unary absolute value operation.
TODO: More info.
"""
def __init__(self, source1: OutputPort = None, name: Name = ""):
super().__init__(name)
self._input_ports = [InputPort(0, self)]
self._output_ports = [OutputPort(0, self)]
if source1 is not None:
self._input_ports[0].connect(source1)
def evaluate(self, a):
return np_abs(a)
@property
def type_name(self) -> TypeName:
return "abs"
class ConstantMultiplication(AbstractOperation):
"""Unary constant multiplication operation.
TODO: More info.
"""
def __init__(self, coefficient: Number, source1: OutputPort = None, name: Name = ""):
super().__init__(name)
self._input_ports = [InputPort(0, self)]
self._output_ports = [OutputPort(0, self)]
self._parameters["coefficient"] = coefficient
if source1 is not None:
self._input_ports[0].connect(source1)
def evaluate(self, a):
return a * self.param("coefficient")
@property
def type_name(self) -> TypeName:
return "cmul"
class ConstantAddition(AbstractOperation):
"""Unary constant addition operation.
TODO: More info.
"""
def __init__(self, coefficient: Number, source1: OutputPort = None, name: Name = ""):
super().__init__(name)
self._input_ports = [InputPort(0, self)]
self._output_ports = [OutputPort(0, self)]
self._parameters["coefficient"] = coefficient
if source1 is not None:
self._input_ports[0].connect(source1)
def evaluate(self, a):
return a + self.param("coefficient")
@property
def type_name(self) -> TypeName:
return "cadd"
class ConstantSubtraction(AbstractOperation):
"""Unary constant subtraction operation.
TODO: More info.
"""
def __init__(self, coefficient: Number, source1: OutputPort = None, name: Name = ""):
super().__init__(name)
self._input_ports = [InputPort(0, self)]
self._output_ports = [OutputPort(0, self)]
self._parameters["coefficient"] = coefficient
if source1 is not None:
self._input_ports[0].connect(source1)
def evaluate(self, a):
return a - self.param("coefficient")
@property
def type_name(self) -> TypeName:
return "csub"
class ConstantDivision(AbstractOperation):
"""Unary constant division operation.
TODO: More info.
"""
def __init__(self, coefficient: Number, source1: OutputPort = None, name: Name = ""):
super().__init__(name)
self._input_ports = [InputPort(0, self)]
self._output_ports = [OutputPort(0, self)]
self._parameters["coefficient"] = coefficient
if source1 is not None:
self._input_ports[0].connect(source1)
def evaluate(self, a):
return a / self.param("coefficient")
@property
def type_name(self) -> TypeName:
return "cdiv"
"""@package docstring
B-ASIC Operation Module.
TODO: More info.
"""
from abc import ABC, abstractmethod
from typing import NewType
Name = NewType("Name", str)
TypeName = NewType("TypeName", str)
class GraphComponent(ABC):
"""Graph component interface.
TODO: More info.
"""
@property
@abstractmethod
def type_name(self) -> TypeName:
"""Return the type name of the graph component"""
raise NotImplementedError
@property
@abstractmethod
def name(self) -> Name:
"""Return the name of the graph component."""
raise NotImplementedError
@name.setter
@abstractmethod
def name(self, name: Name) -> None:
"""Set the name of the graph component to the entered name."""
raise NotImplementedError
class AbstractGraphComponent(GraphComponent):
"""Abstract Graph Component class which is a component of a signal flow graph.
TODO: More info.
"""
_name: Name
def __init__(self, name: Name = ""):
self._name = name
@property
def name(self) -> Name:
return self._name
@name.setter
def name(self, name: Name) -> None:
self._name = name
"""@package docstring
B-ASIC Graph ID module for handling IDs of different objects in a graph.
TODO: More info
"""
from collections import defaultdict
from typing import NewType, DefaultDict
GraphID = NewType("GraphID", str)
GraphIDType = NewType("GraphIDType", str)
GraphIDNumber = NewType("GraphIDNumber", int)
class GraphIDGenerator:
"""A class that generates Graph IDs for objects."""
_next_id_number: DefaultDict[GraphIDType, GraphIDNumber]
def __init__(self):
self._next_id_number = defaultdict(lambda: 1) # Initalises every key element to 1
def get_next_id(self, graph_id_type: GraphIDType) -> GraphID:
"""Return the next graph id for a certain graph id type."""
graph_id = graph_id_type + str(self._next_id_number[graph_id_type])
self._next_id_number[graph_id_type] += 1 # Increase the current id number
return graph_id
"""@package docstring
B-ASIC Operation Module.
TODO: More info.
"""
from abc import abstractmethod
from numbers import Number
from typing import List, Dict, Optional, Any, Set, TYPE_CHECKING
from collections import deque
from b_asic.graph_component import GraphComponent, AbstractGraphComponent, Name
from b_asic.simulation import SimulationState, OperationState
from b_asic.signal import Signal
if TYPE_CHECKING:
from b_asic.port import InputPort, OutputPort
class Operation(GraphComponent):
"""Operation interface.
TODO: More info.
"""
@abstractmethod
def inputs(self) -> "List[InputPort]":
"""Get a list of all input ports."""
raise NotImplementedError
@abstractmethod
def outputs(self) -> "List[OutputPort]":
"""Get a list of all output ports."""
raise NotImplementedError
@abstractmethod
def input_count(self) -> int:
"""Get the number of input ports."""
raise NotImplementedError
@abstractmethod
def output_count(self) -> int:
"""Get the number of output ports."""
raise NotImplementedError
@abstractmethod
def input(self, i: int) -> "InputPort":
"""Get the input port at index i."""
raise NotImplementedError
@abstractmethod
def output(self, i: int) -> "OutputPort":
"""Get the output port at index i."""
raise NotImplementedError
@abstractmethod
def params(self) -> Dict[str, Optional[Any]]:
"""Get a dictionary of all parameter values."""
raise NotImplementedError
@abstractmethod
def param(self, name: str) -> Optional[Any]:
"""Get the value of a parameter.
Returns None if the parameter is not defined.
"""
raise NotImplementedError
@abstractmethod
def set_param(self, name: str, value: Any) -> None:
"""Set the value of a parameter.
The parameter must be defined.
"""
raise NotImplementedError
@abstractmethod
def evaluate_outputs(self, state: "SimulationState") -> List[Number]:
"""Simulate the circuit until its iteration count matches that of the simulation state,
then return the resulting output vector.
"""
raise NotImplementedError
@abstractmethod
def split(self) -> "List[Operation]":
"""Split the operation into multiple operations.
If splitting is not possible, this may return a list containing only the operation itself.
"""
raise NotImplementedError
@property
@abstractmethod
def neighbors(self) -> "List[Operation]":
"""Return all operations that are connected by signals to this operation.
If no neighbors are found, this returns an empty list.
"""
raise NotImplementedError
class AbstractOperation(Operation, AbstractGraphComponent):
"""Generic abstract operation class which most implementations will derive from.
TODO: More info.
"""
_input_ports: List["InputPort"]
_output_ports: List["OutputPort"]
_parameters: Dict[str, Optional[Any]]
def __init__(self, name: Name = ""):
super().__init__(name)
self._input_ports = []
self._output_ports = []
self._parameters = {}
@abstractmethod
def evaluate(self, *inputs) -> Any: # pylint: disable=arguments-differ
"""Evaluate the operation and generate a list of output values given a
list of input values.
"""
raise NotImplementedError
def inputs(self) -> List["InputPort"]:
return self._input_ports.copy()
def outputs(self) -> List["OutputPort"]:
return self._output_ports.copy()
def input_count(self) -> int:
return len(self._input_ports)
def output_count(self) -> int:
return len(self._output_ports)
def input(self, i: int) -> "InputPort":
return self._input_ports[i]
def output(self, i: int) -> "OutputPort":
return self._output_ports[i]
def params(self) -> Dict[str, Optional[Any]]:
return self._parameters.copy()
def param(self, name: str) -> Optional[Any]:
return self._parameters.get(name)
def set_param(self, name: str, value: Any) -> None:
assert name in self._parameters # TODO: Error message.
self._parameters[name] = value
def evaluate_outputs(self, state: SimulationState) -> List[Number]:
# TODO: Check implementation.
input_count: int = self.input_count()
output_count: int = self.output_count()
assert input_count == len(self._input_ports) # TODO: Error message.
assert output_count == len(self._output_ports) # TODO: Error message.
self_state: OperationState = state.operation_states[self]
while self_state.iteration < state.iteration:
input_values: List[Number] = [0] * input_count
for i in range(input_count):
source: Signal = self._input_ports[i].signal
input_values[i] = source.operation.evaluate_outputs(state)[
source.port_index]
self_state.output_values = self.evaluate(input_values)
# TODO: Error message.
assert len(self_state.output_values) == output_count
self_state.iteration += 1
for i in range(output_count):
for signal in self._output_ports[i].signals():
destination: Signal = signal.destination
destination.evaluate_outputs(state)
return self_state.output_values
def split(self) -> List[Operation]:
# TODO: Check implementation.
results = self.evaluate(self._input_ports)
if all(isinstance(e, Operation) for e in results):
return results
return [self]
@property
def neighbors(self) -> List[Operation]:
neighbors: List[Operation] = []
for port in self._input_ports:
for signal in port.signals:
neighbors.append(signal.source.operation)
for port in self._output_ports:
for signal in port.signals:
neighbors.append(signal.destination.operation)
return neighbors
def traverse(self) -> Operation:
"""Traverse the operation tree and return a generator with start point in the operation."""
return self._breadth_first_search()
def _breadth_first_search(self) -> Operation:
"""Use breadth first search to traverse the operation tree."""
visited: Set[Operation] = {self}
queue = deque([self])
while queue:
operation = queue.popleft()
yield operation
for n_operation in operation.neighbors:
if n_operation not in visited:
visited.add(n_operation)
queue.append(n_operation)
def __add__(self, other):
"""Overloads the addition operator to make it return a new Addition operation
object that is connected to the self and other objects. If other is a number then
returns a ConstantAddition operation object instead.
"""
# Import here to avoid circular imports.
from b_asic.core_operations import Addition, ConstantAddition
if isinstance(other, Operation):
return Addition(self.output(0), other.output(0))
elif isinstance(other, Number):
return ConstantAddition(other, self.output(0))
else:
raise TypeError("Other type is not an Operation or a Number.")
def __sub__(self, other):
"""Overloads the subtraction operator to make it return a new Subtraction operation
object that is connected to the self and other objects. If other is a number then
returns a ConstantSubtraction operation object instead.
"""
# Import here to avoid circular imports.
from b_asic.core_operations import Subtraction, ConstantSubtraction
if isinstance(other, Operation):
return Subtraction(self.output(0), other.output(0))
elif isinstance(other, Number):
return ConstantSubtraction(other, self.output(0))
else:
raise TypeError("Other type is not an Operation or a Number.")
def __mul__(self, other):
"""Overloads the multiplication operator to make it return a new Multiplication operation
object that is connected to the self and other objects. If other is a number then
returns a ConstantMultiplication operation object instead.
"""
# Import here to avoid circular imports.
from b_asic.core_operations import Multiplication, ConstantMultiplication
if isinstance(other, Operation):
return Multiplication(self.output(0), other.output(0))
elif isinstance(other, Number):
return ConstantMultiplication(other, self.output(0))
else:
raise TypeError("Other type is not an Operation or a Number.")
def __truediv__(self, other):
"""Overloads the division operator to make it return a new Division operation
object that is connected to the self and other objects. If other is a number then
returns a ConstantDivision operation object instead.
"""
# Import here to avoid circular imports.
from b_asic.core_operations import Division, ConstantDivision
if isinstance(other, Operation):
return Division(self.output(0), other.output(0))
elif isinstance(other, Number):
return ConstantDivision(other, self.output(0))
else:
raise TypeError("Other type is not an Operation or a Number.")
"""@package docstring
B-ASIC Port Module.
TODO: More info.
"""
from abc import ABC, abstractmethod
from typing import NewType, Optional, List
from b_asic.operation import Operation
from b_asic.signal import Signal
PortIndex = NewType("PortIndex", int)
class Port(ABC):
"""Port Interface.
TODO: More documentaiton?
"""
@property
@abstractmethod
def operation(self) -> Operation:
"""Return the connected operation."""
raise NotImplementedError
@property
@abstractmethod
def index(self) -> PortIndex:
"""Return the unique PortIndex."""
raise NotImplementedError
@property
@abstractmethod
def signals(self) -> List[Signal]:
"""Return a list of all connected signals."""
raise NotImplementedError
@abstractmethod
def signal(self, i: int = 0) -> Signal:
"""Return the connected signal at index i.
Keyword argumens:
i: integer index of the signal requsted.
"""
raise NotImplementedError
@property
@abstractmethod
def connected_ports(self) -> List["Port"]:
"""Return a list of all connected Ports."""
raise NotImplementedError
@abstractmethod
def signal_count(self) -> int:
"""Return the number of connected signals."""
raise NotImplementedError
@abstractmethod
def connect(self, port: "Port") -> Signal:
"""Create and return a signal that is connected to this port and the entered
port and connect this port to the signal and the entered port to the signal."""
raise NotImplementedError
@abstractmethod
def add_signal(self, signal: Signal) -> None:
"""Connect this port to the entered signal. If the entered signal isn't connected to
this port then connect the entered signal to the port aswell."""
raise NotImplementedError
@abstractmethod
def disconnect(self, port: "Port") -> None:
"""Disconnect the entered port from the port by removing it from the ports signal.
If the entered port is still connected to this ports signal then disconnect the entered
port from the signal aswell."""
raise NotImplementedError
@abstractmethod
def remove_signal(self, signal: Signal) -> None:
"""Remove the signal that was entered from the Ports signals.
If the entered signal still is connected to this port then disconnect the
entered signal from the port aswell.
Keyword arguments:
- signal: Signal to remove.
"""
raise NotImplementedError
@abstractmethod
def clear(self) -> None:
"""Removes all connected signals from the Port."""
raise NotImplementedError
class AbstractPort(Port):
"""Abstract port class.
Handles functionality for port id and saves the connection to the parent operation.
"""
_index: int
_operation: Operation
def __init__(self, index: int, operation: Operation):
self._index = index
self._operation = operation
@property
def operation(self) -> Operation:
return self._operation
@property
def index(self) -> PortIndex:
return self._index
class InputPort(AbstractPort):
"""Input port.
TODO: More info.
"""
_source_signal: Optional[Signal]
def __init__(self, port_id: PortIndex, operation: Operation):
super().__init__(port_id, operation)
self._source_signal = None
@property
def signals(self) -> List[Signal]:
return [] if self._source_signal is None else [self._source_signal]
def signal(self, i: int = 0) -> Signal:
assert 0 <= i < self.signal_count(), "Signal index out of bound."
assert self._source_signal is not None, "No Signal connect to InputPort."
return self._source_signal
@property
def connected_ports(self) -> List[Port]:
return [] if self._source_signal is None or self._source_signal.source is None \
else [self._source_signal.source]
def signal_count(self) -> int:
return 0 if self._source_signal is None else 1
def connect(self, port: "OutputPort") -> Signal:
assert self._source_signal is None, "Connecting new port to already connected input port."
return Signal(port, self) # self._source_signal is set by the signal constructor
def add_signal(self, signal: Signal) -> None:
assert self._source_signal is None, "Connecting new port to already connected input port."
self._source_signal: Signal = signal
if self is not signal.destination:
# Connect this inputport as destination for this signal if it isn't already.
signal.set_destination(self)
def disconnect(self, port: "OutputPort") -> None:
assert self._source_signal.source is port, "The entered port is not connected to this port."
self._source_signal.remove_source()
def remove_signal(self, signal: Signal) -> None:
old_signal: Signal = self._source_signal
self._source_signal = None
if self is old_signal.destination:
# Disconnect the dest of the signal if this inputport currently is the dest
old_signal.remove_destination()
def clear(self) -> None:
self.remove_signal(self._source_signal)
class OutputPort(AbstractPort):
"""Output port.
TODO: More info.
"""
_destination_signals: List[Signal]
def __init__(self, port_id: PortIndex, operation: Operation):
super().__init__(port_id, operation)
self._destination_signals = []
@property
def signals(self) -> List[Signal]:
return self._destination_signals.copy()
def signal(self, i: int = 0) -> Signal:
assert 0 <= i < self.signal_count(), "Signal index out of bounds."
return self._destination_signals[i]
@property
def connected_ports(self) -> List[Port]:
return [signal.destination for signal in self._destination_signals \
if signal.destination is not None]
def signal_count(self) -> int:
return len(self._destination_signals)
def connect(self, port: InputPort) -> Signal:
return Signal(self, port) # Signal is added to self._destination_signals in signal constructor
def add_signal(self, signal: Signal) -> None:
assert signal not in self.signals, \
"Attempting to connect to Signal already connected."
self._destination_signals.append(signal)
if self is not signal.source:
# Connect this outputport to the signal if it isn't already
signal.set_source(self)
def disconnect(self, port: InputPort) -> None:
assert port in self.connected_ports, "Attempting to disconnect port that isn't connected."
for sig in self._destination_signals:
if sig.destination is port:
sig.remove_destination()
break
def remove_signal(self, signal: Signal) -> None:
i: int = self._destination_signals.index(signal)
old_signal: Signal = self._destination_signals[i]
del self._destination_signals[i]
if self is old_signal.source:
old_signal.remove_source()
def clear(self) -> None:
for signal in self._destination_signals:
self.remove_signal(signal)
"""@package docstring
B-ASIC Precedence Chart Module.
TODO: More info.
"""
from b_asic.signal_flow_graph import SFG
class PrecedenceChart:
"""Precedence chart constructed from a signal flow graph.
TODO: More info.
"""
sfg: SFG
# TODO: More members.
def __init__(self, sfg: SFG):
self.sfg = sfg
# TODO: Implement.
# TODO: More stuff.
"""@package docstring
B-ASIC Schema Module.
TODO: More info.
"""
from b_asic.precedence_chart import PrecedenceChart
class Schema:
"""Schema constructed from a precedence chart.
TODO: More info.
"""
pc: PrecedenceChart
# TODO: More members.
def __init__(self, pc: PrecedenceChart):
self.pc = pc
# TODO: Implement.
# TODO: More stuff.
"""@package docstring
B-ASIC Signal Module.
"""
from typing import Optional, TYPE_CHECKING
from b_asic.graph_component import AbstractGraphComponent, TypeName, Name
if TYPE_CHECKING:
from b_asic.port import InputPort, OutputPort
class Signal(AbstractGraphComponent):
"""A connection between two ports."""
_source: "OutputPort"
_destination: "InputPort"
def __init__(self, source: Optional["OutputPort"] = None, \
destination: Optional["InputPort"] = None, name: Name = ""):
super().__init__(name)
self._source = source
self._destination = destination
if source is not None:
self.set_source(source)
if destination is not None:
self.set_destination(destination)
@property
def source(self) -> "OutputPort":
"""Return the source OutputPort of the signal."""
return self._source
@property
def destination(self) -> "InputPort":
"""Return the destination "InputPort" of the signal."""
return self._destination
def set_source(self, src: "OutputPort") -> None:
"""Disconnect the previous source OutputPort of the signal and
connect to the entered source OutputPort. Also connect the entered
source port to the signal if it hasn't already been connected.
Keyword arguments:
- src: OutputPort to connect as source to the signal.
"""
self.remove_source()
self._source = src
if self not in src.signals:
# If the new source isn't connected to this signal then connect it.
src.add_signal(self)
def set_destination(self, dest: "InputPort") -> None:
"""Disconnect the previous destination InputPort of the signal and
connect to the entered destination InputPort. Also connect the entered
destination port to the signal if it hasn't already been connected.
Keywords argments:
- dest: InputPort to connect as destination to the signal.
"""
self.remove_destination()
self._destination = dest
if self not in dest.signals:
# If the new destination isn't connected to tis signal then connect it.
dest.add_signal(self)
@property
def type_name(self) -> TypeName:
return "s"
def remove_source(self) -> None:
"""Disconnect the source OutputPort of the signal. If the source port
still is connected to this signal then also disconnect the source port."""
if self._source is not None:
old_source: "OutputPort" = self._source
self._source = None
if self in old_source.signals:
# If the old destination port still is connected to this signal, then disconnect it.
old_source.remove_signal(self)
def remove_destination(self) -> None:
"""Disconnect the destination InputPort of the signal."""
if self._destination is not None:
old_destination: "InputPort" = self._destination
self._destination = None
if self in old_destination.signals:
# If the old destination port still is connected to this signal, then disconnect it.
old_destination.remove_signal(self)
def is_connected(self) -> bool:
"""Returns true if the signal is connected to both a source and a destination,
else false."""
return self._source is not None and self._destination is not None
"""@package docstring
B-ASIC Signal Flow Graph Module.
TODO: More info.
"""
from typing import List, Dict, Optional, DefaultDict
from collections import defaultdict
from b_asic.operation import Operation
from b_asic.operation import AbstractOperation
from b_asic.signal import Signal
from b_asic.graph_id import GraphIDGenerator, GraphID
from b_asic.graph_component import GraphComponent, Name, TypeName
class SFG(AbstractOperation):
"""Signal flow graph.
TODO: More info.
"""
_graph_components_by_id: Dict[GraphID, GraphComponent]
_graph_components_by_name: DefaultDict[Name, List[GraphComponent]]
_graph_id_generator: GraphIDGenerator
def __init__(self, input_signals: List[Signal] = None, output_signals: List[Signal] = None, \
ops: List[Operation] = None, **kwds):
super().__init__(**kwds)
if input_signals is None:
input_signals = []
if output_signals is None:
output_signals = []
if ops is None:
ops = []
self._graph_components_by_id = dict() # Maps Graph ID to objects
self._graph_components_by_name = defaultdict(list) # Maps Name to objects
self._graph_id_generator = GraphIDGenerator()
for operation in ops:
self._add_graph_component(operation)
for input_signal in input_signals:
self._add_graph_component(input_signal)
# TODO: Construct SFG based on what inputs that were given
# TODO: Traverse the graph between the inputs/outputs and add to self._operations.
# TODO: Connect ports with signals with appropriate IDs.
def evaluate(self, *inputs) -> list:
return [] # TODO: Implement
def _add_graph_component(self, graph_component: GraphComponent) -> GraphID:
"""Add the entered graph component to the SFG's dictionary of graph objects and
return a generated GraphID for it.
Keyword arguments:
graph_component: Graph component to add to the graph.
"""
# Add to name dict
self._graph_components_by_name[graph_component.name].append(graph_component)
# Add to ID dict
graph_id: GraphID = self._graph_id_generator.get_next_id(graph_component.type_name)
self._graph_components_by_id[graph_id] = graph_component
return graph_id
def find_by_id(self, graph_id: GraphID) -> Optional[GraphComponent]:
"""Find a graph object based on the entered Graph ID and return it. If no graph
object with the entered ID was found then return None.
Keyword arguments:
graph_id: Graph ID of the wanted object.
"""
if graph_id in self._graph_components_by_id:
return self._graph_components_by_id[graph_id]
return None
def find_by_name(self, name: Name) -> List[GraphComponent]:
"""Find all graph objects that have the entered name and return them
in a list. If no graph object with the entered name was found then return an
empty list.
Keyword arguments:
name: Name of the wanted object.
"""
return self._graph_components_by_name[name]
@property
def type_name(self) -> TypeName:
return "sfg"
"""@package docstring
B-ASIC Simulation Module.
TODO: More info.
"""
from numbers import Number
from typing import List
class OperationState:
"""Simulation state of an operation.
TODO: More info.
"""
output_values: List[Number]
iteration: int
def __init__(self):
self.output_values = []
self.iteration = 0
class SimulationState:
"""Simulation state.
TODO: More info.
"""
# operation_states: Dict[OperationId, OperationState]
iteration: int
def __init__(self):
self.operation_states = {}
self.iteration = 0
# TODO: More stuff.
#! bin/sh
for n in `find . -name "*.py"`
do
python3 $n
done
\ No newline at end of file
logo.png 0 → 100644
logo.png

69 KiB

0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment